0

我对卡夫卡很陌生。我正在创建两个主题并从两个生产者那里发布这两个主题。我有一个消费者消费来自两个主题的消息。这是因为我想按照优先级进行处理。

我从两个主题中都得到了一个流，但是一旦我开始迭代ConsumerItreator任何流，它就会在那里阻塞。正如它在文档中所写的那样，它将被阻止，直到它收到一条新消息。

有人知道如何从单个 Kafka 消费者中读取两个主题和两个流吗？

    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
                topicCountMap.put(KafkaConstants.HIGH_TEST_TOPIC, new Integer(1));
                topicCountMap.put(KafkaConstants.LOW_TEST_TOPIC, new Integer(1));
                Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumerConnector.createMessageStreams(topicCountMap);
                KafkaStream<byte[], byte[]> highPriorityStream = consumerMap.get(KafkaConstants.HIGH_TEST_TOPIC).get(0);
                ConsumerIterator<byte[], byte[]> highPrioerityIterator = highPriorityStream.iterator();

                while (highPriorityStream.nonEmpty() && highPrioerityIterator.hasNext())
                {
                    byte[] bytes = highPrioerityIterator.next().message();
                    Object obj = null;
                    CLoudDataObject thunderDataObject = null;
                    try
                    {

                        obj = SerializationUtils.deserialize(bytes);
                        if (obj instanceof CLoudDataObject)
                        {
                            thunderDataObject = (CLoudDataObject) obj;
                            System.out.println(thunderDataObject);
                            // TODO Got the Thunder object here, now write code to send it to Thunder service.
                        }

                    }
                    catch (Exception e)
                    {
                    }
                }
4

1 回答 1

0

如果您不想在高优先级消息之前处理低优先级消息，如何设置 consumer.timeout.ms 属性并捕获 ConsumerTimeoutException 以检测高优先级的流到达最后一条可用消息？默认情况下，它设置为 -1 以阻止直到新消息到达。( http://kafka.apache.org/07/configuration.html )

下面解释了一种同时处理具有不同优先级的多个流的方法。

Kafka 需要多线程编程。在您的情况下，两个主题的流需要由流的线程处理。因为每个线程将独立运行以处理消息，所以一个阻塞流（线程）不会影响其他流。

Java 的 ThreadPool 实现可以帮助创建多线程应用程序。您可以在此处找到示例实现：

https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example

关于执行的优先级，您可以调用 Thread.currentThread.setPriority 方法来根据它们服务的 Kafka 主题来获得适当的线程优先级。

于 2015-06-09T23:18:22.967 回答