4.4.4 消费者的消息处理语义

た入场券 2023-01-13 15:56 342阅读 0赞

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

#### 4.4.4 消费者的消息处理语义 ####

消费者从消息代理节点拉取到分区的消息后，对一条消息的处理语义有下面3种情况。

*  至多一次。消息最多被处理一次，可能会丢失，但绝不会重复传输。
 *  至少一次。消息至少被处理一次，不可能丢失，但可能会重复传输。
 *  正好一次。消息正好被处理一次，不可能丢失，也不可能重复传输。

消费者重新加入消费组会分配到新的分区，为了保证消费者能从分配分区的最近提交位置重新开始拉取井消费消息，消费者可以通过手动方式或定时任务提交分区的偏移量，保存分区的消费进度。消费者处理消息并更新消费进度，有下面几种不同的消息处理语义。

1.  至多一次  
    消费者读取消息，先保存消费进度，然后才处理消息。这样有可能会出现：消费者保存完消费进度，但在处理消息之前挂了。新的消费者会从保存的位置开始，但实际上在这个位置之前的消息可能并没有被真正处理。这种场景对应了“至多一次”的语义，即消息、有可能丢失(没有被处理)。Kafka消费者实现“至多一次”的做法是：设置消费者自动提交偏移量，并且设置较短的提交时间间隔。相关代码如下：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

如图4-39所示，假设消费者l要处理3条消息，它们的偏移量分别是［1,2,3)。如果先提交偏移盐3,然后才开始处理，但消费者l只处理了第一条消息后就失败了，新的消费者2会从消费者l记录的偏移量3开始处理，导致没有处理第二条消息和第三条消息，即消息丢失了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

1.  至少一次

消费者读取消息，先处理消息，最后才保存消费进度。这样有可能会出现：消费者处理完消息，但是在保存消费进度之前挂了。新的消费者从保存的位置开始，有可能会重新处理上一个消费者已经处理过的消息。这种场景对应了“至少一次”的语义，即消息有可能会被重复处理。Kafka消费者实现至少一次的做法是：设置消费者向动提交偏移量，但设置很长的提交间隔(或者关闭向动提交偏移盘)。在处理完消息后，手动调用同步模式的提交偏移量方法。相关代码如下：

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

如图4-40所示，消费者先处理消息，然后才提交偏移量。假设消费者l处理完消息6，但是在提交偏移韭6时失败了，这时偏移量仍然是上一次记录的偏移盘3。消费者2会从偏移fil3开始处理，就会重复处理第三条消息之后的第41516条消息，即消息被重复处理了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

1.  正好一次

实现正好一次的消息处理语义有两种典型的解决方案：在保存消费进度和保存消费结果之间，引人两阶段提交协议；或者让消费者将消费进度和处理结果保存在同一个存储介质中。比如，将读取的数据和偏移盘一起存储到HDFS，确保数据和偏移量要么一起被更新，要么都不会更新。Kafka消费者实现正好一次的做法是：设置消费者不自动提交偏移量，订阅主题时设置自定义的消费者再平衡监昕器(ConsumerRebalanceli.stener)。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]

消费者再平衡监昕器会在分区发生变化时，分别从外部存储系统写入或读取偏移量。只要能保证消费者处理消息的流程和写入偏移量到存储系统是一个原子操作，就可以实现正好一次的消息处理语义。相关代码如下：

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

注意：处理消息和保存偏移量必须是一个原子操作。如果不是原子操作，处理消息和保存偏移量之间发生失败，就又回到“至少一次”的场景。这里的关键还是原子操作，而不是说使用消费者再平衡监听器(ConsufTlerRebalanceli.stener)就可以实现“正好一次”的处理语义。监听器会在分区发生变化时读取或写入外部的偏移量存储，这个外部存储实际上l!tl协调节点没有太大区别。下一章分析协调者时，会详细分析“消费者再平衡监听器”的作用和执行流程。

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

### 4.5 小结 ###

本章主要分析消费者新API的拉取消息流程，消费者消费消息主要和KafkaConsufTler类进行交互。客户端通过subscri.be()方法订阅指定的主题，然后调用poll()方法轮询。轮询主要分成3个步骤：通过拉取器发送拉取请求、通过消费者的网络客户端轮询、从拉取器中获取拉撒结果。下面列举了消费者消费消息以及发生再平衡操作时的具体步骤。

(1)消费者分配到分区，订阅状态中的分区状态，初始时还没有拉取偏移量。
    (2)客户端轮询为没有拉取偏移茸的分区更新位置，会尝试从服务端协调节点读取分区的提交偏移量。
    (3)由于此时没有记录分区的提交偏移量，只能按照客户端设置的重置策略定位到最早或最近的位置。
    (4)消费者根据分区的拉取偏移量，从分区的主副本节点拉取消息，井更新分区状态的拉取偏移量。
    (5)分区有了拉取偏移量，自动提交偏移量的定时任务开始工作。
    (6)定时提交任务会将分区状态最新的拉取偏移量提交到服务端。
    (7)如果分区所有权没有变化，下次拉取消息时，已经存在拉取偏移量的分区不需要更新位置。
    (9)如果分区所有权发生变化，协调者会将分区重新分配给新的消费者。
    (10)新消费者之前没有分配该分区，会从服务端读取其他消费者之前提交的分区偏移量。
    (11)新消费者从分区最近的提交偏移量拉取数据，而且它的定时任务也会提交偏移盘到服务端。
    (12)协调者确保分区一定会分配给消费者，这样让分区一定会被消费者拉取并被消费。

