【Kafka系列】副本机制和请求过程

副本机制

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

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

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

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

Leader 副本

Kafka 在创建分区的时候都要选举一个副本，这个选举出来的副本就是 Leader 领导者副本。

Follower 副本

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

这幅图需要注意以下几点

• Kafka 中，Follower 副本也就是追随者副本是不对外提供服务的。这就是说，任何一个追随者副本都不能响应消费者和生产者的请求。所有的请求都是由领导者副本来处理。或者说，所有的请求都必须发送到 Leader 副本所在的 broker 中，Follower 副本只是用作数据拉取，采用异步拉取的方式，并写入到自己的提交日志中，从而实现与 Leader 的同步
• 当 Leader 副本所在的 broker 宕机后，Kafka 依托于 ZooKeeper 提供的监控功能能够实时感知到，并开启新一轮的选举，从追随者副本中选一个作为 Leader。如果宕机的 broker 重启完成后，该分区的副本会作为 Follower 重新加入。

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

跟随者向领导者发送消息的过程是这样的，先请求消息1，然后再接收到消息1，在时候到请求1之后，发送请求2，在收到领导者给发送给跟随者之前，跟随者是不会继续发送消息的。这个过程如下

跟随者副本在收到响应消息前，是不会继续发送消息，这一点很重要。通过查看每个跟随者请求的最新偏移量，首领就会知道每个跟随者复制的进度。如果跟随者在10s 内没有请求任何消息，或者虽然跟随者已经发送请求，但是在10s 内没有收到消息，就会被认为是不同步的。如果一个副本没有与领导者同步，那么在领导者掉线后，这个副本将不会称为领导者，因为这个副本的消息不是全部的。

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

关于副本机制我们说了这么多，那么副本机制的好处是什么呢？

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

同步复制和异步复制

我在学习副本机制的时候，有个疑问，既然领导者副本和跟随者副本是发送 - 等待机制的，这是一种同步的复制方式，那么为什么说跟随者副本同步领导者副本的时候是一种异步操作呢？

我认为是这样的，跟随者副本在同步领导者副本后会把消息保存在本地 log 中，这个时候跟随者会给领导者副本一个响应消息，告诉领导者自己已经保存成功了，同步复制的领导者会等待所有的跟随者副本都写入成功后，再返回给 producer 写入成功的消息。而异步复制是领导者副本不需要关心跟随者副本是否写入成功，只要领导者副本自己把消息保存到本地 log ，就会返回给 producer 写入成功的消息。下面是同步复制和异步复制的过程

同步复制

• producer 通知 ZooKeeper 识别领导者
• producer 向领导者写入消息
• 领导者收到消息后会把消息写入到本地 log
• 跟随者会从领导者那里拉取消息
• 跟随者向本地写入 log
• 跟随者向领导者发送写入成功的消息
• 领导者会收到所有的跟随者发送的消息
• 领导者向 producer 发送写入成功的消息

异步复制

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

ISR

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 是一个动态调整的集合，而非静态不变的。

Unclean 领导者选举

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

如果你接触过分布式项目的话你一定知道 CAP 理论，那么这种 Unclean 领导者选举其实是牺牲了数据一致性，保证了 Kafka 的高可用性。

你可以根据你的实际业务场景决定是否开启 Unclean 领导者选举，一般不建议开启这个参数，因为数据的一致性要比可用性重要的多。

Kafka 请求处理流程

broker 的大部分工作是处理客户端、分区副本和控制器发送给分区领导者的请求。这种请求一般都是请求/响应式的，我猜测你接触最早的请求/响应的方式应该就是 HTTP 请求了。事实上，HTTP 请求可以是同步可以是异步的。一般正常的 HTTP 请求都是同步的，同步方式最大的一个特点是提交请求->等待服务器处理->处理完毕返回 这个期间客户端浏览器不能做任何事。而异步方式最大的特点是 请求通过事件触发->服务器处理（这时浏览器仍然可以做其他事情）-> 处理完毕。

那么我也可以说同步请求就是顺序处理的，而异步请求的执行方式则不确定，因为异步需要创建多个执行线程，而每个线程的执行顺序不同。

这里需要注意一点，我们只是使用 HTTP 请求来举例子，而 Kafka 采用的是 TCP 基于 Socket 的方式进行通讯

那么这两种方式有什么缺点呢？

我相信聪明的你应该能马上想到，同步的方式最大的缺点就是吞吐量太差，资源利用率极低，由于只能顺序处理请求，因此，每个请求都必须等待前一个请求处理完毕才能得到处理。这种方式只适用于请求发送非常不频繁的系统。

异步的方式的缺点就是为每个请求都创建线程的做法开销极大，在某些场景下甚至会压垮整个服务。

响应式模型

说了这么半天，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 处理线程。

请求类型

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

生产请求

我在 真的，关于 Kafka 入门看这一篇就够了 文章中提到过 acks 这个配置项的含义

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

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

获取请求

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

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

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

元数据请求

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

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

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

你点的每个好看，我都认真当成了喜欢