Kafka权威指南

66

｜

第

4

章

➊

使用一个虚构的方法来提交数据库事务。大致想法是这样的：在处理完记录之后，将记

录和偏移量插入数据库，然后在即将失去分区所有权之前提交事务，确保成功保存了这

些信息。

➋

使用另一个虚构的方法来从数据库获取偏移量，在分配到新分区的时候，使用

seek()

方法定位到那些记录。

➌

订阅主题之后，开始启动消费者，我们调用一次 poll() 方法，让消费者加入到消费者

群组里，并获取分配到的分区，然后马上调用 seek() 方法定位分区的偏移量。要记住，

seek() 方法只更新我们正在使用的位置，在下一次调用 poll() 时就可以获得正确的消

息。如果 seek() 发生错误（比如偏移量不存在），poll() 就会抛出异常。

➍

另一个虚构的方法，这次要更新的是数据库里用于保存偏移量的表。假设更新记录的速

度非常快，所以每条记录都需要更新一次数据库，但提交的速度比较慢，所以只在每个

批次末尾提交一次。这里可以通过很多种方式进行优化。

通过把偏移量和记录保存到同一个外部系统来实现单次语义可以有很多种方式，不过它们

都需要结合使用

ConsumerRebalanceListener

和

seek()

方法来确保能够及时保存偏移量，

并保证消费者总是能够从正确的位置开始读取消息。

4.9

　如何退出

在之前讨论轮询时就说过，不需要担心消费者会在一个无限循环里轮询消息，我们会告诉

消费者如何优雅地退出循环。

如果确定要退出循环，需要通过另一个线程调用

consumer.wakeup()

方法。如果循环运行

在主线程里