3.5.1 消费者提交分区偏移量

拼搏现实的明天。 2022-10-22 09:11 391阅读 0赞

### 3.5 消费者提交分区偏移量 ###

消费者提交偏移量是为了保存分区的消费进度。因为Kafka保证同一个分区只会分配给消费组中的唯一消费者，所以即使发生再平衡后，分区和消费者的所有权关系发生变化，新消费者也可以接着上一个消费者记录的偏移盘位置继续消费消息。

但是消费者即使记录了分区的偏移量，仍然无法解决消息被重复消费的问题。例如，消费者l每隔10秒提交一次偏移量，在10秒时提交的偏移量是100，下一次提交的日们可点是20秒。在20秒之前，消费者l又消费了30条消息，然后消费者l突然挂掉了。由于偏移量现在仍然停留在10。这个位置，因此新的消费者2也只会从10。这个位置继续消费，从而会重复处理偏移量为100之后的30条消息。

通常消息被重复处理是可以接受的，至少不会出现消息丢失这种不可接受的问题。定时提交偏移量的周期时间越长，消息被重复消费的数据量就越多。客户端可以将这个周期时间设置得更短，来减少重复消费的消息量。当然也不能太短，否则会导致客户端和保存偏移量的存储系统产生大量的网络请求。

在旧版本中每个分区的偏移量都保存到ZK中，每个分区都要和ZK产生一次交互，况且还要周期性地写人，这对ZK来说是个不小的负担。在新版本中把偏移量像普通消息一样写入Kafka集群的内部主题。而且正像消息会源源不断地写到集群一样，记录偏移量也是周期性的。Kafka支持高吞吐量的消息写入，对于偏移量的记录当然也不在话下。下面我们会分析两个版本的提交偏移量过程。

--------------------

#### 3.5.1 消费者提交分区偏移量 ####

消费者提交偏移量是为了保存分区的消费进度。因为Kafka保证同一个分区只会分配给消费组中的唯一消费者，所以即使发生再平衡后，分区和消费者的所有权关系发生变化，新消费者也可以接着上一个消费者记录的偏移盘位置继续消费消息。但是消费者即使记录了分区的偏移量，仍然无法解决消息被重复消费的问题。例如，消费者l每隔10秒提交一次偏移量，在10秒时提交的偏移量是100，下一次提交的日们可点是20秒。在20秒之前，消费者l又消费了30条消息，然后消费者l突然挂掉了。由于偏移量现在仍然停留在10。这个位置，因此新的消费者2也只会从10。这个位置继续消费，从而会重复处理偏移量为100之后的30条消息。通常消息被重复处理是可以接受的，至少不会出现消息丢失这种不可接受的问题。定时提交偏移量的周期时间越长，消息被重复消费的数据量就越多。客户端可以将这个周期时间设置得更短，来减少重复消费的消息量。当然也不能太短，否则会导致客户端和保存偏移量的存储系统产生大量的网络请求。在旧版本中每个分区的偏移量都保存到ZK中，每个分区都要和ZK产生一次交互，况且还要周期性地写人，这对ZK来说是个不小的负担。在新版本中把偏移量像普通消息一样写入Kafka集群的内部主题。而且正像消息会源源不断地写到集群一样，记录偏移量也是周期性的。Kafka支持高吞吐量的消息写入，对于偏移量的记录当然也不在话下。下面我们会分析两个版本的提交偏移量过程。

注意：分区信息主要包含3个变量：队列用来存储拉取到的消息，fetchedOffset用来确定拉取的起始位直，consuf!led.Offset表示消费过的位直，用来记录分区的消费进度。

checkpoi.ntedZkOffsets变量用来决定是否需要将偏移量写入ZK。如果分区分配给消费者，但是消费者并没有消费分区的消息，提交偏移量也没有意义，因为当前的消费进度和IZK中保存的偏移量是一样的。比如消费者已经拉取到最新消息了，在这之后一直没有新消息产生，那么即使定时提交的周期时间到了，因为没有消费新的消息，也不需要提交偏移量。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]  
数据有写人就有读取，从ZK中读取偏移量需要指定要读取哪个分区。注意，虽然是由消费者对分区的偏移量进行读写操作，但是分区对应的ZK节点并没有消费者信息。ZK节点的路径中只有消费者所属的组：／consufllers/\[group\_id\]/offsets/\[topic\]/\[partition\_id］。提交分区偏移量以“消费组”为级别，而不是让每个消费者向己维护分区偏移量。目的是：即使某个消费者挂掉，分区偏移量代表的含义也不会改变，再平衡可以将分区调度给“同一个消费组”的其他消费者。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

消费者提交每个分区的偏移iJ:都需要和ZK通信一次，如果集群中的分区数量成千上万，所有消费者和ZK通信会产生大量的网络请求，这对于本身不是作为存储系统的ZK而言，会造成很大的性能问题。但是，记录每个分区的偏移fiJ:又是必须完成的，那么何不考虑重用现有的Kafka集群来存储偏移量呢？

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/20975932232d40ef9594dc4e630ed0ab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/c3dd7ce0ce144e8d83b9fb1b639414e4.png