浅谈MQ-kafka消费者原理

不念不忘少年蓝@ 2023-01-21 10:26 101阅读 0赞

# Kafka消费者原理 #

## 1.Offset维护 ##

### 1.1 Offset的存储 ###

我们知道在partition中，消息是不会删除的，所以才可以追加写入，写入的消息连续有序的。

这种特性决定了kafka可以消费历史消息，而且按照消息的顺序消费指定消息，而不是只能消费队头的消息。

正常情况下，我们希望消费没有被消费过的数据，而且是从最先发送(序号小的) 的开始消费(这样才是有序和公平的)。

从SimpleConsumer和SimpleProducer中看到的默认结果也是这样的。

对于一个partition，消费者组怎么才能做到接着上次消费的位置(offset) 继续消费呢? 肯定要把这个对应关系保存起来，下次消费的时候查找一下。

(还有一种方式是根据时间戳消费)

首先这个对应关系确实是可以查看的。比如消费者组 group-1和 ass5part (5个分区) 的partition的偏移量关系，可使用如下命令查看：

sh kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095 --describe --group group-1

![image-20210505202142900][]

*  CURRENT-OFFSET：指下一个未使用的offset
 *  LEO：Log End Offset 下一条等待写入的消息的offset（最新offset + 1）
 *  LAG：延迟量
    
    注意：不是一个消费者和一个Topic的关系。是一个consumer group和topic中的一个partition的关系（offset在partition中连续编号而不是全局连续编号）

这个对应关系到底是保存在哪里的呢?首先肯定是不可能放在消费者本地的。为什么?因为所有的消费者都可以使用这个consumer groupid，放在本地是做不到统一维护的，肯定要放到服务端。

kafka早期的版本把消费者组和partition的offset直接维护在ZK中，但是读写的性能消耗太大了。后来就放在一个特殊的topic中，名字叫`__consumer_offsets`，默认有50个分区(offsets.topic.num.partitions默认是50) ，每个分区默认一个replication。

sh kafka-topics.sh --topic __consumer_offsets --describe --zookeeper 127.0.0.1:2181

看起来这些分区副本在3个Broker上非常均匀和轮流地分布(123 123 123…) 。

这样一个特殊的Topic怎么存储消费者组group-1对于分区的偏移量的?

Topic里面是可以存放对象类型的value的(经过序列化和反序列化) 。这个Topic里面主要存储两种对象：

*  Group Metadata：保存了消费者组中各个消费者的信息(每个消费者有编号) 。
 *  Offset And Metadata：保存了消费者组和各个partition的offset位移信息元数据。

sh kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --form-beginning

![image-20210505205853077][]

`__consumer_offsets` 大致的数据结构是这个样子：

![image-20210505210018236][]

怎么知道一个consumer group的offset会放在这个特殊的Topic的那个分区呢？

println(Math.abs("group-1".hashCode()) % 50);

### 1.2 如果找不到Offset ###

当然，这个是Broker有记录offset的情况，如果说增加了一个新的消费者组去消费一个topic的某个partion，没有offset的记录，这个时候应该从哪里开始消费呢?

什么情况下找不到offset?就是你没有消费过，没有把当前的offset上报给Broker。

消费者的代码中有一个参数，用来控制如果找不到偏移量的时候从哪里开始消费。

auto.offset.reset

What to do when there is no initial offset in Kafka or if the current offset does not exist anymore on the server(e.g.because that data has been deleted) ：

*  earliest：automatically reset the offset to the earliest offset
 *  latest：automatically reset the offset to the latest offset
 *  none：throw exception to the consumer if no previous offset is found for the consumers group
 *  anything else：throw exception to the consumer.

默认值是latest，也就是从最新的消息(最后发送的) 开始消费的。历史消费是不能消费的。

earliest：代表从最早的(最先发送的) 消息开始消费。可以消费到历史消息。

none，如果consumer group在服务端找不到offset会报错。

### 1.3 Offset更新 ###

前面我们讲了，消费者组的offset是保存在Broker的，但是是由消费者上报给Broker的。并不是消费者组消费了消息，offset就会更新，消费者必须要有一个commit(提交) 的动作。就跟Rabbit MQ中消费者的ACK一样。

一样的，消费者可以自动提交或者手动提交。由消费端的这个参数控制：

enable.auto.commit

默认是true。true代表消费者消费消息以后自动提交此时Broker会更新消费者组的offset。

另外还可以使用一个参数来控制自动提交的频率：

auto.commit.interval.ms

默认是5秒钟。

如果我们要在消费完消息做完业务逻辑处理之后才commit，就要把这个值改成false。如果是false，消费者就必须要调用一个方法让Broker更新offset。

