如何让kafka消费者从上次消费的偏移量而不是从头开始读取

Question

我是 kafka 的新手，并试图了解是否有办法从上次使用的偏移量读取消息，但不是从头开始。

我正在写一个例子，这样我的意图就不会偏离。

Eg:
1) I produced 5 messages at 7:00 PM and console consumer consumed those.
2) I stopped consumer at 7:10 PM
3) I produced 10 message at 7:20 PM. No consumer had read those messages.
4) Now, i have started console consumer at 7:30 PM, without from-beginning.
5) Now, it Will read the messages produced after it has started. Not the earlier ones, which were produced at 7.20 PM

有没有办法从上次消费的偏移量中获取消息。？

Answer 1

在消费者配置中设置auto.offset.reset=earliest 和一个固定的group.id=something 将在最后提交的偏移处启动消费者。在您的情况下，它应该在 7:20 的第一条消息开始消费。如果您希望它在启动后开始阅读发布的消息，那么auto.offset.reset=latest 将忽略在 7:20 发送的 10 条消息，并在启动后阅读任何进来的消息。

如果您希望它从头开始，您必须在第一个 consumer.poll() 之后调用 seekToBeginning，或者将消费者组 ID 更改为唯一的。

Answer 2

我是 kafka 的新手，并试图了解 是否有办法从上次消费的偏移量中读取消息，但不是从头开始。

是的，可以使用控制台消费者从上次消费的偏移量中读取。您必须在调用 kafka-console-consumer 时添加 consumer.config 标志。

例子：-

[root@sandbox bin]# ./kafka-console-consumer.sh --topic test1 --zookeeper localhost:2181 --consumer.config /home/mrnakumar/consumer.properties

这里的 /home/mrnakumar/consumer.properties 是一个包含 group.id 的文件。 以下是 /home/mrnakumar/consumer.properties 的外观：-

group.id=consoleGroup

如果不使用 consumer.config，则可以从 [ 使用 --from-beginning] 的开头或仅从日志的结尾读取。日志结束是指消费者启动后发布的所有消息。

Answer 3

您应该在largest 上的消费者配置中设置auto.offset.reset 参数，这样它将在最后一次提交偏移后读取所有消息。