浅谈分布式存储之 Raft

（给ImportNew加星标，提高Java技能）

分布式一致性是分布式系统中最基本的问题，用来保证分布式系统的高可靠。业界也有很多分布式一致性复制协议：Paxos、Zab、Viewstamped Replication、Raft 等。Raft 相比于其他共识算法简化了协议中的状态以及交互，更加清晰也更加容易理解实现。

Raft 节点有 3 种角色：

Raft 信息有 3 种 RPC：

Raft 将时间划分为一个个的任期（Term），TermID 单调递增，每个 Term 最多只有一个 Leader。

Candidate 先将本地的 currentTerm++，然后向其他节点发送 RequestVote 请求。其他节点根据本地数据版本、长度和之前选主的结果判断应答成功与否。具体处理规则如下：

currentTerm 只是用于忽略老的 Term 的 vote 请求，或者提升自己的 currentTerm，并不参与 Log 新旧的决策。Log 新旧的比较，是基于 lastLogTerm 和 lastLogIndex 进行比较，而不是基于 currentTerm 和 lastLogIndex 进行比较。

关于选举有两个很重要的随机超时时间：心跳超时、选举超时。

心跳超时要小于选举超时一个量级，Leader 才能够发送稳定的心跳消息来阻止 Follower 开始进入选举状态。可以设置：心跳超时=peers max RTT(round-trip time)，选举超时=10 * 心跳超时。

大致流程是：更新操作通过 Leade r写入 Log，复制到多数节点，变为 Committed，再 Apply 业务状态机。

如果 Followers crash 或者丢包，Leader 会不断重试 AppendEntries RPC。

每个 log entry 都存储着一条用于状态机的指令，同时保存着从 Leader 收到该 entry 时的 Term，此外还有 index 指明自己在 Log 中的位置。可以被 Apply 的 entry 叫做 committed，一个 log entry 一旦复制给了大多数节点就成为 committed，committed 的 log 最终肯定会被 Apply。

如果当前待提交 entry 之前有未提交的 entry，即使是以前过时的 leader 创建的，只要满足已存储在大多数节点上就一次性按顺序都提交。

Log Recovery 分为 currentTerm 修复和 prevTerm 修复。Log Recovery 就是要保证一定已经 Committed 的数据不会丢失，未 Committed 的数据转变为 Committed，但不会因为修复过程中断又重启而影响节点之间一致性。

currentTerm 修复主要是解决某些 Follower 节点重启加入集群，或者是新增 Follower 节点加入集群，Leader 需要向 Follower 节点传输漏掉的 Log Entry。如果 Follower 需要的 Log Entry 已经在 Leader上Log Compaction 清除掉了，Leader 需要将上一个 Snapshot 和其后的 Log Entry 传输给 Follower 节点。Leader-Alive 模式下，只要 Leader 将某一条 Log Entry 复制到多数节点上，Log Entry 就转变为 Committed。

prevTerm 修复主要是在保证Leader切换前后数据的一致性。通过上面 RAFT 的选主可以看出，每次选举出来的 Leader 一定包含已经 committed 的数据（抽屉原理，选举出来的 Leader 是多数中数据最新的，一定包含已经在多数节点上 commit 的数据）。新的 Leader 将会覆盖其他节点上不一致的数据。虽然新选举出来的 Leader 一定包括上一个 Term 的 Leader 已经 Committed 的 Log Entry，但是可能也包含上一个 Term 的 Leader 未 Committed 的 Log Entry。这部分 Log Entry 需要转变为 Committed，即通过 Noop。

Leader 为每个 Follower 维护一个 nextId，标识下一个要发送的 logIndex。Leader 通过回溯寻找 Follower 上最后一个 CommittedId，然后 Leader 发送其后的 LogEntry。

重新选取 Leader 之后，新的 Leader 没有之前内存中维护的 nextId，以本地 lastLogIndex+1 作为每个节点的 nextId。这样根据节点的 AppendEntries 应答可以调整 nextId：

local.nextIndex = max(min(local.nextIndex-1, resp.LastLogIndex+1), 1)

在实际系统中，Log 会无限制增长，导致 Log 占用太多的磁盘空间，需要更长的启动时间来加载，将会导致系统不可用。需要对日志做压缩。

Snapshot 是 Log Compaction 的常用方法，将系统的全部状态写入一个 Snapshot 中，并持久化到一个可靠存储系统中，完成 Snapshot 之后这个点之前的 Log 就可以被删除了。

