Kafka 请求处理流程、重平衡流程

末蓝、 2022-11-18 04:50 232阅读 0赞

# Kafka 请求处理流程 #

## 响应式模型 ##

Kafka 采用的是一种 `响应式(Reactor)模型`，那么什么是响应式模型呢？简单的说，**Reactor 模式是事件驱动架构的一种实现方式，特别适合应用于处理多个客户端并发向服务器端发送请求的场景**，如下图所示

![a28c4707ab2343fdd842514fb2720a8c.png][]

Kafka 的 broker 端有个 SocketServer 组件，类似于处理器，SocketServer 是基于 TCP 的 Socket 连接的，它用于接受客户端请求，所有的请求消息都包含一个消息头，消息头中都包含如下信息

*  Request type （也就是 API Key）
 *  Request version（broker 可以处理不同版本的客户端请求，并根据客户版本做出不同的响应）
 *  Correlation ID — 一个具有唯一性的数字，用于标示请求消息，同时也会出现在响应消息和错误日志中（用于诊断问题）
 *  Client ID — 用于标示发送请求的客户端

broker 会在它所监听的每一个端口上运行一个 `Acceptor` 线程，这个线程会创建一个连接，并把它交给 `Processor(网络线程池)`， Processor 的数量可以使用 `num.network.threads` 进行配置，其默认值是 3，表示每台 broker 启动时会创建 3 个线程，专门处理客户端发送的请求。

Acceptor 线程会采用`轮询`的方式将入栈请求公平的发送至网络线程池中，因此，在实际使用过程中，这些线程通常具有相同的机率被分配到待处理`请求队列`中，然后从`响应队列`获取响应消息，把它们发送给客户端。Processor 网络线程池中的请求 - 响应的处理还是比较复杂的。

Processor 网络线程池接收到客户和其他 broker 发送来的消息后，网络线程池会把消息放到请求队列中，注意这个是`共享请求队列`，因为网络线程池是多线程机制的，所以请求队列的消息是多线程共享的区域，然后由 IO 线程池进行处理，根据消息的种类判断做何处理，比如 `PRODUCE` 请求，就会将消息写入到 log 日志中，如果是`FETCH`请求，则从磁盘或者页缓存中读取消息。也就是说，IO 线程池是真正做判断，处理请求的一个组件。在 IO 线程池处理完毕后，就会判断是放入`响应队列`中还是 `Purgatory` 中，Purgatory 是什么我们下面再说，现在先说一下响应队列，响应队列是每个线程所独有的，因为响应式模型中不会关心请求发往何处，因此把响应回传的事情就交给每个线程了，所以也就不必共享了。

> 注意：IO 线程池可以通过 broker 端参数 `num.io.threads` 来配置，默认的线程数是 8，表示每台 broker 启动后自动创建 8 个 IO 处理线程。

![51e1e6a059e4b5cd8d156b55415810fa.png][]

## 请求类型 ##

下面是几种常见的请求类型

**生产请求**

我在之前的文章中提到过 `acks` 这个配置项的含义，简单来讲就是不同的配置对写入成功的界定是不同的，如果 acks = 1，那么只要领导者收到消息就表示写入成功，如果 acks = 0，表示只要领导者发送消息就表示写入成功，根本不用考虑返回值的影响。如果 acks = all，就表示领导者需要收到所有副本的消息后才表示写入成功。

在消息被写入分区的首领后，如果 acks 配置的值是 `all`，那么这些请求会被保存在 `炼狱(Purgatory)`的缓冲区中，直到领导者副本发现跟随者副本都复制了消息，响应才会发送给客户端。

**获取请求**

broker 获取请求的方式与处理生产请求的方式类似，客户端发送请求，向 broker 请求主题分区中特定偏移量的消息，如果偏移量存在，Kafka 会采用 `零复制` 技术向客户端发送消息，Kafka 会直接把消息从文件中发送到网络通道中，而不需要经过任何的缓冲区，从而获得更好的性能。

客户端可以设置获取请求数据的上限和下限，`上限`指的是客户端为接受足够消息分配的内存空间，这个限制比较重要，如果上限太大的话，很有可能直接耗尽客户端内存。`下限`可以理解为攒足了数据包再发送的意思，这就相当于项目经理给程序员分配了 10 个 bug，程序员每次改一个 bug 就会向项目经理汇报一下，有的时候改好了有的时候可能还没改好，这样就增加了沟通成本和时间成本，所以下限值得就是程序员你改完 10 个 bug 再向我汇报！！！如下图所示

![6376f448005cee0273c03ee7994d14af.png][]

如图你可以看到，在`拉取消息` —> `消息` 之间是有一个等待消息积累这么一个过程的，这个消息积累你可以把它想象成超时时间，不过超时会跑出异常，消息积累超时后会响应回执。延迟时间可以通过 `replica.lag.time.max.ms` 来配置，它指定了副本在复制消息时可被允许的最大延迟时间。

**元数据请求**

