kafka客户端API问题

Question

谁能帮我解决以下问题。我正在使用 kafka-clients-0.10.1.1（单节点单代理）

auto.create.topics.enable 的默认值为 true。

1.我正在使用

向主题发送消息

    kafkaProdcuer<String,String> producer> producer...
    producer.send(new ProducerRecord<String, String>("my- topic","message"));
    producer.close();

消费：

    kafkaConsumer<String,String> consumer....
    consumer.subscribe(Arrays.asList("my-topic"));
    ConsumerRecords<String, String> records = consumer.poll(200);

    while(true){
     for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.value());
         }
     }

问题是当我第一次运行消费者时，它没有得到值。而且我必须运行生产者并再次运行消费者以获取值。有时我必须运行生产者 3 次。 为什么会这样？

2.) enable.auto.commit=false

如果 enable.auto.commit 属性为 false，同一消费者能否多次阅读消息？

3.) 考虑我在第一点的消费者代码。我如何打破循环我的意思是消费者如何知道它已读取所有消息然后调用 consumer.close()

Answer 1

1) 你总是在消费者中使用相同的 group.id 吗？你是在消费前生产吗？这可能与消费者群体和偏移管理有关。请参阅this answer about consumer offset behavior。

2) 不确定您的意思是有意还是无意地阅读重复内容。只要该消息由于主题保留策略而未被删除，您始终可以再次阅读相同的消息以寻找该位置。如果你的意思是意外，自动提交设置为 false 只是意味着消费者不会为你提交偏移量，你必须手动调用 commitSync() 或 commitAsync()。在任何情况下，您的消费者仍有可能在提交之前处理消息并崩溃，在这种情况下，当消费者恢复时，它将再次读取那些已处理但未提交的消息。如果您只想要一次语义，则必须执行其他操作，例如使用已处理的消息以原子方式存储偏移量。

3) 正如Lhfcws 提到的，在流中没有像“所有消息”这样的概念。您可以做的一些事情（技巧）是：

• 您可以检查 poll 返回的记录列表是否为空，并且在配置的次数后中断循环并退出。
• 如果消息是有序的（您正在从单个分区读取），您可以发送一种特殊的 END_OF_DATA 消息，当您看到它时，您会关闭消费者。
• 您可以让消费者读取一些消息然后退出，下次它会从上次提交的偏移量继续。