如何从一开始就使用Kafka Consumer API读取数据？

Question

请告诉我，每次我运行消费者时，如何从一开始就使用Kafka消费者API阅读消息。

Answer 1

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

如果您只是避免保存任何补偿，则消费者将始终在开始时重置。

Answer 2

这样做的一个选择是每次开始时都有一个唯一的组标识，这意味着Kafka会从一开始就向您发送主题中的消息。当您为Kafka消费者设置属性时，请执行以下操作：

properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());

另一个选项是使用consumer.seekToStart（consumer.assignment（）），但这将不起作用，除非Kafka首先通过让消费者调用轮询方法从消费者那里获得心跳。因此，如果您希望从一开始就获得所有记录，请调用轮询（），然后执行asikTo开始（），然后再次调用轮询（）。这是一个小hackey，但这似乎是最可靠的方式做到这一点，因为0.9版本。

// At this point, there is no heartbeat from consumer so seekToBeinning() wont work
// So call poll()
consumer.poll(0);
// Now there is heartbeat and consumer is "alive"
consumer.seekToBeginning(consumer.assignment());
// Now consume
ConsumerRecords<String, String> records = consumer.poll(0);

Answer 3

这适用于0.9. x消费者。基本上，当您创建消费者时，您需要使用属性消费者配置为该消费者分配消费者组id。GROUP_ID_CONFIG。每次启动消费者执行类似这样的操作时，随机生成消费者组idproperties.putGROUP_ID_CONFIG，UUID.随机UUID（）. toString（））；（属性是java.util.属性的实例，您将传递给构造函数new Kafka消费者（属性））。

随机生成客户端意味着新的消费者组在kafka中没有任何关联的偏移量。所以在这之后我们要做的是为这个场景设置一个策略。作为auto的文档。抵消重置属性显示：

如果Kafka中没有初始偏移量，或者服务器上不再存在当前偏移量（例如，因为该数据已被删除），该怎么办：

• 最早：自动重置偏移为最早偏移
• 最新：自动重置偏移为最新偏移
• 无：如果没有找到以前的偏移量或消费者的组，则向消费者抛出异常
• 其他：向消费者抛出异常。

因此，从上面列出的选项中，我们需要选择最早的策略，以便新的消费群体每次都从头开始。

您的java代码如下所示：

properties.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "your_client_id");
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumer = new KafkaConsumer(properties);

你现在唯一需要弄清楚的是，当有多个消费者属于同一个消费者组但被分布时，如何生成一个随机标识并在这些实例之间分布，以便它们都属于同一个消费者组。

希望有帮助！