主要所做的改变有下面这几点：

增加了一个 Event Executor Thread，事件执行线程。从图中可以看出，不管是 Event Queue 事件队列还是 Controller context 控制器上下文都会交给事件执行线程进行处理。将原来执行的操作全部建模成一个个独立的事件，发送到专属的事件队列中，供此线程消费。
将之前同步的 ZooKeeper 全部改为异步操作。ZooKeeper API 提供了两种读写的方式：同步和异步。之前控制器操作 ZooKeeper 都是采用的同步方式，这次把同步方式改为异步，据测试，效率提升了10倍。
根据优先级处理请求，之前的设计是 broker 会公平性的处理所有 controller 发送的请求。什么意思呢？公平性难道还不好吗？在某些情况下是的，比如 broker 在排队处理 produce 请求，这时候 controller 发出了一个 StopReplica 的请求，你会怎么办？还在继续处理 produce 请求吗？这个 produce 请求还有用吗？此时最合理的处理顺序应该是，赋予 StopReplica 请求更高的优先级，使它能够得到抢占式的处理。

3. Kafka 的副本机制

复制功能是 Kafka 架构的核心功能，在 Kafka 文档里面 Kafka 把自己描述为 一个分布式的、可分区的、可复制的提交日志服务。复制之所以这么关键，是因为消息的持久存储非常重要，这能够保证在主节点宕机后依旧能够保证Kafka 高可用。副本机制也可以称为备份机制(Replication)，通常指分布式系统在多台网络交互的机器上保存有相同的数据备份/拷贝。

Kafka 使用主题来组织数据，每个主题又被分为若干个分区，分区会部署在一到多个 broker 上，每个分区都会有多个副本，所以副本也会被保存在 broker 上，每个 broker 可能会保存成千上万个副本。下图是一个副本复制示意图:

如上图所示，为了简单我只画出了两个 broker , 每个 broker 指保存了一个 Topic 的消息，在 broker1 中 分区0 是 Leader，它负责进行分区的复制工作，把 broker1 中的 分区0 复制一个副本到 broker2 的主题 A 的 分区0。同理，主题 A 的 分区1 也是一样的道理。

3.1. Leader 副本

副本类型分为两种：一种是 Leader(领导者) 副本，一种是Follower(跟随者)副本

3.2. Follower 副本

除了 Leader 副本以外的副本统称为 Follower 副本，Follower 不对外提供服务。下面是 Leader 副本的工作方式:

需要注意以下几点：

Kafka 中，Follower 副本也就是追随者副本是不对外提供服务的。这就是说，任何一个追随者副本都不能响应消费者和生产者的请求。所有的请求都是由领导者副本来处理。或者说，所有的请求都必须发送到 Leader 副本所在的 broker 中，Follower 副本只是用做数据拉取，采用异步拉取的方式，并写入到自己的提交日志中，从而实现与 Leader 的同步；

当 Leader 副本所在的 broker 宕机后，Kafka 依托于 ZooKeeper 提供的监控功能能够实时感知到，并开启新一轮的选举，从追随者副本中选一个作为 Leader。如果宕机的 b

【一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义】浏览器打开：qq.cn.hn/FTf 免费领取

roker 重启完成后，该分区的副本会作为 Follower 重新加入。

3.3. Follower 和 Leader 副本同步

Leader 的另一个任务是搞清楚哪个 Follower 的状态与自己是一致的。Follower 为了保证与 Leader 的状态一致，在有新消息到达之前先尝试从 Leader 那里复制消息。为了与 Leader 保持一致，Follower 向 Leader 发起获取数据的请求，这种请求与消费者为了读取消息而发送的信息是一样的。

Follower 向 Leader 发送消息的过程是这样的，先请求消息1，然后再接收到消息1，在时候到请求1之后，发送请求2，在收到领导者给发送给跟随者之前，跟随者是不会继续发送消息的。这个过程如下：
Follower 副本在收到响应消息前，是不会继续发送消息，这一点很重要。通过查看每个 Follower 请求的最新偏移量， Leader 就会知道每个 Follower 复制的进度。

