message-queue之Kafka之如何使用高级消费者在每条消息后提交偏移量

我正在使用Kafka的高级消费者。因为我将 K​​afka 用作我的应用程序的“事务队列”，所以我需要绝对确保我不会错过或重新阅读任何消息。我对此有两个问题:

如何将偏移量提交给 Zookeeper？每条消息成功消费后，我将关闭自动提交和提交偏移量。我似乎无法找到有关如何使用高级使用者执行此操作的实际代码示例。谁能帮我这个？

另一方面，我听说提交到 zookeeper 可能很慢，所以另一种方法可能是在本地跟踪偏移量？这种替代方法是否可取？如果是，您将如何处理？

请您参考如下方法:

来自 http://kafka.apache.org/documentation.html#consumerconfigs 的两个相关设置.

auto.commit.enable 

和

auto.commit.interval.ms 

如果您想设置它以便消费者在每条消息之后提交偏移量，这将很困难，因为唯一的设置是在一个计时器间隔之后，而不是在每条消息之后。您必须对传入消息进行一些速率预测并相应地设置时间。

一般来说，不建议将此间隔保持得太小，因为它会大大增加 zookeeper 的读/写速率，而zookeeper 会变慢，因为它在其仲裁中具有很强的一致性。