Kafka 中手动偏移管理的用例

Question

我正在尝试在 Java 上实现 Kafka 消费者。

假设消费者包含一些可能引发异常的消息处理逻辑。在这种情况下，消费者应该休眠一段时间并重新处理最后一条消息。

我的想法是使用手动偏移管理：偏移不会在失败时提交，因此消费者可能会从旧偏移中读取。

在测试过程中，我发现一条消息实际上只读取了一次，尽管没有提交偏移量。仅在应用程序重新启动时才考虑上次提交的偏移量。

我的问题是：

• 我的做法是否正确？
• 手动偏移管理有哪些用例？

Answer 1

KafkaConsumer 保留最新的偏移量in-memory，因此，如果发生异常（并且您从中恢复）并且您想再次读取消息，则需要使用 @ 987654322@ 在第二次轮询之前。

提交偏移量是“唯一的”在那里，以在客户端关闭或崩溃时保留偏移量（即，可靠地存储偏移量与内存中的偏移量）。在客户端启动时，会获取最新提交的偏移量，然后客户端仅使用它自己的内存中偏移量。

如果您想将偏移提交与其他一些操作“捆绑”在一起（例如，另一个系统中的第二个“提交”必须与已提交的 Kafka 偏移同步），手动偏移管理很有用。