如果 Follower 在10s 内没有请求任何消息，或者虽然 Follower 已经发送请求，但是在10s 内没有收到消息，就会被认为是不同步的。如果一个副本没有与 Leader 同步，那么在 Leader 掉线后，这个副本将不会称为 Leader ，因为这个副本的消息不是全部的。

与之相反的，如果 Follower 同步的消息和 Leader 副本的消息一致，那么这个 Follower 副本又被称为同步的副本。也就是说，如果 Leader 掉线，那么只有同步的副本能够称为 Leader

副本机制的好处是什么？

• 能够立刻看到写入的消息，就是你使用生产者 API 成功向分区写入消息后，马上使用消费者就能读取刚才写入的消息；
• 能够实现消息的幂等性，啥意思呢？就是对于生产者产生的消息，在消费者进行消费的时候，它每次都会看到消息存在，并不会存在消息不存在的情况；

3.4. 同步复制和异步复制

既然 Leader 副本和 Follower 副本是 发送 - 等待机制 的，这是一种同步的复制方式，那么为什么说 Follower 副本同步 Leader 的时候是一种异步操作呢？

Follower 副本在同步 Leader 副本后会把消息保存在本地 log 中，这个时候 Follower 会给 Leader 副本一个响应消息，告诉 Leader 自己已经保存成功了，同步复制的 Leader 会等待所有的 Follower 副本都写入成功后，再返回给 producer 写入成功的消息。而异步复制是 Leader 副本不需要关心 Follower 副本是否写入成功，只要 Leader 副本自己把消息保存到本地 log ，就会返回给 producer 写入成功的消息。

同步复制：

1.producer 通知 ZooKeeper 识别领导者；
2.producer 向领导者写入消息；
3.领导者收到消息后会把消息写入到本地 log；
4.跟随者会从领导者那里拉取消息；
5.跟随者向本地写入 log；
6.跟随者向领导者发送写入成功的消息；
7.领导者会收到所有的跟随者发送的消息；
8.领导者向 producer 发送写入成功的消息；

异步复制：

和同步复制的区别在于，领导者在写入本地log之后，直接向客户端发送写入成功消息，不需要等待所有跟随者复制完成。

3.5. ISR（In-Sync Replicas）

Kafka 动态维护了一个同步状态的副本的集合（a set of In-Sync Replicas），简称ISR。

ISR 也是一个很重要的概念，我们之前说过，追随者副本不提供服务，只是定期的异步拉取领导者副本的数据而已，拉取这个操作就相当于是复制，ctrl-c + ctrl-v大家肯定用的熟。那么是不是说 ISR 集合中的副本消息的数量都会与领导者副本消息数量一样呢？那也不一定，判断的依据是 broker 中参数 replica.lag.time.max.ms 的值，这个参数的含义就是跟随者副本能够落后领导者副本最长的时间间隔。

replica.lag.time.max.ms 参数默认的时间是 10秒，如果跟随者副本落后领导者副本的时间不超过 10秒，那么 Kafka 就认为领导者和跟随者是同步的。即使此时跟随者副本中存储的消息要小于领导者副本。如果跟随者副本要落后于领导者副本 10秒以上的话，跟随者副本就会从 ISR 被剔除。倘若该副本后面慢慢地追上了领导者的进度，那么它是能够重新被加回 ISR 的。这也表明，ISR 是一个动态调整的集合，而非静态不变的。

3.6. Unclean 领导者选举

既然 ISR 是可以动态调整的，那么必然会出现 ISR 集合中为空的情况，由于领导者副本是一定出现在 ISR 集合中的，那么 ISR 集合为空必然说明领导者副本也挂了，所以此时 Kafka 需要重新选举一个新的领导者，那么该如何选举呢？现在你需要转变一下思路，我们上面说 ISR 集合中一定是与领导者同步的副本，那么不再 ISR 集合中的副本一定是不与领导者同步的副本了，也就是不再 ISR 列表中的跟随者副本会丢失一些消息。

如果你开启 broker 端参数 unclean.leader.election.enable的话，下一个领导者就会在这些非同步的副本中选举。这种选举也叫做 Unclean 领导者选举。

如果你接触过分布式项目的话你一定知道 CAP 理论，那么这种 Unclean 领导者选举其实是牺牲了数据一致性，保证了 Kafka 的高可用性。你可以根据你的实际业务场景决定是否开启 Unclean 领导者选举，一般不建议开启这个参数，因为数据的一致性要比可用性重要的多。

