apache-kafka - Simple-Kafka-consumer 消息传递重复

Question

我正在尝试用 Java 实现一个简单的 Producer-->Kafka-->Consumer 应用程序。我能够成功地生产和消费消息，但是当我重新启动消费者时出现问题，其中一些已经消费的消息再次被消费者从 Kafka 获取(不是所有消息，而是最后的一些消息)消费的消息)。

我已经设置了autooffset.reset=largest在我的消费者和我的 autocommit.interval.ms 中属性设置为 1000 毫秒。

这种“重新传递一些已经消费的消息”是一个已知问题，还是我在这里遗漏了任何其他设置？

基本上，有没有办法确保之前消费的消息都没有被消费者接收/消费？

Answer 1

Kafka 使用 Zookeeper 来存储消费者偏移量。由于 Zookeeper 操作非常慢，因此不建议在消费完每条消息后提交偏移量。

可以向消费者添加关闭 Hook ，以便在退出前手动提交主题偏移量。但是，这在某些情况下无济于事(例如 jvm 崩溃或 kill -9 )。为了防止这种情况，我建议实现自定义提交逻辑，该逻辑将在处理每条消息(文件或本地数据库)后在本地提交偏移量，并每 1000 毫秒向 Zookeeper 提交偏移量。在消费者启动时，应该查询这两个位置，并且应该使用最多两个值作为消费偏移量。