生产请求和响应请求都必须发送给领导者副本，如果 broker 收到一个针对某个特定分区的请求，而该请求的首领在另外一个 broker 中，那么发送请求的客户端会收到`非分区首领`的错误响应；如果针对某个分区的请求被发送到不含有领导者的 broker 上，也会出现同样的错误。Kafka 客户端需要把请求和响应发送到正确的 broker 上。怎么知道要往哪发送？

事实上，客户端会使用一种 `元数据请求` ，这种请求会包含客户端感兴趣的主题列表，服务端的响应消息指明了主题的分区，领导者副本和跟随者副本。元数据请求可以发送给任意一个 broker，因为所有的 broker 都会缓存这些信息。

一般情况下，客户端会把这些信息缓存，并直接向目标 broker 发送生产请求和相应请求，这些缓存需要隔一段时间就进行刷新，使用`metadata.max.age.ms` 参数来配置，从而知道元数据是否发生了变更。比如，新的 broker 加入后，会触发重平衡，部分副本会移动到新的 broker 上。这时候，如果客户端收到 `不是首领`的错误，客户端在发送请求之前刷新元数据缓存。

# Kafka 重平衡流程 #

实际上，归纳为一点就是让组内所有的消费者实例就消费哪些主题分区达成一致。

我们知道，一个消费者组中是要有一个`群组协调者(Coordinator)`的，而重平衡的流程就是由 Coordinator 的帮助下来完成的。

这里需要先声明一下重平衡发生的条件

*  消费者订阅的任何主题发生变化
 *  消费者数量发生变化
 *  分区数量发生变化
 *  如果你订阅了一个还尚未创建的主题，那么重平衡在该主题创建时发生。如果你订阅的主题发生删除那么也会发生重平衡
 *  消费者被群组协调器认为是 `DEAD` 状态，这可能是由于消费者崩溃或者长时间处于运行状态下发生的，这意味着在配置合理时间的范围内，消费者没有向群组协调器发送任何心跳，这也会导致重平衡的发生。

## 在了解重平衡之前，你需要知道这两个角色 ##

`群组协调器（Coordinator）`：群组协调器是一个能够从消费者群组中收到所有消费者发送心跳消息的 broker。在最早期的版本中，元数据信息是保存在 ZooKeeper 中的，但是目前元数据信息存储到了 broker 中。每个消费者组都应该和群组中的群组协调器同步。当所有的决策要在应用程序节点中进行时，群组协调器可以满足 `JoinGroup` 请求并提供有关消费者组的元数据信息，例如分配和偏移量。群组协调器还有权知道所有消费者的心跳，消费者群组中还有一个角色就是领导者，注意把它和领导者副本和 kafka controller 进行区分。领导者是群组中负责决策的角色，所以如果领导者掉线了，群组协调器有权把所有消费者踢出组。因此，消费者群组的一个很重要的行为是选举领导者，并与协调器读取和写入有关分配和分区的元数据信息。

`消费者领导者`： 每个消费者群组中都有一个领导者。如果消费者停止发送心跳了，协调者会触发重平衡。

## 在了解重平衡之前，你需要知道状态机是什么 ##

Kafka 设计了一套`消费者组状态机(State Machine)` ，来帮助协调者完成整个重平衡流程。消费者状态机主要有五种状态它们分别是 **Empty、Dead、PreparingRebalance、CompletingRebalance 和 Stable**。

![5fe18383aa793d9c679afae4f0db01ee.png][]

了解了这些状态的含义之后，下面我们用几条路径来表示一下消费者状态的轮转

消费者组一开始处于 `Empty` 状态，当重平衡开启后，它会被置于 `PreparingRebalance` 状态等待新消费者的加入，一旦有新的消费者加入后，消费者群组就会处于 `CompletingRebalance` 状态等待分配，只要有新的消费者加入群组或者离开，就会触发重平衡，消费者的状态处于 PreparingRebalance 状态。等待分配机制指定好后完成分配，那么它的流程图是这样的

![5e94c7da6f072537cef144ad31e9049d.png][]

在上图的基础上，当消费者群组都到达 `Stable` 状态后，一旦有新的**消费者加入/离开/心跳过期**，那么触发重平衡，消费者群组的状态重新处于 PreparingRebalance 状态。那么它的流程图是这样的。

![473e78b15f82b6ad88a9b4743c2b0c66.png][]

在上图的基础上，消费者群组处于 PreparingRebalance 状态后，很不幸，没人玩儿了，所有消费者都离开了，这时候还可能会保留有消费者消费的位移数据，一旦位移数据过期或者被刷新，那么消费者群组就处于 `Dead` 状态了。它的流程图是这样的

![d922ecf17f14b7d81f74390b9f619862.png][]

在上图的基础上，我们分析了消费者的重平衡，在 `PreparingRebalance`或者 `CompletingRebalance` 或者 `Stable` 任意一种状态下发生位移主题分区 Leader 发生变更，群组会直接处于 Dead 状态，它的所有路径如下

![e9fe3cbcbb59631e3ee0a2e2d4d51659.png][]

> 这里面需要注意两点：
> 
> 一般出现 **Required xx expired offsets in xxx milliseconds** 就表明 Kafka 很可能就把该组的位移数据删除了
> 
> 只有 Empty 状态下的组，才会执行过期位移删除的操作。

