Kafka Consumer挂在java的.hasNext

Question

我在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;
    }

发布消息时,数据会成功读取,但是当它返回到检查it.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
}

Answer 1

解决方案是添加属性

"consumer.timeout.ms"

现在,当达到超时时,抛出ConsumerTimeoutException

Answer 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 +实施例