Kafka权威指南

64

｜

第

4

章

consumer.commitSync(currentOffsets);

} finally {

consumer.close();

System.out.println("Closed consumer and we are done");

}

}

➊

首先实现 ConsumerRebalanceListener 接口。

➋

在获得新分区后开始读取消息，不需要做其他事情。

➌

如果发生再均衡，我们要在即将失去分区所有权时提交偏移量。要注意，提交的是最近

处理过的偏移量，而不是批次中还在处理的最后一个偏移量。因为分区有可能在我们还

在处理消息的时候被撤回。我们要提交所有分区的偏移量，而不只是那些即将失去所有

权的分区的偏移量——因为提交的偏移量是已经处理过的，所以不会有什么问题。调用

commitSync()

方法，确保在再均衡发生之前提交偏移量。

➍

把

ConsumerRebalanceListener

对象传给

subscribe()

方法，这是最重要的一步。

4.8

　从特定偏移量处开始处理记录

到目前为止，我们知道了如何使用

poll()

方法从各个分区的最新偏移量处开始处理消息。

不过，有时候我们也需要从特定的偏移量处开始读取消息。

如果你想从分区的起始位置开始读取消息，或者直接跳到分区的末尾开始读取消息，可以使

用 seekToBeginning(Collection<TopicPartition> tp) 和 seekToEnd(Collection<TopicPartition> ...