Kafka 源码分析——Consumer

本是古典何须时尚 2023-10-16 11:11 26阅读 0赞

#### 文章目录 ####

*  前言
 *  Consumer 消费流程
 *  Consumer初始化
 *  如何选举Consumer Leader
 *  消费者分区策略
 *  Consumer拉取数据
 *  提交偏移量

## 前言 ##

当生产者将消息发送到Broker时，这些消息将被存储在磁盘上。消费者是如何消费这些消息呢？

## Consumer 消费流程 ##

![image.png][]  
站在源码的核心角度，可以把Consumer分成以下几个核心部分：

1.  Consumer初始化
2.  如何选举Consumer Leader
3.  Consumer Leader是如何指定分区
4.  Consumer如何拉取数据
5.  Consumer偏移量提交

## Consumer初始化 ##

从KafkaConsumer的构造方法出发，跟踪到核心实现方法

![image.png][image.png 1]

这个方法的前面代码部分都是一些配置

**Consumer与Broker的核心通讯组件**

![image.png][image.png 2]

**ConsumerCoordinator：在Kafka消费中是组消费，协调器在具体进行消费之前要做很多的组织协调工作。**

![image.png][image.png 3]

**Fetcher：提取器，因为Kafka消费是拉数据的，所以这个Fetcher就是拉取数据的核心类**

![image.png][image.png 4]

这个核心类中，有很多很多的参数设置，这些就跟消费的配置有关系

**fetch.min.bytes**

每次fetch请求时，server应该返回的最小字节数。如果没有足够的数据返回，请求会等待，直到足够的数据才会返回，缺省为1个字节。多消费者下，可以设大这个值，以降低broker的工作负载。

**fetch.max.bytes**

每次fetch请求时，server应该返回的最大字节数。这个参数决定了可以成功消费到的最大数据。

比如这个参数设置的是50M，那么consumer能成功消费50M以下的数据，但是最终会卡在消费大于10M的数据上无限重试。fetch.max.bytes一定要设置到大于等于最大单条数据的大小才行。

![image.png][image.png 5]

**fetch.wait.max.ms**

如果没有足够的数据能够满足fetch.min.bytes，则此项配置是指在应答fetch请求之前，server会阻塞的最大时间。缺省为500毫秒。和上面的fetch.min.bytes结合起来。

**max.partition.fetch.bytes**

指定了服务器从每个分区里返回给消费者的最大字节数，默认1MB。

假设一个主题有20个分区和5个消费者，那么每个消费者至少要有4MB的可用内存来接收记录，而且一旦有消费者崩溃，这个内存还需更大。注意，这个参数要比服务器的message.max.bytes更大，否则消费者可能无法读取消息。

**max.poll.records**

控制每次poll方法返回的最大记录数量。

![image.png][image.png 6]

## 如何选举Consumer Leader ##

**消费者协调器与组协调器的通讯**

![image.png][image.png 7]

![image.png][image.png 8]

![image.png][image.png 9]

![image.png][image.png 10]

![image.png][image.png 11]

向Broker发送请求

![image.png][image.png 12]  
对Broker响应进行处理  
![image.png][image.png 13]

**消费者协调器发起入组请求**

![image.png][image.png 14]

![image.png][image.png 15]

![image.png][image.png 16]

![image.png][image.png 17]

![image.png][image.png 18]

## 消费者分区策略 ##

**partition.assignment.strategy**，分区分配给消费者的策略。默认为Range。允许自定义策略。

**Range**

把主题的连续分区分配给消费者。（如果分区数量无法被消费者整除、第一个消费者会分到更多分区）

**RoundRobin**

把主题的分区循环分配给消费者。

![image.png][image.png 19]

**StickyAssignor**：初始分区，和RoundRobin是一样。每一次分配变更相对上一次分配做最少的变动。

目标：

1、**分区的分配尽量的均衡**

2、**每一次重分配的结果尽量与上一次分配结果保持一致**

当这两个目标发生冲突时，优先保证第一个目标

比如有3个消费者（C0、C1、C2）、4个topic(T0、T1、T2、T34)，每个topic有2个分区（P1、P2）

![image.png][image.png 20]

**自定义策略**

extends 类AbstractPartitionAssignor，然后在消费者端增加参数：

properties.put(ConsumerConfig.PARTITION\_ASSIGNMENT\_STRATEGY\_CONFIG,类.class.getName());

**分区策略源码分析**

![image.png][image.png 21]

![image.png][image.png 22]

![image.png][image.png 23]

![image.png][image.png 24]

![image.png][image.png 25]

## Consumer拉取数据 ##

拉取数据核心Fetch类

![image.png][image.png 26]

![image.png][image.png 27]

![image.png][image.png 28]

## 提交偏移量 ##

![image.png][image.png 29]

![image.png][image.png 30]

![image.png][image.png 31]

![image.png][image.png 32]

![image.png][image.png 33]

![image.png][image.png 34]

![image.png][image.png 35]

![image.png][image.png 36]

![image.png][image.png 37]

当然，自动提交auto.commit.interval.ms

![image.png][image.png 38]

默认5s

![image.png][image.png 39]

maybeAutoCommitOffsetsAsync 最后这个就是poll的时候会自动提交，而且没到auto.commit.interval.ms间隔时间也不会提交，如果没到下次自动提交的时间也不会提交。

