进击消息中间件系列（十）：Kafka 副本（Replication）机制

所谓的副本机制（Replication），也可以称之为备份机制，通常是指分布式系统在多台网络互联的机器上保存有相同的数据拷贝。副本机制有什么好处呢？

• 提供数据冗余。即使系统部分组件失效，系统依然能够继续运转，因而增加了整体可用性以及数据持久性。
• 提供高伸缩性。支持横向扩展，能够通过增加机器的方式来提升读性能，进而提高读操作吞吐量。
• 改善数据局部性。允许将数据放入与用户地理位置相近的地方，从而降低系统延时。

这些优点都是在分布式系统教科书中最常被提及的，但是有些遗憾的是，对于 ApacheKafka 而言，目前只能享受到副本机制带来的第 1 个好处，也就是提供数据冗余实现高可用性和高持久性。我会在这一讲后面的内容中，详细解释 Kafka 没能提供第 2 点和第 3 点好处的原因。

副本定义

在讨论具体的副本机制之前，我们先花一点时间明确一下副本的含义。

我们之前谈到过，Kafka 是有主题概念的，而每个主题又进一步划分成若干个分区。副本的概念实际上是在分区层级下定义的，每个分区配置有若干个副本。

所谓副本（Replica），本质就是一个只能追加写消息的提交日志。根据 Kafka 副本机制的定义，同一个分区下的所有副本保存有相同的消息序列，这些副本分散保存在不同的Broker 上，从而能够对抗部分 Broker 宕机带来的数据不可用。

在实际生产环境中，每台 Broker 都可能保存有各个主题下不同分区的不同副本，因此，单个 Broker 上存有成百上千个副本的现象是非常正常的。

接下来我们来看一张图，它展示的是一个有 3 台 Broker 的 Kafka 集群上的副本分布情况。从这张图中，我们可以看到，主题 1 分区 0 的 3 个副本分散在 3 台 Broker 上，其他主题分区的副本也都散落在不同的 Broker 上，从而实现数据冗余。

副本角色

既然分区下能够配置多个副本，而且这些副本的内容还要一致，那么很自然的一个问题就是：我们该如何确保副本中所有的数据都是一致的呢？特别是对 Kafka 而言，当生产者发送消息到某个主题后，消息是如何同步到对应的所有副本中的呢？针对这个问题，最常见的解决方案就是采用基于领导者（Leader-based）的副本机制。Apache Kafka 就是这样的设计。

基于领导者的副本机制的工作原理如下图所示，我来简单解释一下这张图里面的内容。

• 第一，在 Kafka 中，副本分成两类：领导者副本（Leader Replica）和追随者副本（Follower Replica）。每个分区在创建时都要选举一个副本，称为领导者副本，其余的副本自动称为追随者副本。
• 第二，Kafka 的副本机制比其他分布式系统要更严格一些。在 Kafka 中，追随者副本是不对外提供服务的。这就是说，任何一个追随者副本都不能响应消费者和生产者的读写请求。所有的请求都必须由领导者副本来处理，或者说，所有的读写请求都必须发往领导者副本所在的 Broker，由该 Broker 负责处理。追随者副本不处理客户端请求，它唯一的任务就是从领导者副本异步拉取消息，并写入到自己的提交日志中，从而实现与领导者副本的同步。
• 第三，当领导者副本挂掉了，或者说领导者副本所在的 Broker 宕机时，Kafka 依托于ZooKeeper 提供的监控功能能够实时感知到，并立即开启新一轮的领导者选举，从追随者副本中选一个作为新的领导者。老 Leader 副本重启回来后，只能作为追随者副本加入到集群中。

你一定要特别注意上面的第二点，即追随者副本是不对外提供服务的。还记得刚刚我们谈到副本机制的好处时，说过 Kafka 没能提供读操作横向扩展以及改善局部性吗？具体的原因就在于此。

对于客户端用户而言，Kafka 的追随者副本没有任何作用，它既不能像 MySQL 那样帮助领导者副本“抗读”，也不能实现将某些副本放到离客户端近的地方来改善数据局部性。

方便实现“Read-your-writes”

所谓 Read-your-writes，顾名思义就是，当你使用生产者 API 向 Kafka 成功写入消息后，马上使用消费者 API 去读取刚才生产的消息。

举个例子，比如你平时发微博时，你发完一条微博，肯定是希望能立即看到的，这就是典型的 Read-your-writes 场景。如果允许追随者副本对外提供服务，由于副本同步是异步的，因此有可能出现追随者副本还没有从领导者副本那里拉取到最新的消息，从而使得客户端看不到最新写入的消息。

方便实现单调读（Monotonic Reads）

什么是单调读呢？就是对于一个消费者用户而言，在多次消费消息时，它不会看到某条消息一会儿存在一会儿不存在。

如果允许追随者副本提供读服务，那么假设当前有 2 个追随者副本 F1 和 F2，它们异步地拉取领导者副本数据。倘若 F1 拉取了 Leader 的最新消息而 F2 还未及时拉取，那么，此时如果有一个消费者先从 F1 读取消息之后又从 F2 拉取消息，它可能会看到这样的现象：第一次消费时看到的最新消息在第二次消费时不见了，这就不是单调读一致性。但是，如果所有的读请求都是由 Leader 来处理，那么 Kafka 就很容易实现单调读一致性。

