带你吃透Kafka的可靠性设计
1 前言
本文里面涉及到较多基础概念，如果忘记了，那么可以去看下《一文带你快速入门kafka》。对于一个消息中间件而言，可靠性是是至关重要的要素之一。不管是面试或者实际工作中，我们都不得不面对几个问题：是几个九？消息会不会丢失？如何保证幂等？如何顺序消费？在这篇文章中，笔者会和大家一起去看 Kafka 是如何设计的。
2 可靠性分析
针对上面的几个问题，Kafka 需要考虑包括并不限于以下问题：- 可用性
如何避免脑裂问题（这个要了解 Kafka 的 leader 选举机制）
– 多副本机制，支持容灾备份
- 数据一致性如何保证
- 数据同步要如何实现
- 消息问题
broker回复投递成功，但是消息丢失了。出现这种情况，一般是以下几种情况：
- acks 配置成 0，生产者以为自己投递成功了，但其实并没有成功写入 leader
- 消息持久化在 leader 分区，但是没有同步给 follower 就宕机了
– 消费者消费消息遇到消息丢失或者消息重复处理
- 消息丢失
- 消费者拿到消息了，但是处理过程中发生异常
- 消费者提交消费位移的设计不合理
- 重复消费
– 消息需要有序消费
我们知道 Kafka 是分区内消息有序的。当然，需要有序的消息就只能使用一个分区，无疑是以 Kafka 的水平扩展能力作为代价的。如果是需要全局有序，而我们又确定使用 Kafka，而且单分区的吞吐量不能满足要求，那么我们只能自己进行额外设计来保证了。
2.1 acks配置对消息丢失的影响
2.1.1 acks=1
消息成功写入 leader 后，就会告诉生产者投递成功。
如上图例子，一共三个分区，其中 follower1 和 follower2 均属于 ISR。假设 leader 成功写入 3 和 4 之后，ISR 还没同步，leader 就宕机了，这样就丢失了 3 和 4，如下图：
2.1.2 acks=-1 或者 acks=all
消息不仅要成功写入 leader，还要 ISR 中的所有 follower 同步完成后，才会告知生产者投递成功。还是 2.1.1 的例子，这里无非会有两种情况：
- leader 在同步完成后宕机
- leader 在同步完成前宕机
这个配置对 Kafka 的性能会有较大影响，需要自己斟酌得失。
2.2 unclean.leader.election.enable
这个配置是用来控制 Kafka 是否可以选举非 ISR 中的副本为 leader，0.11.0.0 之后的版本默认为 false。虽然设置为 true 可以提高 Kafka 的可用性，但是会降低 Kafka 数据的可靠性。2.3 小结
上面提出的问题均有指出和 Kafka 相关部分的设计是哪些，这里再总结一下：- 如何避免脑裂问题——了解 Kafka 的 leader 选举机制
- 数据同步&数据一致性问题——了解 Kafka 的多副本设计
- 消息顺序消费问题——了解 Kafka 的日志同步机制（分区有序）
3.副本设计
副本（ Replica ）是分布式系统中常见的概念之一，指的是分布式系统对数据和服务提供的一种冗余方式。我们知道 Kafka 通过多副本机制，增强了容灾备份的能力，并且实现了故障转移。这无疑是大大提高了 Kafka 的可用性，下面笔者会带着大家一起看 Kafka 的副本机制是如何设计的。
在此之前，先简单复习几个相关的概念：
- 副本是相对分区而言的，即副本是指某个分区的副本
- 在多副本情况下，其中一个副本为 leader，其它均为 follower。只有 leader 对外提供服务，follower 仅同步leader 数据
- 分区中所有的副本集合称为 AR，ISR 是与 leader 保持同步状态的副本集合，leader 也是 ISR 中的一员
- LEO 是每个分区下一条消息要写入的位置
- HW 是 ISR 中最小的 LEO，消费者只能拉取 HW 之前的消息
3.1 失效副本
正常情况下，分区中所有副本都应该属于 ISR，但是网络具有不可靠性。因此，难免在某一个时刻会有一些成员会被踢出 ISR，这些副本要么处于同步失效状态，要么功能失效，这些副本统称为失效副本。功能失效指的是无法工作，比如某个 broker 宕机了，那么在它上面的分区就会失效。
同步失效又是怎么判断是否同步失效的呢？是通过参数 replica.lag.time.max.ms 来判断的，默认是 10000 毫秒。当某个 follower 同步滞后 leader 的时间超过 10 秒，则判定为同步失效。
具体实现原理如下：
当 follower 将 leader LEO 之前的消息全部同步完成，那么会认为该 follower 已经追上 leader，并更新 lastCaughtUpTimeMs。Kafka 的副本管理器有一个副本过期检测的定时任务，如果发现当前时间 - lastCaughtUpTimeMs > 10秒，则判定同步失效。
除了时间设置以外，还有另一个参数 replica.lag.max.message（默认4000，这个是 broker 级别的参数），也是用来判定失效副本的。
一般情况下，这两个参数都是使用默认值就可以，因为如果没有调优经验，自己乱配置，容易导致 ISR 变动过于频繁。同时，需要监控失效副本的数量，因为它是衡量 Kafka 是否健康的一个很重要的指标。
PS：新加入的副本因子/宕机恢复后重新加入的副本在追赶上 leader 之前，也会一直处于失效状态。
3.1.1 失效副本的作用
失效副本为 Kafka 带来了什么收益呢？为什么需要设计这么一个状态呢？大家不妨试想下：假设允许 ISR 中有一个副本同步一直跟不上 leader。当 leader 发生宕机时，这个 follower 被选举成了新的 leader，那么这时就会发生消息丢失。
一般会造成副本失效基本是以下两个原因：
- follower 副本进程卡顿，在一段时间内无法发起同步请求，比如说频繁发生 FULL GC
- follower 同步过慢，在一段时间内无法追上 leader，比如 I/O有问题（笔者实际工作中遇到过一次，公司搭建自己的物理机房，用了二手服务器，有一台服务器I/O老化导致读写数据慢，导致了副本失效，消息堆积等问题）
3.2 LEO 与 HW
这一小节会更进一步去讲解它们之间的关系，让大家可以更清楚 Kafka 的副本同步机制。假设现在有 3 个 broker，某个 topic 有 1 个分区，3 个副本。现在有一个 producer 发送了一条消息，这 3 个副本会发生些什么操作。
具体步骤如下：
- producer 发送消息到 leader
- leader 将消息追加到日志，并且更新日志的偏移量
- follower 执行定时任务向 leader 发送 fetch request 同步数据，该请求会带上自己的 LEO
- leader 读取本地日志，并更新 follower 的信息
- leader 返回 fetch response 给 follower，response 会包含 HW
- follower 将消息追加到本地日志，并更新日志的偏移量
1.一个新建的 topic 被写入了 5 条消息，两个 follower 去拉取数据
2.leader 给 follower 返回 fetch response，并且 leader 又被写入了 5 条消息
其中 follower1 同步了 2 条数据，而 follower2 同步了 3 条数据。
而 follower 的 HW = min(自己的LEO, 同步回来的HW)
3.follower 再次同步数据，同时 leader 又被写入了 5 条消息
leader 更新了 HW
4.leader 给 follower 返回 fetch response
根据公式，follower 更新 HW = 3
在一个分区中，leader 所在 broker 会记录所有副本的 LEO 和 自己的 HW；而 follower 所在的 broker 只会记录自己的 LEO 和 HW。因此，在逻辑层面上，我们可以得到下图：
0.11.0.0版本之前，Kafka 是基于 HW 的同步机制，但是这个设计有可能出现数据丢失和数据不一致的问题。Kafka 后面的版本通过 leader epoch 来进行优化。
3.3 数据丢失 & 数据不一致的解决方案
3.2小节说到了 LEO 与 HW 的更新机制，并且提到这种设计可能会出现数据丢失和数据不一致。我们先一起来看下这两个问题是如何产生的。3.3.1 数据丢失
假设某一分区在某一时刻的状态如下图（L 代表是 leader）：
可以看见副本A的 LEO 是 2，HW 是 1；副本B的 LEO 是 2，HW 是 2。显然，哪怕没有新的消息写入副本B中，副本A也要过一小段时间才能追上副本A，并更新 HW。
假设在副本A更新 HW = 2之前，A宕机了，随后立马就恢复。这里会有一个截断机制——根据宕机之前持久化的HW 恢复消息。也就是说，A只恢复了 m1，m2 丢失了。
再假设 A 刚恢复，B 也宕机了，A 成为了 leader。这时 B 又恢复了，并成为了 follower。由于 follower 的 HW 不能比 leader 的 HW 高，所以 B 的 m2 也丢失了。
总结：这里大家可以发现 follower 的 HW 更新是有一定间隙的，像我这个例子其实 follower 是拿到 m2 了，只不过 HW 要追上 leader 需要等下一次的 fetch request。除非配置 acks=-1 并且配置min.insync.replicas 大于 1，unclean.leader.election.enable = true 才行。
3.3.2 数据不一致
假设某一分区在某一时刻，副本A 的 HW = 2，LEO = 2；副本B 的 HW = 1，LEO = 1。
又假设它们同时挂了，B 先恢复。这时，B 会成为 leader，如下图：
此时，B 写入新消息 m3，并将 HW、LEO 更新为 2。此时，A 也恢复了。由于 A 的 HW 也是 2，所以没有截断消息。如下图：
这样一来，A 中 offset = 1 的消息是 m2，B 中 offset = 1 的消息是 m3，数据不一致了。
3.3.3 leader epoch
为了解决 3.3.1 和 3.3.2 的问题，Kafka 从 0.11.0.0 开始引入 leader epoch，在需要截断时使用 leader epoch 作为依据，而不再是 HW。如果看框架代码比较多的同学应该知道 epoch 是相当于版本的这么一个概念。leader epoch 的初始值是 0，每变更一次 leader，leader epoch 就会增加 1。另外，每个副本中还会增加一个矢量<LeaderEpoch => StartOffset>，其中 StartOffset 是当前 leader epoch 下写入第一条消息的偏移量。每个副本的 Log 下都有一个 leader-epoch-checkpoint 文件，在发生 leader 变更时，会将对应的矢量追加到这个文件中。
3.3.3.1 解决数据丢失问题
还是3.3.1的例子，只不过多了 leader epoch 矢量信息。
副本A：HW=1，LEO=2，LE(leader epoch)=0，Offset(StartOffset)=0
leader-副本B：HW=2，LEO=2，LE=0，Offset(StartOffset)=0
假设在副本A更新 HW = 2之前，A宕机了，随后立马就恢复。不过这里不会立马进行截断日志操作，而是会发送一个 OffsetsForLeaderEpochRequest 请求给 B，B 作为目前的 leader 在收到请求之后会返回 OffsetsForLeaderEpochResponse 给 A。
我们先来看下 OffsetsForLeaderEpochRequest 和 OffsetsForLeaderEpochResponse 的数据结构。如下图：
- OffsetsForLeaderEpochRequest
A 会将自己的 leader epoch 信息给 leader（A的 leader epoch 这里简化成 LE_A）。这里会出现两种情况：
– 变更了 leader
B 会返回 LE_A+1 的 StartOffset 给 A
– 没有变更 leader
B 会返回 A 的 LEO 给 A
因此，我们可以把 OffsetsForLeaderEpochRequest 看作是一个查询 follower 当前 leader_epoch 的 LEO。
- OffsetsForLeaderEpochResponse
这个例子中，B 会返回2给 A，而此时的 A 的 LEO 刚好是 2，所以不用进行截断日志。如下图：
如果此时B挂了，A成了 leader，并有 m3 写入，就会得到下图：
可以看见 m2 并没有丢失，并且也更新了 leader_epoch 矢量为 (1,2)。
3.3.3.2 解决数据不一致问题
上图是3.3.2的例子。副本A是 leader，B 是 follower。
A 的 HW=2，LEO=2，LE=（0，0）
B 的 HW=1，LEO=1，LE=（0，0）
此时，A 和 B 同时宕机，并且 B 先恢复成为了 leader。此时，epoch 变成了 1。另外，新消息 m3 成功写入，就会得到下图：
接着，A 也恢复了，这时 A 不会急着截断日志，而是给 leader 发送 OffsetsForLeaderEpochRequest，B 会返回 LEO = 1 给 A。因此，A 会截断日志，删除 m2。之后，再给 B 发送 fetch request，得到 B 的响应并更新后，将得到下图：
这样数据不一致问题就解决了。
这里大家可能会有疑问，m2不是丢失了吗？是的，这种设计因为更新具有一定的间隙，并且没有事务管理，所以会有丢失消息的风险。
从 CAP 定理来看，这里的设计属于 AP。为什么这么说呢？大家不妨想一下，如果为了不丢失数据，这里加了事务控制的设计，那么对于分区而言它的吞吐量是会下降的，甚至是不可用的，因为响应速度是由短板的副本所决定的。对于定位是高吞吐量的 Kafka 而言，这显然是不可接受的。
3.4 小结
Kafka 通过多副本机制增强了容灾备份的能力，并且基于多副本机制实现了故障转移，避免了单点问题，但同时也引进了新的问题——数据丢失和数据不一致。从 0.11.0.0 版本开始，Kafka 增加了 leader epoch，它对这两个问题进行了优化。虽然无法完全避免消息丢失，但是从实际的使用角度而言，这个问题其实并不大。有实际工作经验的同学应该都知道，我们发送消息难以避免需要重推，哪怕消息中间件做到了百分百不丢失，其实我们在使用时仍然会做防止消息丢失的设计。相对而言，数据一致性就更重要了，否则很容易让订阅消息的下游系统出现脏数据。4 leader 选举机制
在 Kafka 集群中会有一个或者多个 broker，其中有一个 broker 会被选举为控制器，它负责管理整个集群中所有分区和副本的状态。分区的 leader 出现故障时，由控制器负责为其选举新的 leader；当某个分区的 ISR 发生变化时，由控制器负责通知所有 broker 更新其元数据信息；当某个 topic 的分区数量发生变化时，还是由控制器负责分区的重新分配。因此，只要控制器正常工作，分区的 leader 就是唯一的，不会有脑裂问题。那么， Kafka 是如何保证控制器只有一个的呢？如果控制器发生异常了怎么办？控制器的选举和异常恢复又是怎样的？
4.1 控制器
控制器是 broker 维度的角色，它负责管理整个集群中所有分区和副本的状态。Kafka 中的控制器选举工作依赖于 ZooKeeper，成功竞选为控制器的 broker 会在 ZooKeeper 中创建 /controller 临时节点，节点会存储以下信息：
{
"version ": 1,
"brokerid": 0,
"timestamp": "1529210278988"
}
- 1.
- 2.
- 3.
- 4.
- 5.
在任意时刻，Kafka 集群中有且仅有一个控制器。每个 broker 启动时会尝试读取 ZooKeeper 的 /controller 节点的 brokerid 的值，如果 brokerid ≠ -1，则表示当前集群已有控制器，broker 就会放弃竞选；如果不存在 /controller 节点，broker 就会尝试创建节点，创建成功的 broker 就会成为控制器，将自己的 ID 赋予 brokerid，而对于创建节点失败的 broker 则会在内存中保存当前控制器的 brokerid 值，这个值标识为 activeControllerId。
上面是启动 Kafka 集群以及正常情况下添加 broker 情况下的选举过程。那么当控制器出现故障时，就需要重新选举了。ZooKeeper 中还有一个与控制器有关的 /controller_epoch 节点，该节点是持久节点，里面存储了一个整型的 controller_epoch 值，初始值是 1。当控制器发生变化时，controller_epoch 就会加 1。每个和控制器交互的请求一定会带上 controller_epoch，当控制器发现请求带上的 controller_epoch 比自己内存的小，那么这个请求则是无效请求；如果请求带上的 controller_epoch 比自己内存的大，说明自己不再是控制器。由此可见，Kafka 是通过 controller_epoch 来保证控制器的唯一性，进而保证相关操作的一致性。
这里再扩展一下，说下作为控制器的 broker 多出来的责任：
- 监听分区相关的变化
– 在 ZooKeeper 的 /isr_change_notification 节点注册 IsrChangeNotificetionHandler，用来处理 ISR 集合变更的动作。
– 在 ZooKeeper 的 /admin/preferred-replica-election 节点注册 PreferredReplicaElectionHandler，用来处理优先副本的选举动作。
- 监听主题相关的变化
– 在 ZooKeeper 的 /admin/delete_topics 节点注册 TopicDeletionHandler，用来处理删除主题的动作。
- 监听 broker 相关的变化
- 从 ZooKeeper 中读取当前所有与主题、分区及 broker 有关的信息并进行相应的管理
- 启动并管理分区状态机和副本状态机
- 更新集群的元数据信息
- 如果设置了 auto.leader.rebalance.enable = true，则还会开启一个名为“auto-leader-rebalance-task”的定时任务来负责维护分区的优先副本的均衡
控制器、非控制器 broker、ZooKeeper 的关系图如下：
只有控制器会注册相应的监听器关注节点的数据变化，其他 broker 则不关注这些节点的数据变化（除了 /controller）。因为所有 broker 都会关心当前的控制器到底是谁，当 /controller 的数据发生变化时，就要更新自己内存中的 activeControllerId。如果原来是控制器的 broker，发现自己现在不是了，就需要关闭资源，如注销只有控制器才需要的监听器等。不管什么原因造成 /controller 的 brokerid 变更，再重新选举控制器之前，要先确定参选的 broker 里面是否有前控制器，如果有，就要先“退位”，再开始新的选举。
优点：只有控制器注册监听器，可以有效避免严重依赖 ZooKeeper 的设计的通病——脑裂、羊群效应、ZooKeeper 过载。
5 日志同步机制
在多副本的设计中，要实现数据一致性和顺序性，最简单有效的办法就是选举 leader，由 leader 负责写入顺序，follower 复制同步即可。只要 leader 不出问题，如宕机、脑裂等，那么就不需要担心 follower 的数据同步问题。不过，一个分布式系统肯定需要考虑故障转移的。这时就需要考虑 leader 宕机后，选举新 leader 的问题。上面讲了控制器会保证分区 leader 的唯一性，但是数据丢失的问题，还是需要 follower 里面有跟上 leader 的才行。因此，Kafka 里面会有 ISR 这么一个概念。另外，如果需要做到告知客户端成功提交了某条消息，就需要保证新 leader 里面必需有这条消息，那么需要配置 acks=all 等相关配置。
