
62
|
第
4
章
现,因为它们只会提交最后一个偏移量,而此时该批次里的消息还没有处理完。
幸运的是,消费者
API
允许在调用 commitSync() 和 commitAsync() 方法时传进去希望提交
的分区和偏移量的
map
。假设你处理了半个批次的消息,最后一个来自主题“
customers
”
分区
3
的消息的偏移量是
5000
,你可以调用 commitSync() 方法来提交它。不过,因为消
费者可能不只读取一个分区,你需要跟踪所有分区的偏移量,所以在这个层面上控制偏移
量的提交会让代码变复杂。
下面是提交特定偏移量的例子:
private Map<TopicPartition, OffsetAndMetadata> currentOffsets =
new HashMap<>();
➊
int count = 0;
...
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
{
System.out.printf("topic = %s, partition = %s, offset = %d,
customer = %s, country = %s\n",
record.topic(), record.partition(), record.offset(),
record.key(), ...