有两种方式：

*  consumer.commitSync()的手动同步提交
 *  consumer.commitAsync()的手动异步提交

如果不提交或者提交失败，Broker的Offset不会更新，消费者组下次消费的时候会消费到重复的消息。

## 2.消费者消费策略 ##

### 2.1 消费策略 ###

在kafka中，存在三种分区分配策略，一种是RangeAssignor(范围分区-默认)、 另一种是RoundRobinAssignor（轮询分区）、StickyAssignor(粘性分区)。 在消费端中的ConsumerConfig中，通过这个属性来指定分区分配策略

public static final String PARTITION_ASSIGNMENT_STRATEGY_CONFIG = "partition.assignment.strategy";

#### RangeAssignor（范围分区） ####

Range策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。

假设
    n = 分区数 / 消费者数量
    m = 分区数 % 消费者数量
    那么前m个消费者每个分配n+l个分区，后面的（消费者数量-m)个消费者每个分配n个分区

假设我们有11个分区，3个消费者，排完序的分区将会是0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 11；消费者线程排完序将会是C1-0, C2-0, C3-0。然后将partitions的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。在我们的例子里面，我们有10个分区，3个消费者线程，11 / 3 = 4，而且除不尽，那么消费者线程 C1-0, C2-0 将会多消费一个分区

结果：
    C1-0 将消费 0, 1, 2, 3 分区
    C2-0 将消费 4, 5, 6, 7 分区
    C3-0 将消费 8, 9, 10 分区

假如我们有10个分区，那么最后分区分配的结果看起来是这样的：

C1-0 将消费 0, 1, 2, 3 分区
    C2-0 将消费 4, 5, 6 分区
    C3-0 将消费 7, 8, 9 分区

假如我们有2个主题(T1和T2)，分别有10个分区，那么最后分区分配的结果看起来是这样的：

C1-0 将消费 T1主题的 0, 1, 2, 3 分区以及 T2主题的 0, 1, 2, 3分区
    C2-0 将消费 T1主题的 4, 5, 6 分区以及 T2主题的 4, 5, 6分区
    C3-0 将消费 T1主题的 7, 8, 9 分区以及 T2主题的 7, 8, 9分区

**可以看出，C1-0 消费者线程比其他消费者线程多消费了2个分区，这就是Range strategy的一个很明显的弊端**

#### RoundRobinAssignor（轮询分区） ####

轮询分区策略是把所有partition和所有consumer线程都列出来，然后按照hashcode进行排序。最后通过轮询算法分配partition给消费线程。如果所有consumer实例的订阅是相同的，那么partition会均匀分布。

在我们的例子里面，假如按照 hashCode 排序完的topic-partitions组依次为T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9，我们的消费者线程排序为C1-0, C1-1, C2-0, C2-1，最后分区分配的结果为：

C1-0 将消费 T1-5, T1-2, T1-6 分区
    C1-1 将消费 T1-3, T1-1, T1-9 分区
    C2-0 将消费 T1-0, T1-4 分区
    C2-1 将消费 T1-8, T1-7 分区

使用轮询分区策略必须满足两个条件

1.  每个主题的消费者实例具有相同数量的流
2.  每个消费者订阅的主题必须是相同的

#### StrickyAssignor（粘性分区策略） ####

kafka在0.11.x版本支持了StrickyAssignor, 翻译过来叫粘滞策略，它主要有两个目的

*  分区的分配尽可能的均匀
 *  分区的分配尽可能和上次分配保持相同

当两者发生冲突时，第 一 个目标优先于第二个目标。 鉴于这两个目标，StickyAssignor分配策略的具体实现要比RangeAssignor和RoundRobinAssi gn or这两种分配策略要复杂得多，假设我们有这样一个场景

假设消费组有3个消费者：C0,C1,C2，它们分别订阅了4个Topic(t0,t1,t2,t3),并且每个主题有两个分区(p0,p1),也就是说，整个消费组订阅了8个分区：tOpO、tOpl、tlpO、tlpl、t2p0、t2pl、t3p0、t3pl
    
    那么最终的分配场景结果为
    CO: tOpO、tlpl、 t3p0
    Cl: tOpl、t2p0、 t3pl
    C2: tlpO、t2pl
    
    这种分配方式有点类似于轮询策略，但实际上并不是，因为假设这个时候，C1这个消费者挂了，就势必会造成重新分区（reblance），如果是轮询，那么结果应该是
    CO: tOpO、tlpO、t2p0、t3p0
    C2: tOpl、tlpl、t2pl、t3pl
    
    然后，strickyAssignor它是一种粘滞策略，所以它会满足`分区的分配尽可能和上次分配保持相同`，所以分配结果应该是
    CO: tOpO、tlpl、t3p0、t2p0
    C2: tlpO、t2pl、tOpl、t3pl
    
    也就是说，C0和C2保留了上一次是的分配结果，并且把原来C1的分区分配给了C0和C2。 这种策略的好处是使得分区发生变化时，由于分区的“粘性”，减少了不必要的分区移动