当 Raft 集群进行节点变更时，新加入的节点可能会因为需要花费很长时间同步 Log 而降低集群的可用性，导致集群无法 commit 新的请求。

假设原来集群有 3 个节点，可以容忍 3 - (3/2+1) = 11 个节点出错，这时由于机器维修、增加副本解决热点读等原因又新加入了一个节点，这时也是可以容忍 4 - (4/2+1) = 11 个节点出错，恰好原来的一个节点出错了，此时虽然可以 commit 但是得等到新的节点完全追上 Leader 的日志才可以，而新节点追上 Leader 日志花费的时间比较长，在这期间就没法 commit，会降低系统的可用性。

为了避免这个问题，引入了节点的 Learner 状态，当集群成员变更时，新加入的节点为 Learner 状态，Learner 状态的节点不算在 Quorum 节点内，不能参与投票；只有 Leader 确定这个 Learner 节点接收完了 Snapshot，可以正常同步 Log 了，才可能将其变成可以正常的节点。

预选举（Pre-Vote）主要避免了网络分区节点加入集群时，引起集群中断服务的问题。

Follower 在转变为 Candidate 之前，先与集群节点通信，获得集群 Leader 是否存活的信息。如果当前集群有 Leader 存活，Follower 就不会转变为 Candidate，也不会增加 Term，就不会引起 Leader StepDown，从而不会导致集群选主中断服务。

Leader 转移可以把当前 Raft Group 中的 Leader 转换给另一个 Follower，可用于负载均衡、重启机器等。

在进行 transfer leadership 时，先 block 当前 Leader 的写入，然后使 Transferee 节点日志达到 Leader 的最新状态，进而发送 TimeoutNow 请求，触发 Transferee 节点立即选主。

但是不能无限制的 block Leader 的写入，会影响线上服务。通常可以为 transfer leadership 设置一个超时时间。超时之后如果发现 Transferee 节点 Term 没有发生变化，说明 Transferee 节点没有追上数据，没有选主成功，transfer leadership 就失败了。

网络分区主要包含对称网络分区（Symmetric network partitioning）和非对称网络分区（Asymmetric network partitioning）。

对称网络分区

S1 为当前 Leader，网络分区造成 S2 和 S1、S3 心跳中断。S2 既不会被选成 Leader，也不会收到 Leader 的消息，而是会一直不断地发起选举。Term 会不断增大。为了避免网络恢复后，S2 发起选举导致正在工作的 Leader step-down，从而导致整个集群重新发起选举，可以使用 pre-vote 来阻止对称网络分区节点在重新加入时，会中断集群的问题。因为发生对称网络分区后的节点，pre-vote 不会成功，也就不会导致集群一段时间内无法正常提供服务的问题。

S1、S2、S3 分别位于三个 IDC，其中 S1 和 S2 之间网络不通，其他之间可以联通。这样一旦 S1 或者是 S2 抢到了 Leader，另外一方在超时之后就会触发选主，例如 S1 为 Leader，S2 不断超时触发选主，S3 提升 Term 打断当前 Lease，从而拒绝 Leader 的更新。

可以增加一个 trick 的检查，每个 Follower 维护一个时间戳记录收到 Leader 上数据更新的时间，只有超过 ElectionTimeout 之后才允许接受 Vote 请求。这个类似 ZooKeeper 中只有 Candidate 才能发起和接受投票，就可以保证 S1 和 S3 能够一直维持稳定的 quorum 集合，S2 不能选主成功。

Raft 只能在多数节点存活的情况下才可以正常工作，在实际环境中可能会存在多数节点故障只存活一个节点的情况，这个时候需要提供服务并修复数据。因为已经不能达到多数，不能写入数据，也不能做正常的节点变更。Raft 库需要提供一个 SetPeer 的接口，设置每个节点的复制组节点列表，便于故障恢复。

假设只有一个节点 S1 存活的情况下，SetPeer 设置节点列表为 {S1}，这样形成一个只有 S1 的节点列表，让 S1 继续提供读写服务，后续再调度其他节点进行 AddPeer。通过强制修改节点列表，可以实现最大可用模式。

在分布式系统中，对于一个请求都有三种返回结果：成功、失败、超时。

