管理 kafka 中的当前偏移量

Question

我在文档中找不到任何关于 kafka 如何为消费者管理当前偏移量的信息。我猜消费者默认情况下会将最后一个偏移量保存在内存中，并在显式调用commitSync 或commitAsync 或根据enable.auto.commit 策略提交。这是正确的还是我错过了什么？如果有人可以指出偏移管理这方面的文档或参考，我们将不胜感激。

提前致谢。

Answer 1

消费者组，是一组消费者，它们被协调用于消费来自主题的消息。现在，您的 Kafka 代理之一充当组协调员，负责协调属于该组的所有消费者。

根据enable.auto.commit 的配置以及您在代码中处理偏移管理的方式，将提交并存储在名为__consumer_offsets 的主题中。

---

如果enable.auto.commit 设置为True，那么消费者的偏移量会在后台定期提交。另一方面，commitSync() 和 commitAsync() 分别是允许手动提交偏移量的阻塞和非阻塞调用。如果您使用commitSync() 或commitAsync() 之一，建议将enable.auto.commit 设置为False。

在极少数情况下，您的策略 enable.auto.commit 设置为 True，但同时您使用 commitSync() 或 commitAsync() 之一，那么在这两种情况下都会提交偏移量：

• 每次拨打commitSync()或commitAsync()
• 每N 毫秒，其中N 是一个可配置参数(auto.commit.interval.ms)

---

当enable.auto.commit 设置为true 时，每auto.commit.interval.ms 时间提交一次最大偏移量。但是，只有在调用 poll() 时才会发生这种情况。在每次投票 (max.poll.interval.ms) 中，enable.auto.commit 都会被选中。每当您 poll() 时，消费者都会检查是否该提交它在上次轮询中返回的偏移量。

更多详情可以参考Confluent's documentation for Offset Management。