4. Kafka 的请求处理流程

broker 的大部分工作是处理客户端、分区副本和控制器发送给分区领导者的请求。这种请求一般都是请求/响应式的，我猜测你接触最早的请求/响应的方式应该就是 HTTP 请求了

事实上，HTTP 请求可以是同步可以是异步的。一般正常的 HTTP 请求都是同步的，同步方式最大的一个特点是 提交请求->等待服务器处理->处理完毕返回 这个期间客户端浏览器不能做任何事。

而异步方式最大的特点是 请求通过事件触发->服务器处理（这是浏览器仍然可以作其他事情）-> 处理完毕。

注意: 我们只是使用 HTTP 请求来举例子，而 Kafka 采用的是 TCP 基于 Socket 的方式进行通讯。

以说同步请求就是顺序处理的，而异步请求的执行方式则不确定，因为异步需要创建多个执行线程，而每个线程的执行顺序不同。那么这两种方式有什么缺点呢？

• 同步的方式最大的缺点就是吞吐量太差，资源利用率极低，由于只能顺序处理请求，因此，每个请求都必须等待前一个请求处理完毕才能得到处理。这种方式只适用于请求发送非常不频繁的系统。
• 异步的方式的缺点就是 为每个请求都创建线程的做法开销极大，在某些场景下甚至会压垮整个服务。

4.1. 响应式模型

Kafka 采用同步还是异步的呢？都不是，Kafka 采用的是一种 响应式(Reactor)模型。

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

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 处理线程。

4.2. 生产请求

producer 在向 kafka 写入消息的时候，怎么保证消息不丢失呢？那就是通过 ACK 应答机制！在生产者向队列写入数据的时候可以设置参数来确定是否确认 kafka 接收到数据，这个参数可设置的值为 0、1、all 。

简单来讲就是不同的配置对写入成功的界定是不同的，如果 acks = 1，那么只要领导者收到消息就表示写入成功，如果 acks = 0，表示只要领导者发送消息就表示写入成功，根本不用考虑返回值的影响。如果 acks = all，就表示领导者需要收到所有副本的消息后才表示写入成功。

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

4.3. 获取请求

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

客户端可以设置获取请求数据的上限和下限，上限指的是客户端为接受足够消息分配的内存空间，这个限制比较重要，如果上限太大的话，很有可能直接耗尽客户端内存。下限可以理解为攒足了数据包再发送的意思，这样就增加了时间成本。如下图所示：

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

4.4. 元数据请求

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

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

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

5. Kafka 重平衡流程

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

群组协调器（Coordinator）：群组协调器是一个能够从消费者群组中收到所有消费者发送心跳消息的
broker。在最早期的版本中，元数据信息是保存在 ZooKeeper 中的，但是目前元数据信息存储到了 broker 中。每个消费者组都应该和群组中的群组协调器同步。当所有的决策要在应用程序节点中进行时，群组协调器可以满足 JoinGroup请求并提供有关消费者组的元数据信息，例如分配和偏移量。

群组协调器还有权知道所有消费者的心跳，消费者群组中还有一个角色就是领导者，注意把它和领导者副本和 kafka controller 进行区分。领导者是群组中负责决策的角色，所以如果领导者掉线了，群组协调器有权把所有消费者踢出组。因此，消费者群组的一个很重要的行为是选举领导者，并与协调器读取和写入有关分配和分区的元数据信息。

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

重平衡发生的条件:

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

5.1. 重平衡状态转化

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

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

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

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

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

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

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

5.2. 从消费者看重平衡

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

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

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

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

5.3. 从协调者来看重平衡

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

1. 新成员加入组
2. 组成员主动离开
3. 组成员崩溃离开
4. 组成员提交位移

5.3.1. 新成员加入组

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

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

5.3.3. 组成员崩溃离开

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

5.3.4. 组成员提交位移

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

看到这里的小伙伴，给个❤叭!!!

session.timeout.ms 控制的。如下图所示：
[外链图片转存中…(img-PUiNy45r-1636184587824)]
5.3.4. 组成员提交位移

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

看到这里的小伙伴，给个❤叭!!!