什么是ISR

在生产环境下，因为各种不可抗因素，服务可能会发生宕机，例如对外提供服务的leader副本，如果其发生宕机不可用，将会影响系统的使用，

因此在leader副本发生宕机时，follower副本就发生作用了，kafka将从follower副本中选取一个作为新的leader副本对外提供服务，

当然，并不是所有的follower副本都有资格成为leader，因为有些follower副本可能因为各种原因，此时保存的数据落后于之前的leader，

如果数据落后的follower成为了leader，将会引发消息的丢失因此kafka引入了ISR的概念。

ISR，全称 in-sync replicas，是一组动态维护的同步副本集合，每个topic分区都有自己的ISR列表，ISR中的所有副本都与leader保持同步状态(也包括leader本身)，只有ISR中的副本才有资格被选为新的leader，

Producer发送消息时，消息只有被全部写到了ISR中，才会被视为已提交状态，若分区ISR中有N个副本，那么该分区ISR最多可以忍受 N-1 个副本崩溃而不丢失消息。

follower副本同步

follower副本只做一件事，向leader副本请求数据，副本有如下几个概念：

1.起始位移(base offset)

表示该副本当前所含第一条消息的offset。

2.高水印值(high watermark, HW)

副本高水印值，它保存了该副本最新一条已提交消息的位移。

leader副本的HW值决定副本中已提交消息的范围，也确定了consumer能够消费到的消息的上限，超过HW值的所有消息都被视为未提交成功，consumer看不到这些未提交成功的消息

每个follower副本也都有自己的HW值，不过只有leader的HW值才能决定consumer可以看到的消息数量。

3.日志末端位移(log end offset, LEO)

副本日志中下一条待写入消息的offset，所有副本都需要维护自己的LEO，每当leader接收到producer的消息时，其更新自己的LEO(通常+1)，follower副本接收到了leader的数据后，也会更新自己的LEO，

只有当ISR中的副本都更新了对应的LEO后，leader副本才会向右移动HW值，表示写入成功。

副本同步过程

假设某Kafka集群中(broker1、2、3)仅有一个Topic，该Topic只有一个分区，该分区有3个副本，ISR中也是这3个副本，该Topic中目前没有任何数据，因此3个副本中的LEO和HW都是0。

此时某Producer(Producer的acks参数设置成了-1)向broker1中的leader副本发送了一条消息，接下的流程如下：

• broker1上的leader副本接收到消息，将自己的LEO更新为1
• broker2和3上的follower副本各自发送请求给broker1
• broker1分别将消息推送给broker2、3上的副本
• follower副本收到消息后，进行写入然后将自己的LEO也更新为1
• leader副本收到其他follower副本的数据请求响应(response)后，更新HW值为1，此时位移为0的消息可以被consumer消费

ISR设计

ISR是与leader副本保持同步的集合，这意味着是存在与leader副本无法保持同步的副本的(out-of-sync)，那么如何界定ISR，在Kafka中大体可以分为两种情况，Kafka0.9版本前后的界定ISR方式不同。

0.9版本前的界定方式

relica.lag.max.messages

0.9版本之前，Kafka提供了一个名为relica.lag.max.messages用于控制follower副本落后leader副本的消息数，一旦落后的消息超过这个参数数值，则该follower被视为不同步的副本，随后要被踢出ISR集合中。

举一个案例描述follower副本如何落后于leader副本的情况，假设现有集群broker1、2、3，有一个单分区的Topic，副本数量是3，

leader副本处于broker1中，其他两个broker中的副本都是follower副本，此时relica.lag.max.messages参数的值设置为4，

现有一个Producer每次向该Topic发送3条消息，在正常初始状态下，每个follower副本都是可以追上leader副本的，如下：

随后再一次，Producer发送了一条消息过来，此时broker3发生了full gc，导致无法与leader副本保持一致，于是状态就变成了下图：

此时，leader的LEO已经不再与HW相等，最新生产的这条消息不会被认为已提交，除非broker3上的follower副本被踢出ISR集合，但此时relica.lag.max.messages的值是4，而broker3的副本仅落后一条，因此也不会被踢出，

对于broker3上的副本而言，只需要追上leader的LEO即可，因此当full gc过去后，此时的日志状态如下：

此时leader的HW和LEO再次重叠，两个follower也已经与leader保持同步。

除了FullGC导致的副本同步落后，一般还有下面的几种情况导致同步落后：

• 1.请求速度追不上
  • 由于follower副本所在的broker的网络IO开销过大导致从leader处获取消息缓慢。
• 2.进程卡住
  • 由于频繁GC或者程序bug导致follower无法向leader请求数据。
• 3.副本是新创建的
  • 新增加的副本在创建完毕后会全力追赶leader的进度，在追赶这段时间内，通常与leader副本的状态都是不同步的。

