Kafka：消费者消费消息流程

╰+哭是因爲堅強的太久メ 2024-03-24 17:06 128阅读 0赞

消费者组初始化过程

4aee20e5335044cbaff1dd63e17c4b6e.png

步骤：

1.每一个Broker都有coordinator（辅助实现消费者组的初始化和分区的分配），根据groupid进行哈希取模得到选举那个coordinator对消费者组进行管理；

2.消费者组内每个consumer都向选举的coordinator发送JoinGroup请求；

3.coordinator选出一个 consumer作为leader；

4.coordinator把要消费的topic情况发送给leader 消费者；

5.leader消费者会负责制定消费方案；

6.leader消费者把消费方案发给coordinator；

7.Coordinator就把消费方案下发给其他consumer；

8.每个消费者都会和coordinator保持心跳（默认3s），一旦超时 （session.timeout.ms=45s），该消费者会被移除，并触发再平衡； 或者消费者处理消息的时间过长（max.poll.interval.ms5分钟），也 会触发再平衡。

消费消息过程

1afa66d1da064ac4b056a1d875d307b5.png

步骤：

1.ConsumerNetworkClient建立了与Kafka集群的通信通道；

2.Consumer调用sendFetches方法要ConsumerNetworkClient调用send方法拉取消息；

中间ConsumerNetworkClient拉取消息的一些参数：

  • Fetch.min.bytes每批次最 小抓取大小，默认1字节
  • fetch.max.wait.ms一批数据最小值 未达到的超时时间，默认500ms
  • Fetch.max.bytes每批次最 大抓取大小，默认50m

3.通过回调保存到队列当中；

4.Consumer从队列中拉取消息；

中间Consumer从队列中拉取消息的一些参数：

  • Max.poll.records一次 拉取数据返回消息的 最大条数，默认500条

5.对消息进行反序列化；

6.如果有拦截器，就进行处理；

7.就是对消息进行处理的业务逻辑。

发表评论

表情：
评论列表 （有 0 条评论，128人围观）

还没有评论，来说两句吧...

相关阅读