1

我编写了一个高级 Kafka 消费者作为 Java 应用程序的一部分。

所以核心代码是这样的：

public void start() {
    ConsumerConnector consumerConnector = conf.getConsumerConnector();
    String topic = conf.getTopic();
    int numOfThereads = conf.getNumOfThreads();

    Map<String, Integer> topicCountMap = ImmutableMap.of(topic, numOfThereads);
    Map<String, List<KafkaStream<Message>>> topicMessageStreams = consumerConnector.createMessageStreams(topicCountMap);
    List<KafkaStream<Message>> streams = topicMessageStreams.get(topic);

    // create 4 threads to consume from each of the partitions
    executor = Executors.newFixedThreadPool(numOfThereads);

    // consume the messages in the threads
    for (final KafkaStream<Message> stream : streams) {
        executor.submit(new ConsumerThread(stream));
    }
}

为了测试我的消费者，我还创建了一个生产者，写信给 kafka，然后启动了我的消费者，它可以工作。由于线程是在循环中执行的，我不确定我是否做对了。我希望我的消费者永远运行并继续使用来自 kafka 的消息。

让它永远运行的正确方法是什么？

4

1 回答 1

1

@forhas 感谢您的确认

基本上从文档中他们迭代消费消息的方式如下

    ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
    while (it.hasNext())
        System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));

它还指出 The interesting part here is the while (it.hasNext()) section. Basically this code reads from Kafka until you stop it ..

所以理想情况下，它应该继续运行，除非我们明确地杀死它，并且一旦产生了新消息，消费者端就可以使用它。

于 2013-11-11T05:32:30.090 回答