南通城乡住房建设厅网站首页成都关键词seo推广电话

概要

继上一篇讲Producer原理的文章过去已经一个多月了，今天来讲讲Consumer的原理。
其实源码早就读了部分了，但是最近工作比较忙，一直没空写文章。

整体架构流程

技术名词解释

• coordinator：Consumer协调器，负责管理Consumer需要加入到哪个消费组、消费哪个partition、提交offset等操作
• fetcher：主要作用是获取待消费的records，也是Consumer端最重要的组件
• keyDeserializer：对record中的key进行反序列化
• valueDeserializer：对record中的value进行反序列化
• client：执行RPC请求时的网络client，当然会包括一些Kafka内部的操作

技术细节

coordinator

其实协调器对于Consumer的处理分为几个阶段：

1. Consumer加入的时候：负责判断Consumer加入到哪个Consumer group、协调消费哪个partition
2. Consumer消费过程中：负责记录Consumer消费的partition的元数据、partition的消费状态、消费offset；更新partition的offset

fetcher

从Fetcher的数据结构里其实就可以猜到它的作用：缓存已Fetch到的records、去fetch更多的records

• completedFetch：每次fetch请求得到的数据，拆分到topicPartition维度。因为fetch请求是基于server的node维度，请求回来的数据按照tp维度拆分，得到不同的completedFetch
• completedFetchs: 已经fetch到的所有completedFetch
• nextInLineRecords：当前正在被消费消息的completedFetch对应的所有records，由于对于同一个tp，当时Producer发消息时，是按照batch维度发送的，所以此时completedFetch里也包含多个batch，每个batch包含多个record，也就是records
  如果缓存里没有消息呢？
  也就是completedFetchs和nextInLineRecords都是空

client

类型是ConsumerNetworkClient，里面包含了一个NetWorkClient。至于NetWorkClient是如何进行数据处理及RPC的，可以参考Producer原理解析那篇文章

• unsent：保存的是当前需要发送的fetchRequest
• pendingCompletion：需要被处理的已完成的请求，其实也就是之前的fetchRequest的response
• client：该client是NetWorkClient，Producer端是直接使用了该client
  所以ConsumerNetworkClient的主要作用：1. 处理之前fetch回来的数据；2. 调用NetWorkClient将当前的fetchRequest发送出去

consumer#poll的主要流程

1. 判断是否需要commit offset（默认情况下，5秒进行一次异步offset的commit）

2. 读取Fetcher的缓存，如果有数据，直接跳转到5

3. 缓存里没有数据，基于coordinator里保存的partition元数据，封装fetchRequest
   

4. 执行client#poll：1. 处理之前fetch回来的数据，解析为completedFetchs；2. 调用NetWorkClient将当前的fetchRequest发送出去；
   

5. 调用自定义的消费逻辑（程序员自己写的Consumer），处理records

全局总览

小结

可以看到Consumer和Producer在逻辑处理上还是有较大不同的。

组件	处理请求	处理方式
producer	主要处理发送消息。对应RPC，主要是写请求	将业务逻辑和IO逻辑解耦。业务逻辑：组装batch；IO逻辑：基于batch组装request并发送request
consumer	既要发送fetchRequest，同时还要处理fetchResponse。对于RPC，读写请求都占比较大	业务逻辑和IO逻辑解耦，但是串行化。业务逻辑：从fetcher里poll已经fetch到的数据；IO逻辑：基于partition元数据组装fetchRequest，处理fetchResponse，发送fetchRequest

Producer的IO是一个Sender线程在异步运行，为什么Consumer不这么干呢？
笔者觉得原因是：
Producer的逻辑是把消息往外发，所以Sender运行的越快，client这边为了维护batch而消耗的资源（内存和CPU越少）；而如果Consumer也这么干，实际消费速度赶不上fetch速度的话，会需要额外的内存和CPU资源来维持更多的completedFetchs，更别说如果发生了rebalance的话，fetch过来的completedFetchs可能都是白fetch了。所以，总结下：1. 兼顾消费速度；2. 兼顾client的资源消耗&性能