### 2.2 rebalance分区重分配 ###

有两种情况需要重新分配分区和消费者的关系：

*  消费者组的消费者数量发生变化，比如新增了消费者，消费者关闭连接 —— 学生数量变多了；
 *  Topic的分区数发生变更，新增或者减少 —— 座位数量发生了变化。

为了让分区分配尽量地均衡，这个时候会触发rebalance机制。

我帮大家简单地总结了一下，分区重新分配分成这么几步：

1.  找出Coordinator
    
    找一个话事人，它起到一个监督和保证公平的作用。每个Broker上都有一个用来管理offset、消费者组的实例，叫做Group Coordinator。
    
    consumer group如何确定自己的coordinator是谁呢, 消费者向kafka集群中的任意一个broker发送一个GroupCoordinatorRequest请求，服务端会返回一个负载最小的broker节点的id，并将该broker设置为coordinator
2.  join group
    
    清点一下人数。所有的消费者连接到Group Coordinator报数，这个叫join group请求。
    
    在rebalance之前，需要保证coordinator是已经确定好了的，整个rebalance的过程分为两个步骤，Join和Sync。
    
    join: 表示加入到consumer group中，在这一步中，所有的成员都会向coordinator发送joinGroup的请求。一旦所有成员都发送了joinGroup请求，那么coordinator会选择一个consumer担任leader角色，并把组成员信息和订阅信息发送消费者
    
    leader选举算法比较简单，如果消费组内没有leader，那么第一个加入消费组的消费者就是消费者leader，如果这个时候leader消费者退出了消费组，那么重新选举一个leader，这个选举很随意，类似于随机算法
    
    ![image][]
    
        protocol_metadata: 序列化后的消费者的订阅信息
        leader_id： 消费组中的消费者，coordinator会选择一个座位leader，对应的就是member_id
        member_metadata 对应消费者的订阅信息
        members：consumer group中全部的消费者的订阅信息
        generation_id： 年代信息，类似于zookeeper的epoch是一样的，对于每一轮rebalance，generation_id都会递增。主要用来保护consumer group。隔离无效的offset提交。也就是上一轮的consumer成员无法提交offset到新的consumer group中。
    
    每个消费者都可以设置自己的分区分配策略，对于消费组而言，会从各个消费者上报过来的分区分配策略中选举一个彼此都赞同的策略来实现整体的分区分配，这个"赞同"的规则是，消费组内的各个消费者会通过投票来决定
    
     *  在joingroup阶段，每个consumer都会把自己支持的分区分配策略发送到coordinator
     *  coordinator收集到所有消费者的分配策略，组成一个候选集
     *  每个消费者需要从候选集里找出一个自己支持的策略，并且为这个策略投票
     *  最终计算候选集中各个策略的选票数，票数最多的就是当前消费组的分配策略（三种分配策略上面有讲到）
3.  选出消费者leader，确定方案
    
    选组长，Group Coordinator从所有消费者里面选一个leader。这个消费者会根据消费者的情况和设置的策略，确定一个方案。Leader把方案上报给Group Coordinator，Group Coordinator会通知所有消费者。
    
    Synchronizing Group State阶段
    
    完成分区分配之后，就进入了Synchronizing Group State阶段，主要逻辑是向GroupCoordinator发送SyncGroupRequest请求，并且处理SyncGroupResponse响应，简单来说，就是leader将消费者对应的partition分配方案同步给consumer group 中的所有consumer
    
    ![image][image 1]
    
    每个消费者都会向coordinator发送syncgroup请求，不过只有leader节点会发送分配方案，其他消费者只是打打酱油而已。当leader把方案发给coordinator以后，coordinator会把结果设置到SyncGroupResponse中。这样所有成员都知道自己应该消费哪个分区。
    
    **consumer group的分区分配方案是在客户端执行的！Kafka将这个权利下放给客户端主要是因为这样做可以有更好的灵活性**
4.  通知分区方案

### 2.3小结 ###

consumer group rebalance的过程

