4.4.1 自动提交任务

「爱情、让人受尽委屈。」 2022-11-18 04:48 152阅读 0赞

### 4.4 消费者提交偏移量 ###

消费组发生再平衡时分区会被分配给新的消费者，为了保证新消费者能够从分区的上一次消费位置继续拉取并处理消息，每个消费者需要将分区的消费进度，定时地同步给消费组对应的协调者节点。新API为客户端提供了两种提交偏移盐的方式：异步模式和同步模式。相关代码如下：

![在这里插入图片描述][2021041216283153.png]

另外，如果消费者客户端设置了向动提交(enable.auto.commit=true，默认开启)的选项，会在客户端的轮询操作中调度定时任务，定时任务也属于异步模式提交偏移量的一种运用场景。

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

#### 4.4.1 自动提交任务 ####

如果消费者开启向动提交消费者会通过“消费者的协调者”对象的向动提交任务  
(AutoCommitTask)定时将分区的拉取偏移量(position)保存到服务端，然后更新分区状态的“提交偏移量＂(Committed)。自动提交任务定义的run()方法只会执行一次，在每次任务完成后，也要像发送心跳请求一样创建下一次的延迟任务。

另外，创建第一个延迟任务也非常关键，在心跳任务中也提到了：如果没有创建第一个延迟任务，就永远不会有定时任务产生。只有创建了第一个延迟任务并放入队列，当延迟任务超时后，会从队列弹出这个超时任务并执行；任务执行完毕后会创建一个新的延迟任务放入队列。通过这种方式，队列中只会存在一个延迟任务，并且确保一直有一个延迟任务。心跳任务是在消费者连接上协调者或者消费者加入消费组后，调用心跳任务的重置方法创建第一个延迟任务。向动提交任务也是在消费者加入消费组后，调用向动提交任务的enable()方法创建第一个延迟的自动提交任务。

如图4-36所示，假设定时提交任务的时间间隔为5秒，消费者轮询的时间为1秒。但并不是说消费者发送心跳的时间点是：0秒、5秒、10秒。下一次定时任务的调度时间点，要根据上一次时任务完成时的时间来决定。比如第一次定时任务第8秒完成，下一次定时任务的时间点就是第13秒，具体步骤如下。

(1)开启定时任务，第一个延迟任务的时间＝当前时间＋时间间隔＝0秒＋5秒＝5秒。
    (2)前面几次轮询都不会弹出延迟任务，第五次轮询弹出延迟任务，执行异步的提交偏移业任务。
    (3)异步提交任务在第8秒完成，新延迟任务的时间＝当前时间＋时间间隔＝8秒＋5秒＝13秒。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

向动提交任务运行时，调用COmmltOffsetsAsync()方法，采用异步模式提交偏移盘。消费者发送OFFSET\_COMMIT请求给协调者，“提交偏移量请求”定义的响应处理回调器(OffsetCommitResponseHandler)实现了CoordlnatorResponseHandler抽象父类。提交偏移量的逻辑定义了多个回调方法，在回调方法的不同阶段都会处理不同的业务逻辑。

(1)消费者收到“提交偏移量请求”的响应结果，更新分区状态的“提交偏移£1＂变量。
    (2)组合模式异步请求对象的监昕器回调方法中，调用OffsetCOmmltCallback的回调方法。
    (3)偏移量提交的最后一个回调方法，会创建新的延迟任务。

下面3段代码中出现的offsets参数和变量都来自于订阅状态的allConsumed()方法，消费者“所有消费的分区偏移量”实际上来自于分区状态对象(ToplcPartltlonState)的拉取偏移盐(posltlon变量)，而不是提交偏移量(COmmltted):

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

异步发送请求，没有使用阻塞式的cli.ent.poll(future)轮询，而是使用无阻塞的cli.ent.qi.ckPoll()。由于向动提交任务不需要关心结果，也不会通过future.value()获取异步请求的结果。但发送完请求后不关心结果，并不代表着不处理结果，客户端请求一定都有对应的客户端响应结果。

客户端提交偏移量后，如果没有响应结果返回，或者客户端不处理响应结果。那么仅仅将分状  
态中拉取偏移hl(posHi.on)代表的提交偏移量(consl』附dOffset)保存到服务端，而没有更分区状态的提交偏移址(COmmitted)，就会导致服务端保存的“己提交偏移量”和消费者本地订阅状态的“提交偏移盐”数据不一致。

[2021041216283153.png]: /images/20221022/9f370d7eae494cd49c23494377539704.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/cbbdd297398346969119903d28face54.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/94b71011cc5a4426998b04f75aabd18b.png