在 failover 时，新的 Leader 由于没有持久化 commitIndex，所以并不清楚当前日志的 commitIndex 在哪，也即不清楚 log entry 是 committed 还是 uncommitted 状态。通常在成为新 Leader 时提交一条空的 log entry 来提交之前所有的 entry。

RAFT 中增加了一个约束：对于之前 Term 的未 Committed 数据，修复到多数节点，且在新的 Term 下至少有一条新的 Log Entry 被复制或修复到多数节点之后，才能认为之前未 Committed 的 Log Entry 转为 Committed。即最大化 commit 原则：Leader 在当选后立即追加一条 Noop 并同步到多数节点，实现之前 Term uncommitted 的 entry 隐式 commit。

元数据相比数据来说整体数据量要小的多，通常单台机器就可以存储。我们也通常借助于 Etcd 等使用单个 Raft Group 来进行元数据的复制和管理。但是单个 Raft Group，存在以下两点弊端：

对于集群元数据来说使用单个 Raft Group 是够了，但是如果想让 Raft 用于数据的复制，那么必须得使用 MultiRaft，也即有多个复制组，类似于 Ceph 的 PG，每个 PG、Raft Group 是一个复制组。

但是 Raft Group 的每个副本间都会建立链接来保持心跳，如果多个 Raft Group 里的副本都建立链接的话，那么物理节点上的链接数就太多了，需要复用物理节点的链接。如下图 cockroachdb multi raft 所示：

MultiRaft 还需要解决以下问题：

Batch 并不会对请求做延迟来达到批量处理的目的，对单个请求的延迟没有影响。

Raft 依赖 Leader 来保持集群的数据一致性，数据的复制都是从 Leader 到 Follower。一个简单的写入流程如下，性能是完全不行的：

Leader 跟其他节点之间的 Log 同步是串行 Batch 的方式，如果单纯使用 Batch，每个Batch 发送之后 Leader 依旧需要等待该 Batch 同步完成之后才能继续发送下一个 Batch，这样会导致较长的延迟。可以通过 Leader 跟其他节点之间的 PipeLine 复制来改进，会有效降低延迟。

顺序提交

将 Leader Append 持久化日志和向 Followers 发送日志并行处理。Leader 只需要在内存中保存未 Committed 的 Log Entry，在多数节点已经应答的情况下，无需等待 Leader 本地 IO 完成，直接将内存中的 Log Entry 直接 Apply 给状态机即可。

乱序提交

乱序提交要满足以下两点：

上层不同的应用场景限制了提交的方式：

不同的分布式存储需要的提交方式：

简单分析

单个 Raft Group 只能顺序提交日志，多个 Raft Group 之间虽然可以做到并行提交日志，但是受限于上层应用（数据库等）的跨 Group 分布式事物，可能导致其他不相关的分布式事物不能并行提交，只能顺序提交。

上层应用比如数据库的分布式事物是跨 Group(A、B、C) 的，Group A 被阻塞了，分布式事务不能提交， 那么所有的参与者 Group(B、C) 就不能解锁，进而不能提交其他不相关的分布式事物，从而引发多个 Group 的链式反应。

Raft 不适用于多连接的高并发环境中。Leader 和 Follower 维持多条连接的情况在生产环境也很常见，单条连接是有序的，多条连接并不能保证有序，有可能发送次序靠后的 Log Entry 先于发送次序靠前的 Log Entry 达到 Follower。但是 Raft 规定 Follower 必须按次序接受 Log Entry，就意味着即使发送次序靠后的 Log Entry 已经写入磁盘了（实际上不能落盘得等之前的 Log Entry 达到）也必须等到前面所有缺失的 Log Entry 达到后才能返回。如果这些 Log Entry 是业务逻辑顺序无关的，那么等待之前未到达的 Log Entry 将会增加整体的延迟。

其实 Raft 的日志复制和 Ceph 基于 PG Log 的复制一样，都是顺序提交的，虽然可以通过 Batch、PipeLine 优化，但是在并发量大的情况下延迟和吞吐量仍然上不去。

具体 Raft 乱序提交的实现可参考：PolarFS: ParallelRaft

http://www.vldb.org/pvldb/vol11/p1849-cao.pdf

我们知道被 committed 的日志肯定是可以被 Apply 的，在什么时候 Apply 都不会影响数据的一致性。所以在 Log Entry 被 committed 之后，可以异步的去 Apply 到业务状态机，这样就可以并行的 Append Log 和 Apply Log 了，提升系统的吞吐量。