[image.png]: https://img-blog.csdnimg.cn/img_convert/219d8b6131b02119e792d2446d858b76.png
[image.png 1]: https://img-blog.csdnimg.cn/img_convert/e93b03f692289d4a2e636dd9430a95ec.png
[image.png 2]: https://img-blog.csdnimg.cn/img_convert/bed8eae492d896ce5ebc72d795bc12eb.png
[image.png 3]: https://img-blog.csdnimg.cn/img_convert/47487d6f9f19dba74c30a04faf832359.png
[image.png 4]: https://img-blog.csdnimg.cn/img_convert/51d96e786137d4cc3eda71383ca06f8c.png
[image.png 5]: https://img-blog.csdnimg.cn/img_convert/090dbbc4a69d68aac89ee123f549fa48.png
[image.png 6]: https://img-blog.csdnimg.cn/img_convert/c09775d61b3e84abe207b506cce158c2.png
[image.png 7]: https://img-blog.csdnimg.cn/img_convert/41fa0b1c46bcb913963b924c290c71d4.png
[image.png 8]: https://img-blog.csdnimg.cn/img_convert/a1b12f2f42ab76752cf171402e8a8b0f.png
[image.png 9]: https://img-blog.csdnimg.cn/img_convert/bacee378c468284c2da771c32457e4b3.png
[image.png 10]: https://img-blog.csdnimg.cn/img_convert/77194c0fb78b23e1bbc7d41ad1bfa21a.png
[image.png 11]: https://img-blog.csdnimg.cn/img_convert/af4c3b2146258f746225a7983a120bf2.png
[image.png 12]: https://img-blog.csdnimg.cn/img_convert/ecc138b6c32080c20e1cf7c0dc934061.png
[image.png 13]: https://img-blog.csdnimg.cn/img_convert/7195324da84f831f007e3854ef897602.png
[image.png 14]: https://img-blog.csdnimg.cn/img_convert/1227526b3eb62486aa6bcd6b64a161e7.png
[image.png 15]: https://img-blog.csdnimg.cn/img_convert/9258be0c264845b70f69c274f0c22a9e.png
[image.png 16]: https://img-blog.csdnimg.cn/img_convert/b4fe00c678a1736165b1b40f20bd5c99.png
[image.png 17]: https://img-blog.csdnimg.cn/img_convert/451ea9dbaf21880495b1f53a78920652.png
[image.png 18]: https://img-blog.csdnimg.cn/img_convert/deff17fe8964d804910a4252c5ed0dd7.png
[image.png 19]: https://img-blog.csdnimg.cn/img_convert/c1273ee6f87e8074d46f1f555a32c68d.png
[image.png 20]: https://img-blog.csdnimg.cn/img_convert/53c8a8214859f4d45f3438d7cb8387d2.png
[image.png 21]: https://img-blog.csdnimg.cn/img_convert/6d6415691584da96abf52fffd2fcab57.png
[image.png 22]: https://img-blog.csdnimg.cn/img_convert/47dbc8b0bc70f8b140a0854a79bf0300.png
[image.png 23]: https://img-blog.csdnimg.cn/img_convert/7ee8e5cd74ec37a9aa6b2baff2159ab1.png
[image.png 24]: https://img-blog.csdnimg.cn/img_convert/d9a98701fa1a8e400f61eb057f278717.png
[image.png 25]: https://img-blog.csdnimg.cn/img_convert/6ebbebbf56e0f0fbaf1cff618a1d9d58.png
[image.png 26]: https://img-blog.csdnimg.cn/img_convert/4665e51d6cbfa595cf1dad1a9c00c131.png
[image.png 27]: https://img-blog.csdnimg.cn/img_convert/b6d63afae86ea8e5f9bf7e628dd44529.png
[image.png 28]: https://img-blog.csdnimg.cn/img_convert/f0595c1edb092dfeee2836eaa4a23943.png
[image.png 29]: https://img-blog.csdnimg.cn/img_convert/36c4c182ed99388617e0be7e9779c962.png
[image.png 30]: https://img-blog.csdnimg.cn/img_convert/6dc2dc478508d69bac68b8a6160cccb7.png
[image.png 31]: https://img-blog.csdnimg.cn/img_convert/82c308845da321387573c2122a133200.png
[image.png 32]: https://img-blog.csdnimg.cn/img_convert/0d805ff81b98d3858b2808fa7a6f47c4.png
[image.png 33]: https://img-blog.csdnimg.cn/img_convert/922bfe70c807db8694629dbdb05570b8.png
[image.png 34]: https://img-blog.csdnimg.cn/img_convert/3ff352e53274d2d7fbf0efe2547a0a86.png
[image.png 35]: https://img-blog.csdnimg.cn/img_convert/a8ad15efefb161431a6a8b4340108b24.png
[image.png 36]: https://img-blog.csdnimg.cn/img_convert/80aba60165c6cbf53ef156b853956eef.png
[image.png 37]: https://img-blog.csdnimg.cn/img_convert/e6bbbd7c5b7eedf6a3e3cb9076e4721e.png
[image.png 38]: https://img-blog.csdnimg.cn/img_convert/28e9c3fbba5639e09d7eb6b9b9dea097.png
[image.png 39]: https://img-blog.csdnimg.cn/img_convert/02845b1d8c264502695e6cea954b28da.png