如何使用java从apache kafka开始使用所有消息答案

【问题标题】：How to consume all messages from begining from apache kafka using java如何使用java从apache kafka开始使用所有消息
【发布时间】：2019-01-02 01:51:26
【问题描述】：

我正在尝试使用 Apache Kafka 中主题开头的所有消息。我可以使用当时正在生成的消息。这是我获取消息的代码。

 public void consume() {
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Arrays.asList(topic));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("\"%s\"\n", record.value());
        }
    }
}

【问题讨论】：

您能给我们一些关于您要实现的用例的想法吗？或者这只是你正在尝试的一个实验？
我需要分析来自 Apache Kafka 的消息，但一开始，我需要主题开头的所有数据。

标签： apache-kafka kafka-consumer-api

【解决方案1】：

除了设置auto.offset.reset=earliest，尝试为属性group.id 设置一个新的/随机值并试一试。另外，如果你对跟踪消费者位置不感兴趣，但总是想从头开始，你也可以设置enable.auto.commit=false以避免污染偏移主题。

希望对你有帮助。

谢谢。

【讨论】：

【解决方案2】：

控制台消费者生成一个随机消费者组 id 以完成此操作并设置 auto.offset.reset=earliest

确保关闭消费者对象以防止在 Zookeeper 中添加大量临时消费者组 id

【讨论】：

【解决方案3】：

订阅主题后，您可以使用seekToBeginning 方法来设置主题分区开头的偏移量。当然，它对每个分区都有效，因为具有不同分区的主题具有不同的开始偏移量（如果发生消息删除）。

【讨论】：