## 重平衡流程 ##

上面我们了解到了消费者群组状态的转化过程，下面我们真正开始介绍 `Rebalance` 的过程。重平衡过程可以从两个方面去看：消费者端和协调者端，首先我们先看一下消费者端

## 从消费者看重平衡 ##

从消费者看重平衡有两个步骤：分别是 `消费者加入组` 和 `等待领导者分配方案`。这两个步骤后分别对应的请求是 `JoinGroup` 和 `SyncGroup`。

新的消费者加入群组时，这个消费者会向协调器发送 `JoinGroup` 请求。在该请求中，每个消费者成员都需要将自己消费的 topic 进行提交，我们上面描述群组协调器中说过，这么做的目的就是为了让协调器收集足够的元数据信息，来选取消费者组的领导者。通常情况下，第一个发送 JoinGroup 请求的消费者会自动称为领导者。**领导者的任务是收集所有成员的订阅信息，然后根据这些信息，制定具体的分区消费分配方案**。如图

![fcac64494e0d60b2eaddcd7ea2b9c129.png][]

在所有的消费者都加入进来并把元数据信息提交给领导者后，领导者做出分配方案并发送 `SyncGroup`请求给协调者，协调者负责下发群组中的消费策略。下图描述了 SyncGroup 请求的过程

![ac0d55055710c674cd425c74ae5509bb.png][]

当所有成员都成功接收到分配方案后，消费者组进入到 Stable 状态，即开始正常的消费工作。

## 从协调者来看重平衡 ##

从协调者角度来看重平衡主要有下面这几种触发条件，

*  新成员加入组
 *  组成员主动离开
 *  组成员崩溃离开
 *  组成员提交位移

我们分别来描述一下，先从新成员加入组开始

#### 新成员加入组 ####

我们讨论的场景消费者集群状态处于`Stable` 等待分配的过程，这时候如果有新的成员加入组的话，重平衡的过程

![f745b1e6f64fec854db3e04f63db2f56.png][]

从这个角度来看，协调者的过程和消费者类似，只是刚刚从消费者的角度去看，现在从领导者的角度去看

### 组成员离开 ###

组成员离开消费者群组指的是消费者实例调用 `close()` 方法主动通知协调者它要退出。这里又会有一个新的请求出现 `LeaveGroup()请求` 。如下图所示

![8adcac25099dce8f17837ffb5aa2dd04.png][]

### 组成员崩溃 ###

组成员崩溃是指消费者实例出现严重故障，宕机或者一段时间未响应，协调者接收不到消费者的心跳，就会被认为是`组成员崩溃`，崩溃离组是被动的，协调者通常需要等待一段时间才能感知到，这段时间一般是由消费者端参数 session.timeout.ms 控制的。如下图所示

![be8b7c9f3b048646f167520e8cb6ae3f.png][]

### 重平衡时提交位移 ###

这个过程我们就不再用图形来表示了，大致描述一下就是 消费者发送 JoinGroup 请求后，群组中的消费者必须在指定的时间范围内提交各自的位移，然后再开启正常的 JoinGroup/SyncGroup 请求发送。

[a28c4707ab2343fdd842514fb2720a8c.png]: /images/20221022/261d0c64da4446ccb2ae8c715b774d59.png
[51e1e6a059e4b5cd8d156b55415810fa.png]: /images/20221022/63f182c4edeb4f01b983d2f8e94ec5c2.png
[6376f448005cee0273c03ee7994d14af.png]: /images/20221022/f54fff60c2d642508f55691f12a276b3.png
[5fe18383aa793d9c679afae4f0db01ee.png]: /images/20221022/b4add7b5b9294f1b9dcc499b65637f00.png
[5e94c7da6f072537cef144ad31e9049d.png]: /images/20221022/322bcd133ecc43fb9d41306cd8c6a398.png
[473e78b15f82b6ad88a9b4743c2b0c66.png]: /images/20221022/3550bcd6db004e1c9df2e9d88aa6a4f4.png
[d922ecf17f14b7d81f74390b9f619862.png]: /images/20221022/654ec1b8889f4c379f190220998d5b44.png
[e9fe3cbcbb59631e3ee0a2e2d4d51659.png]: /images/20221022/b2b0f5614dbd409da5e4a9664b4f254f.png
[fcac64494e0d60b2eaddcd7ea2b9c129.png]: /images/20221022/18bcf70a3b4b428a9c8a1426037e7741.png
[ac0d55055710c674cd425c74ae5509bb.png]: /images/20221022/420bb7c02a604746a86dc67225361b73.png
[f745b1e6f64fec854db3e04f63db2f56.png]: /images/20221022/32c10c82059448838d71e6510e122342.png
[8adcac25099dce8f17837ffb5aa2dd04.png]: /images/20221022/528e7ecfe24246f0a92dcb23cfbf4847.png
[be8b7c9f3b048646f167520e8cb6ae3f.png]: /images/20221022/955f77bf977f44f0be94f93fde714de9.png