客户端在发送请求后返回一个异步请求对象，表示客户端会在未来的某个时刻收到服务端返回的响应结果。客户端可以在返回的异步请求上添加一个监听器，当异步请求完成时，就会自动触发监昕器的回调方法。异步请求还有其他高级的用法，比如组合模式、链接模式。组合模式返回的是一个新的异步请求，也可以在这个新异步请求上再添加一个监听器，形成组合加监昕器模式。

使用异步请求的步骤有3步：调用发送请求返回异步请求、客户端轮询、获取异步请求的结果。客户端轮询有3种方式。

*  快速轮询，调用该方法后会立即返回到主线程，这是无阻塞的轮询。
 *  带超时时间的轮询，如果在给定时间内没有结果返回，会返回到主线程，这是阻塞的轮询。
 *  没有时间限制的轮询，只有在异步请求完成后才会返回到主线程，这是阻塞的轮询。

客户端发送请求得到的异步请求，它的泛型类型是客户端响应(Cl1.entResponse)。使用“组合加适配器”模式后，可以将客户端响应转换为自定义的类型。比如获取分区的偏移量(LIST\_OFFSETS)返回的异步请求对象是RequestFuture<Long＞，获取分区的提交偏移量(OFFSET\_FETCH)对应类型是Map<Top’i.cPart’i.t’i.on,OffsetAndMetadata＞，加入消费组对应类型是ByteBuffer，心跳和自动提交任务对应类型是RequestFuture<Vo’i.d＞。下面列举了使用异步请求的3种做法，最后都要获取异步请求的结果，用于回调处理。

第一种使用方式：客户端使用组合模式发送请求，返回异步请求对象后，立即调用client.poll(future)进行阻塞式地轮询操作。客户端只有在异步请求完成的时候，才可以获取异步请求的结果。相关代码如下：

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

第二种使用方式：为客户端发送请求返回的异步请求对象添加一个监听器，然后才开始阻塞式的轮询。异步请求监昕器囚调方法的参数是客户端响应对象，当客户端收到服务端的响应结果后，会先将客户端的响应结果设置为异步请求对象的结果值然后调用监昕器的回调方法。异步请求监听器回调方法的参数Cl’i.entResponse实际上就是异步请求对象的结果，它和future.value()数据一样。相关代码如下：

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

第三种使用方式：客户端使用组合发送请求，并给返回的异步请求对象添加一个监听器，然后开始阻塞式的轮询。使用组合模式发送请求和第一种方式一样，给异步请求对象添加监听器和第二种模式一样，所以第三种实际上结合了两种模式，它和第二种模式的区别是异步请求的类型为ByteBuffer，而不是客户端响应。第三种方式的执行步骤和第二种一样，最后都会调用异步请求监昕器的回调。相关代码如下：

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

如图4-41所示，以组合模式的异步请求为例，客户端发送请求并获取响应结果的具体步骤如下。

(1)客户端调用sendRequest()向服务端节点发送请求。
    (2)客户端不需要等待服务端返回结果，返回异步请求。
    (3)在步骤(2)返回的异步请求上添加一个监昕器。
    (4)客户端在异步请求上轮询，会阻塞式地等待请求完成。
    (5)如果异步请求没有完成，则继续轮询。
    (6)当收到服务端返回的响应结果后，调用handleResponse()回调方法。
    (7)在回调方法中会解析出客户端响应，调用异步请求的complete()方法，完成异步请求。
    (8)从客户端响应对象解析出来的数据，会被设置为异步请求的结果值。
    (9)客户端调用异步请求的value()方法，获取异步请求的结果。

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

图4-42总结Kafka的消费者和其他组件的关系。Kafka消费者主要有拉取器、消费者的协调者两个主要的类。拉取器会向服务端拉取消息，消费者的协调者会发送心跳和提交偏移量给服务端的协调者节点。属于同一个消费组的所有消费者涉及消费组相关的请求，都会和服务端的协调者节点通信。

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

本章主要分析的是一个消费者的处理流程，并没有和其他消费者联系起来，以全局的视角来解  
消费组整体的行为。消费者的大部分工作都要和服务端的协调者通信，服务端的协调者会管理同一个消费组的所有消费者。下一章会分析消费者发送请求后，在服务端的协调者上的处理流程。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/c8db340496cb47b6a7190053f6361aea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/992588811ae24d3ca821233325c0701f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/970dc85473df4a808b42d3e277689acc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/fe5be9ffd6c84e96afedfdcd31c8cb42.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/aff2699ecc6c431cae25df5d89a0c6d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/502e2fb6595e4c99befcf33102341c7e.png
[20210412164329967.png]: /images/20221022/f2ce19bf26c74b3c8545af2646ead2aa.png
[20210412164339954.png]: /images/20221022/ff8b8061dee144a4a32c9e62c9b736f6.png
[20210412164349774.png]: /images/20221022/b12daf1c959f4ff59fba7d9c593363f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/b9bf8313a0d046189788c416db9d6e8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]: /images/20221022/a98d3f996da647ad9a70f60ceea112ed.png