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 回答