全球搜索网站排名,重庆公司章程如何查询下载,域名网站如何做,广东深圳最新消息今天文章目录 概要整体架构流程技术名词解释技术细节coordinatorfetcherclientconsumer#poll的主要流程 全局总览小结 概要
继上一篇讲Producer原理的文章过去已经一个多月了#xff0c;今天来讲讲Consumer的原理。 其实源码早就读了部分了#xff0c;但是最近工作比较忙#x… 文章目录 概要整体架构流程技术名词解释技术细节coordinatorfetcherclientconsumer#poll的主要流程 全局总览小结 概要
继上一篇讲Producer原理的文章过去已经一个多月了今天来讲讲Consumer的原理。 其实源码早就读了部分了但是最近工作比较忙一直没空写文章。
整体架构流程 技术名词解释
coordinatorConsumer协调器负责管理Consumer需要加入到哪个消费组、消费哪个partition、提交offset等操作fetcher主要作用是获取待消费的records也是Consumer端最重要的组件keyDeserializer对record中的key进行反序列化valueDeserializer对record中的value进行反序列化client执行RPC请求时的网络client当然会包括一些Kafka内部的操作
技术细节
coordinator
其实协调器对于Consumer的处理分为几个阶段
Consumer加入的时候负责判断Consumer加入到哪个Consumer group、协调消费哪个partitionConsumer消费过程中负责记录Consumer消费的partition的元数据、partition的消费状态、消费offset更新partition的offset
fetcher 从Fetcher的数据结构里其实就可以猜到它的作用缓存已Fetch到的records、去fetch更多的records
completedFetch每次fetch请求得到的数据拆分到topicPartition维度。因为fetch请求是基于server的node维度请求回来的数据按照tp维度拆分得到不同的completedFetchcompletedFetchs: 已经fetch到的所有completedFetchnextInLineRecords当前正在被消费消息的completedFetch对应的所有records由于对于同一个tp当时Producer发消息时是按照batch维度发送的所以此时completedFetch里也包含多个batch每个batch包含多个record也就是records 如果缓存里没有消息呢 也就是completedFetchs和nextInLineRecords都是空
client
类型是ConsumerNetworkClient里面包含了一个NetWorkClient。至于NetWorkClient是如何进行数据处理及RPC的可以参考Producer原理解析那篇文章
unsent保存的是当前需要发送的fetchRequestpendingCompletion需要被处理的已完成的请求其实也就是之前的fetchRequest的responseclient该client是NetWorkClientProducer端是直接使用了该client 所以ConsumerNetworkClient的主要作用1. 处理之前fetch回来的数据2. 调用NetWorkClient将当前的fetchRequest发送出去
consumer#poll的主要流程 判断是否需要commit offset默认情况下5秒进行一次异步offset的commit 读取Fetcher的缓存如果有数据直接跳转到5 缓存里没有数据基于coordinator里保存的partition元数据封装fetchRequest 执行client#poll1. 处理之前fetch回来的数据解析为completedFetchs2. 调用NetWorkClient将当前的fetchRequest发送出去 调用自定义的消费逻辑程序员自己写的Consumer处理records
全局总览 小结
可以看到Consumer和Producer在逻辑处理上还是有较大不同的。
组件处理请求处理方式producer主要处理发送消息。对应RPC主要是写请求将业务逻辑和IO逻辑解耦。业务逻辑组装batchIO逻辑基于batch组装request并发送requestconsumer既要发送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的资源消耗性能
