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());
         }
     }

Answer 1

1）您是否总是在消费者中使用相同的group.id？您是否先生产后消费？这可能与消费者群体和抵销管理有关。请看关于消费者抵销行为的答案。

2）不确定你是有意阅读副本还是无意阅读副本。只要该消息未因主题保留策略而被删除，您就可以再次阅读相同的消息，并找到该位置。如果您的意思是无意中，将auto-commit设置为false只是意味着使用者不会为您提交偏移量，那么您必须手动调用commitSync（）或commitAsync（）来完成此操作。在任何情况下，您的使用者仍有可能处理消息并在提交之前崩溃，在这种情况下，当使用者恢复时，它将再次读取那些已处理但未提交的消息。如果您只想要一次语义化，那么您必须做一些其他的事情，比如用处理过的消息原子地存储偏移量。

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

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