3.5.4 服务端处理提交偏移量的请求

淡淡的烟草味﹌ 2023-01-13 04:25 76阅读 0赞

#### 3.5.4 服务端处理提交偏移量的请求 ####

协调节点会将消费者的偏移量提交请交给GroupCoordlnator类的handleCommitOffsets()方法处理，其中参数offsetMetadata表示分配给消费者的所有分区消费进度。相关代码如下：  
![在这里插入图片描述][2021032611450653.png]

写入偏移消息会调用RepllcaManager.appendMessages()方法，将消息集追加到本地日志文件，并且会把分区和对应的偏移量保存在协调节点的缓存中。目的是：再平衡后如果其他消费者需要读取分区的偏移毡，在连接上协调节点后，可以直接读取缓存，而不需要从日志文件中读取。在prepareStoreOffsets()方法内部的putCacheCallback()方法会更新缓存，回调函数的调用只有在主流程即“追加消息到日志文件完成”后才会发生；如果主流程没有完成，回调函数就不会调用。DelayedStore对象包含了需要追加到日志文件的消息集，以及更新缓存的回调方法。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

如图3-27所示，消费者发送提交偏移量和获取偏移虽有1会被j服务端的KafkaAp"i.s处理，服务端处理这两个请求的具体步骤如下。

(1)KafkaApi.s将提交偏移量请求的处理交给消费组的协调者(GroupCoordi.nator)。
    (2)消费组的协调者再交给消费组的元数据管理类(GroupMetadataManager)去处理。
    (3)延迟的存储对象(DelayedStore)会调用副本管理器的appendMessages()存储消息。
    (4)副本管理器将消息追加到底层文件系统的日志文件中，这样分区的偏移量就抒储到服务端了。
    (5)分区和l对应的偏移量会在消息存储成功后，被缓存至服务端的消费组元数据管理类。
    (6)服务端处理客户端的获取分区偏移盐请求，会首先从缓存中获取。
    (7)如果缓存中没有分区的偏移量，就从日志文件中读取。

如图3-28所示，我们用一个示例说明消费者提交偏移量的过程，具体步骤如下。

(1)消费者分配到分区，比如消费者l(Cl)分配到主题(test!)的分区PO和分区Pl。
    (2)分区PO的主副本是消息代理节点1(Broker!)，分区Pl的主副本是消息代理节点2(Broker2),
    消费者创建拉取线程拉取分区消息。
    (3)消费者拉取到每个分区的消息后，客户端迭代每条消息，会更新分区信息对象的消费进度。
    (4)消费者定时提交分区偏移盏，连接消费组的协调节点，消费组l对应内部主题的P1~nBroker2。
    (5)消费者l将向己负责的分区(~PPO和Pl)偏移革A提交到协调节点Broker2上。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]  
内部主题(\_consuMer\_offsets)和普通主题一样也有多个分区，内部主题的分区方式是消费组编号，即相同消费组编号的分区是一样的。所以如果消费者属于同一个消费组，它们提交和读取分区偏移量都是被同一个协调节点处理的。

在前面的再平衡操作中，消费者分配到分区后，会从ZK中读取偏移盘作为分区信息对象(Pat1tionTop1cinfo)的拉取偏移量(fetchedOffset)和消费偏移量(consuMedOffset)。如果偏移量保存在Kafka中，获取偏移量就不是从ZK中读取了，而是从Kafka的内部主题中读取。但读取Kafka需要读取日志文件，为了加快数据的读取，服务端会将内部主题的分区偏移盘缓存起来。

[2021032611450653.png]: /images/20221022/dc953fd91da74c408e7c71e0b927515a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/d3c286147cb44e1ebe132750ff8d7313.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/d1d7c0036e874d6cbfdada9ea85f6231.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/a38381a4a11e437fb43374ba6767d8db.png