1.  对于每个consumer group子集，都会在服务端对应一个GroupCoordinator进行管理，GroupCoordinator会在zookeeper上添加watcher，当消费者加入或者退出consumer group时，会修改zookeeper上保存的数据，从而触发GroupCoordinator开始Rebalance操作
2.  当消费者准备加入某个Consumer group或者GroupCoordinator发生故障转移时，消费者并不知道GroupCoordinator的在网络中的位置，这个时候就需要确定GroupCoordinator，消费者会向集群中的任意一个Broker节点发送ConsumerMetadataRequest请求，收到请求的broker会返回一个response作为响应，其中包含管理当前ConsumerGroup的GroupCoordinator，
3.  消费者会根据broker的返回信息，连接到groupCoordinator，并且发送HeartbeatRequest(心跳)，发送心跳的目的是要GroupCoordinator这个消费者是正常在线的。当消费者在指定时间内没有发送心跳请求，则GroupCoordinator会触发Rebalance操作。
4.  发起join group请求，两种情况

*  如果GroupCoordinator返回的心跳包数据包含异常，说明GroupCoordinator因为前面说的几种情况导致了Rebalance操作，那这个时候，consumer会发起join group请求
 *  新加入到consumer group的consumer确定好了GroupCoordinator以后消费者会向GroupCoordinator发起join group请求，GroupCoordinator会收集全部消费者信息之后，来确认可用的消费者，并从中选取一个消费者成为group\_leader。并把相应的信息（分区分配策略、leader\_id、…）封装成response返回给所有消费者，但是只有group leader会收到当前consumer group中的所有消费者信息。当消费者确定自己是group leader以后，会根据消费者的信息以及选定分区分配策略进行分区分配
 *  接着进入Synchronizing Group State阶段，每个消费者会发送SyncGroupRequest请求到GroupCoordinator，但是只有Group Leader的请求会存在分区分配结果，GroupCoordinator会根据Group Leader的分区分配结果形成SyncGroupResponse返回给所有的Consumer。
 *  consumer根据分配结果，执行相应的操作

到这里为止，我们已经知道了消息的发送分区策略，以及消费者的分区消费策略和rebalance。对于应用层面来说，还有一个最重要的东西没有讲解，就是offset，他类似一个游标，表示当前消费的消息的位置。

# 为什么kafka能做到这么高的吞吐？ #

MQ的消息存储有几种选择，一种是内存，比如Zero MQ，速度很快但是不可靠。一种是第三方的数据库，会产生额外的网络消耗，而且数据库出问题会影响存储。所以最常见的是把数据放在磁盘上存储。

但是我们也都知道，磁盘的I/O是比较慢的，选择磁盘做为存储怎么实现高吞吐、低延迟、高性能呢?

(案例显示在普通服务器上可以达到百万级TPS)  
https://engineering.linkedin.com/kafka/benchmarking-apache-kafka-2-million-writes-second-three-cheap-machines

**总结起来主要是4点：磁盘顺序I/O、索引机制、批量操作、文件压缩（压紧）、零拷贝。**

# Kafka消息不丢失配置 #

1.  producer端使用 `producer.send(msg, callback)` 带有回调的send方法，而不是 `producer.send(msg)` 方法。根据回调，一旦出现消息提交失败的情况，就可以有针对性地进行处理。
2.  设置acks=all。acks是Producer的一个参数，代表“已提交”消息的定义。如果设置成all，则表明所有Broker都要接收到消息，该消息才算是“已提交”
3.  设置retries为一个较大的值。同样是Producer的参数。当出现网络抖动时，消息发送可能会失败，此时配置了retries的Producer能够自动重试发送消息，尽量避免消息丢失。
4.  设置 `unclean.leader.election.enable = false`。
5.  设置 `replication.factor >= 3`。需要三个以上的副本。
6.  设置 `min.insync.replicas > 1`。Broker端参数，控制消息至少要被写入到多少个副本才算是“已提交”。设置成大于1可以提升消息持久性。在生产环境中不要使用默认值1。确保 `replication.factor > min.insync.replicas` 。如果两者相等，那么只要有一个副本离线，整个分区就无法正常工作了。推荐设置成 `replication.factor=min.insync.replicas + 1`
7.  确保消息消费完成再提交。Consumer端有个参数enable.auto.commit，最好设置成false，并自己来处理offset的提交更新。

[image-20210505202142900]: /images/20221020/0fc924a84b5b428c93d547fab2a52954.png
[image-20210505205853077]: /images/20221020/e4922e6d761c41c0be5df9d5a96cb027.png
[image-20210505210018236]: /images/20221020/cc7fec45779840c2ae9666951c8f0363.png
[image]: /images/20221020/81111b955298468a8659e15f0b458f13.png
[image 1]: /images/20221020/9313238af0a44330ba968088f4cf040c.png