其实就和 Ceph BlueStore 的 kv_sync_thread 和 kv_finalize_thread 一样，每个线程都有其队列。kv_sync_thread 去写入元数据到 RocksDB（请求到此已经成功），kv_finalize_thread 去异步的回调上层应用通知请求成功。

Raft 的写入流程会走一遍 Raft，保证了数据的一致性。为了实现线性一致性读，读流程也可以走一遍 Raft，但是会产生磁盘 IO，性能不好。Leader 具有最新的数据，理论上 Leader 可以读取到最新的数据。但是在网络分区的情况下，无法确定当前的 Leader 是不是真的 Leader，有可能当前 Leader 与其他节点发生了网络分区，其他节点形成了一个 Group 选举了新的 Leader 并更新了一些数据，此时如果 Client 还从老的 Leader 读取数据，便会产生 Stale Read。

读流程走一遍 Raft、ReadIndex、Lease Read 都是用来实现线性一致性读，避免 Stale Read。

线性一致性读：在 T1 时刻写入的值，在 T1 时刻之后肯定可以读到。也即读的数据必须是读开始之后的某个值，不能是读开始之前的某个值。不要求返回最新的值，返回时间大于读开始的值就可以。

注意：在新 Leader 刚刚选举出来 Noop 的 Entry 还没有提交成功之前，是不能够处理读请求的，可以处理写请求。也即需要步骤 1 来防止 Stale Read。

原因：在新 Leader 刚刚选举出来 Noop 的 Entry 还没有提交成功之前，这时候的 commitIndex 并不能够保证是当前整个系统最新的 commitIndex。

考虑这个情况：

因为 commitIndex 之后可能还有 Log Entry 对该值更新，只要 w1Apply 到业务状态机就可以满足 applyIndex >= ReadIndex，此时就可以返回 w1 的值。但是此时 w2、w3 还未 Apply 到业务状态机，就没法返回 w3，就会产生 Stale Read。必须等到 Noop 执行完才可以执行读，才可以避免 Stale Read。

如果是热点数据么可以通过提供 Follower Read 来减轻 Leader 的读压力，可用非常方便的通过 ReadIndex 实现。

Lease Read 相比 ReadIndex 更进一步，不仅省去了 Log  的磁盘开销，还省去了Heartbeat的网络开销，提升读的性能。

基本思路

Leader 获取一个比 election timeout 小的租期（Lease）。因为 Follower 至少在 election timeout 时间之后才会发送选举，那么在 Lease 内是不会进行 Leader 选举。就可以跳过 ReadIndex 心跳的环节，直接从 Leader 上读取。但是 Lease Read 的正确性是和时间挂钩的，如果时钟漂移比较严重，那么 Lease Read 就会产生问题。

我们知道 Raft 把数据 Append 到自己的 Log 的同时发送请求给 Follower，多数回复 ACK 就认为 commit，就可以 Apply 到业务状态机了。如果业务状态机（分布式 KV、分布式对象存储等）也把数据持久化存储，那么数据便 Double Write-Store，集群中存在两份相同的数据。如果是三副本，那么就会有 6 份。

接下来主要思考元数据、数据做的一点点优化。

通常的一个优化方式就是先把数据写入 Journal（环形队列、大小固定、空间连续、使用 3D XPoint、NVME），然后再把数据写入内存即可返回，最后异步的把数据刷入 HDD（最好带有 NVME 缓存）。

元数据

元数据通常使用分布式 KV 存储，数据量比较小，Double Write-Store 影响不是很大，即使存储两份也不会浪费太多空间，而且以下改进也相比数据方面的改进更容易实现。

可以撸一个简单的 Append-Only 的单机存储引擎 WAL 来替代 RocksDB 作为 Raft Log 的存储引擎，Apply 业务状态机层的存储引擎可以使用 RocksDB，但是可以关闭 RocksDB 的 WAL，因为数据已经存储在 Append-Only 的 Raft Log 了，细节仍需考虑。

数据

这里的数据通常指非结构化数据：图片、文档、音视频等。非结构化数据通常使用分布式对象存储、块存储、文件存储等来存储，由于数据量比较大，Double Store 是不可接受的，大致有两种思路去优化：

- EOF -

看完本文有收获？请转发分享给更多人

关注「ImportNew」，提升Java技能

点赞和在看就是最大的支持❤️

ImportNew

收藏 分享 链接