Kafka consumer消息的拉取及偏移的管理

Posted 2021-01-14 benfly

消费者拉取消息并处理主要有4个步骤：

• 获取消费者所拉取分区的偏移位置OffsetFetchRequest（新的消息是从偏移位置开始的）
• 创建FetchReqeust，生成Map<Node, FetchRequest>，以消费者所拉取消息的节点为key来分组，所消费的TopicPartition的数据为value，并放入到unsent队列
• 调用poll方法实际发送请求给相应的node，如果返回成功，在onSecuss方法中，消息被保存在completedFetches中
• 从completedFetches中取出数据，转换成consumerRecord，清空缓冲区及更新消费偏移位置

偏移管理：更新拉取偏移，updateFetchPositions，发送OffsetFetchRequest请求

消费者在启动后，需要获取其所消费的分区的最后提交的偏移位置。消费者在消费完消息后需要提交消费偏移（committed offset），当发生再平衡（reblance）后，分区（partition）有可能被不同的消费者去拉取消息，那新的消费者需要知道上次是消费到哪个偏移位置的，那么新的消费者就需要发出请求给coordinator，以取得提交偏移（committed offset，前一个消费者最后的提交偏移）并更新本地的拉取偏移（fetch position）。消费者在提交偏移的时候，有2种策略可以选择，自动提交（auto commit）和手动提交（manually commit）

自动提交：

通常情况下，为了提高性能，会使用自动提交方式，自动提交的间隔（auto.commit.interval.ms）默认为5000毫秒，是通过延时队列的任务来实现的，在consumer每次拉取消息消费后，如果延时队列的auto commit task到了提交间隔时间，则提交任务更新committed offset，如果没有到延迟任务的timeout时间，则不执行延迟任务，继续拉取消息，但在实际消费处理消息后，提交偏移前，消费者有可能崩溃，这就导致存在重复消费

 

手动提交：

在某些场景下，为了能更准确的控制消费偏移，以保证消息不会重复消费或者不会丢失，由消费者客户端手动控制是否提交偏移

 

偏移与消费语义

 

 

 

重复消费（至少一次消费语义 at least once）：

如果消费流程是：拉取消息，处理消息，提交消费偏移。每次poll拉取N条消息，并处理消息后，在消费者线程保存commited offset，然后执行延时任务，但有可能因为未到时间间隔，没有执行提交偏移的任务，如果这个时候，消费者崩溃，触发再平衡，导致刚才消费的offset没有被提交，那实际的commited offset要小于真正的消费偏移，当新的消费者从新获取的提交偏移出拉取消息后，就导致重复消费了，如下图所示

 

遗漏消费（至多一次消费语义 at most once）：

如果消费流程是：拉取消息，提交消费偏移，处理消息。当提交消费偏移后，加入consumer崩溃触发再平衡，当新的消费者试图更新偏移信息时，将会导致遗漏消息，如下图所示：

 

消息的拉取及消费

消费者如果在上一次的消息拉取过程中有消息存在，则直接返回，否则从前面更新的拉取偏移位置处重新发送拉取消息的请求。

拉取消息的请求以消费者所消费的TopicPartition所在的节点分组Map<Node, FetchRequest>，然后再通过poll到相应的节点来获取分区消息，一旦成功获取掉消息，将被保存在completedFetches，在返回时转换为按TopicPartition分组的record

Map<TopicPartition, List<ConsumerRecord<K, V>>> drained

另外，会在本地记录所消费的最后一条消息的偏移+1，在下次消费时，进行偏移检查，判断第一条记录的offset必须与这个值相等，否则则忽略

 private int append(Map<TopicPartition, List<ConsumerRecord<K, V>>> drained,
                       PartitionRecords<K, V> partitionRecords,
                       int maxRecords) {
    ......
    List<ConsumerRecord<K, V>> partRecords = partitionRecords.take(maxRecords);
    long nextOffset = partRecords.get(partRecords.size() - 1).offset() + 1; //下一条消费位置
    ......

    subscriptions.position(partitionRecords.partition, nextOffset); //记录消费的此TopicPartition的位置为nextOffset
    return partRecords.size();
    .......
  }