replica.lag.time.max.ms

relica.lag.max.messages 存在的问题

relica.lag.max.messages 参数限定了follower副本与leader副本之间同步时，follower副本可落后的消息数量，例如上面设置的是4，意味着主从副本之间消息同步不可超过这值，

但是，如果Producer一次性发送了4条消息，此时与relica.lag.max.messages的值相等，那么上面的两个follower副本都会认为与leader副本不同步，从而被踢出ISR，

此时两个follower副本都处于存活状态(alive)，且没有任何性能问题，很快就可以追上leader的LEO，并且重新加入ISR，

因此当后续的Producer发送的消息都是4或者大于4时，上面的follower副本被踢出ISR然后重新加入ISR的过程就会一直重复，造成很大的资源开销浪费。

有些用户会将 relica.lag.max.messages 调的过大来解决Producer发送过多的消息时导致的follower副本被来回踢出ISR的情况。

例如将该值设置为4000，这样follower副本就不会被来回踢出了，但是这又会引发另一个问题， relica.lag.max.messages 的值是对全局生效的，即所有的topic都受到该值的影响，

如果kafka集群中有两个topic，topic1和topic2，这两个topic的流量差异比较大，topic1的生产者可能一次性生产了5000条消息，直接突破了relica.lag.max.messages的限定值，又引发了ISR的进出重复，

而topic2的生产者每次仅生产10条消息，relica.lag.max.messages的值过大导致可能topic2的follower副本有些真的已经落后不同步了，

但是需要达到relica.lag.max.messages的值后才会被发现，这样一旦出现宕机重选leader副本，很容易造成数据的丢失。

0.9版本后的界定方式

由于 relica.lag.max.messages 的弊端，很难被把控和调优，在0.9版本之后，kafka采用统一的参数来处于界定ISR不同步，

摈弃了 relica.lag.max.messages ，只留下了 replica.lag.time.max.ms，该值默认时间为10秒。

0.9版本后针对由于请求速度追不上的情况，检测机制做了调整，即如果一个follower副本落后leader的时间持续性的超过了这个参数值，

那么该follower副本则被认定为不同步，这样在出现Producer瞬时峰值流量时，只要follower不是持续性落后，也不会反复在ISR中进出。

Kafka 副本备份机制

Kafka 0.11版本之前水印副本备份机制

步骤

• 初始leader,follower参数都为0，其中leader中的remote LEO记录的是follower的LEO
• 生产者发送消息m1到leader中，更新leader的LEO为1
• follower fetch leader,follower写入日志，更新follwer的LEO为1 - follower再次fetch leader,leader接受到以后，更新leader中的remote LEO为1，并更新HW（取leader中的LEO，remote LEO 最小值）为1，
• follower获取fetch返回信息，leader的HW为1，更新follower自身的HW为1。

更新HW，需要额外的一轮fetch rpc请求。

水印备份机制缺陷

数据丢失

数据不一致/离散

造成上述两个问题的根本原因在于HW值被用于衡量副本备份的成功与否，但HW值的更新是异步延迟的，特别是需要额外的FETCH请求处理流程才能更新，故这中间发生的任何崩溃都可能导致HW值的过期。鉴于这些原因，Kafka 0.11引入了leader epoch来取代HW值。

leader epoch

Kafka 引入了 leader epoch 机制，在每个副本日志目录下都创建一个 leader-epoch-checkpoint 文件，用于保存 leader 的 epoch 信息，如下，leader epoch 长这样：

它的格式为 (epoch offset)，epoch指的是 leader 版本，它是一个单调递增的一个正整数值，每次 leader 变更，epoch 版本都会 +1，offset 是每一代 leader 写入的第一条消息的位移值，比如：

(0, 0)
(1, 300)

以上第二个版本是从位移300开始写入消息，意味着第一个版本写入了 0-299 的消息。

leader epoch工作机制

1）当副本成为 leader 时：

这时，如果此时生产者有新消息发送过来，会首先新的 leader epoch 以及 LEO 添加到 leader-epoch-checkpoint 文件中。

2）当副本变成 follower 时：

• 1.发送 LeaderEpochRequest 请求给 leader 副本，该请求包括了 follower 中最新的 epoch 版本；
• 2.leader 返回给 follower 的相应中包含了一个 LastOffset，如果 follower last epoch = leader last epoch，则 LastOffset = leader LEO，否则取大于 follower last epoch 中最小的 leader epoch 的 start offset 值，举个例子：假设 follower last epoch = 1，此时 leader 有 (1, 20) (2, 80) (3, 120)，则 LastOffset = 80；
• 3.follower 拿到 LastOffset 之后，会对比当前 LEO 值是否大于 LastOffset，如果当前 LEO 大于 LastOffset，则从 LastOffset 截断日志；
• 4.follower 开始发送 fetch 请求给 leader 保持消息同步。

解决数据丢失

解决数据不一致/离散

参考文章：https://www.jianshu.com/p/23483d8eed97 https://blog.csdn.net/cainiao1412/article/details/ 125281771 https://www.cnblogs.com/hongdada/p/ 16918984.html