基于Kafka2.1解读Consumer原理
文章目录
- 概要
- 整体架构流程
- 技术名词解释
- 技术细节
- coordinator
- fetcher
- client
- consumer#poll的主要流程
- 全局总览
- 小结
概要
继上一篇讲Producer原理的文章过去已经一个多月了，今天来讲讲Consumer的原理。
其实源码早就读了部分了，但是最近工作比较忙，一直没空写文章。
整体架构流程
技术名词解释
- coordinator：Consumer协调器，负责管理Consumer需要加入到哪个消费组、消费哪个partition、提交offset等操作
- fetcher：主要作用是获取待消费的records，也是Consumer端最重要的组件
- keyDeserializer：对record中的key进行反序列化
- valueDeserializer：对record中的value进行反序列化
- client：执行RPC请求时的网络client，当然会包括一些Kafka内部的操作
技术细节
coordinator
其实协调器对于Consumer的处理分为几个阶段：
- Consumer加入的时候：负责判断Consumer加入到哪个Consumer group、协调消费哪个partition
- 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的主要流程
-
判断是否需要commit offset（默认情况下，5秒进行一次异步offset的commit）
-
读取Fetcher的缓存，如果有数据，直接跳转到5
-
缓存里没有数据，基于coordinator里保存的partition元数据，封装fetchRequest
-
执行client#poll：1. 处理之前fetch回来的数据，解析为completedFetchs；2. 调用NetWorkClient将当前的fetchRequest发送出去；
-
调用自定义的消费逻辑（程序员自己写的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的资源消耗&性能