Kafka消费者挂在java中的.hasNext答案

【问题标题】：Kafka Consumer hanging at .hasNext in javaKafka消费者挂在java中的.hasNext
【发布时间】：2015-04-11 13:24:08
【问题描述】：

我在 java 中有一个简单的 Kafka Consumer，代码如下

    public void run() {
        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
        while (it.hasNext()&& !done){
            try {
                System.out.println("Parsing data");
                byte[] data = it.next().message();
                System.out.println("Found data: "+data);
                values.add(data); // array list
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
        }
        done = true;
    }

当消息发布时，数据被成功读取，但是当它返回检查它时。hasNext()，它保持等待状态，永远不会回来。

是什么阻碍了这一切？

m_stream 是一个KafkaStream，获取方式如下：

Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, new Integer(a_numThreads));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
executor = Executors.newFixedThreadPool(a_numThreads);
for (final KafkaStream stream : streams) {
   // m_stream is one of these streams
}

【问题讨论】：

标签： java multithreading apache-kafka

【解决方案1】：

解决方案是添加属性

“consumer.timeout.ms”

现在当达到超时时，会抛出 ConsumerTimeoutException

【讨论】：

【解决方案2】：

hasNext() 方法被阻塞。

你可以在属性consumer.timeout.ms中更改阻塞的超时时间

请注意，当超时到期时，它会抛出一个TimeoutException。

会阅读这些关于消费者的文档： https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example

【讨论】：