我在 java 中有一个简单的 Kafka 消费者,代码如下
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
}
最佳答案
解决方案是添加属性
“消费者.timeout.ms”
现在,当达到超时时,将抛出 ConsumerTimeoutException
关于java - Kafka Consumer 卡在 .hasNext in java,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/28449851/