Kafka消费者-接收消息不一致

Question

我可以在命令行上针对Kafka位置安装发送和接收消息。我也可以通过Java代码发送消息。这些消息显示在Kafka命令提示符中。我还有一个Kafka消费者的Java代码。代码昨天收到了消息。但是今天早上没有收到任何消息。代码没有更改。我想知道属性配置是否不太正确。这是我的配置：

制片人：

bootstrap.servers - localhost:9092
group.id - test
key.serializer - StringSerializer.class.getName()
value.serializer - StringSerializer.class.getName()

生产记录设置为

ProducerRecord<String, String>("test", "mykey",  "myvalue")

消费者：

zookeeper.connect - "localhost:2181"
group.id - "test"
zookeeper.session.timeout.ms - 500
zookeeper.sync.time.ms - 250
auto.commit.interval.ms - 1000
key.deserializer - org.apache.kafka.common.serialization.StringDeserializer
value.deserializer - org.apache.kafka.common.serialization.StringDeserializer

对于Java代码：

   Map<String, Integer> topicCount = new HashMap<>();
   topicCount.put("test", 1);

   Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumer
            .createMessageStreams(topicCount);
   List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(topic);

少了什么？

Answer 1

可能会发生很多事情。

首先，消费者的ZooKeeper会话超时非常低，这意味着消费者可能会由于垃圾收集暂停而经历许多“软故障”。当这种情况发生时，消费者群体将重新平衡，从而暂停消费。如果这种情况发生得非常频繁，消费者可能会进入一种从不使用消息的状态，因为消息不断被重新平衡。我建议将ZooKeeper会话超时增加到30秒，看看这是否解决了问题。如果是这样，您可以尝试将其设置得更低。

其次，您能否确认正在为“测试”主题生成新消息？您的消费者只会消费它尚未提交的新消息。该主题可能没有任何新消息。

第三，您是否有同一消费者组中的其他消费者可以处理消息？如果一个消费者遇到频繁的软故障，其他消费者将被分配其分区。

最后，您使用的是最终将被删除的“旧”消费者。如果可能的话，我建议转移到Kafka 0.9中提供的“新”消费者（KafkaConsumer.java）。虽然我不能保证这会解决你的问题。

希望这有帮助。