分布式系统是指一组独立的计算机，通过网络协同工作的系统，客户端看来就如同单台机器在工作。随着互联网时代数据规模的爆发式增长，传统的单机系统在性能和可用性上已经无法胜任，分布式系统具有扩展性强、可用性高、廉价高效等优点得以广泛应用。
但与单机系统相比，分布式系统在实现上要复杂很多。CAP 理论是分布式系统的理论基石，它提出以下 3 个要素：
l Consistency（强一致性）：任何客户端都可以访问到同一份最新的数据副本。
l Availability（可用性）： 系统一直处于可服务状态，每次请求都能获得非错的响应。
l Partition-tolenrance（分区可容忍性）：单机故障或网络分区，系统仍然可以保证强一致性和可用性。
一个分布式系统最多只能满足其中 2 个要素。对于分布式系统而言，P 显然是必不可少的，那么只能在 AP 和 CP 之间权衡。AP 系统牺牲强一致性，这在某些业务场景下（如金融类）是不可接受的，CP 系统可以满足这类需求，问题的关键在于会牺牲多少可用性。传统的主备强同步模式虽然可以保证一致性，但一旦机器故障或网络分区系统将变得不可用。paxos 和 raft 等一致性算法的提出，弥补了这一缺陷。它们在保证 CP 的前提下，只要求大多数节点可以正常互联，系统便可以一直处于可用状态，可用性上显著提高。paxos 的理论性偏强，开发者需要自己处理很多细节，这也是它有很多变种的原因，相对而言 raft 更易理解和工程化，一经提出便广受欢迎。
在我们关注的消息中间件领域，金融支付类业务往往对数据的强一致性和高可靠性有严格要求。
在对主流的消息中间件进行调研后，发现它们在应对这种场景时都存在一定的不足：
l RabbitMQ：一个请求需要在所有节点上处理 2 次才能保证一致性，性能不高。
l Kafka：主要应用在日志、大数据等方向，少量丢失数据业务可以忍受，但不适合要求数据高可靠性的系统。
l RocketMQ：未采用一致性算法，如果配置成异步模式可能丢失数据，同步模式下节点故障或网络分区都会影响可用性。
l SQS：只提供最终一致性，不保证强一致性。
鉴于以上分析，我们设计开发了基于 Raft 的强一致高可靠消息中间件 CMQ。接下来会介绍 raft 算法原理细节、如何应用在 CMQ 中在保证消息可靠不丢失，以及实现过程中在性能方面所作的优化。
二 Raft 算法介绍2.1 概述Raft 算法是 Diego Ongaro 博士在论文《In Search of an Understandable Consensus Algorithm》,2014 USENIX 中首次提出，算法主要包括选举和日志同步两部分:
以下是贯穿 Raft 算法的重要术语：
节点之间通过 RPC 通信来完成选举和日志同步，发送方在发送 RPC 时会携带自身的 Term，接收方在处理 RPC 时有以下两条通用规则：
1) RPC 中的 RTerm 大于自身当前 Term，更新自身 Term = RTerm、votedFor = null，转为 Follower。
2) RPC 中的 RTerm 小于自身当前 Term，拒绝请求，响应包中携带自身的 Term。
2.2 选举算法Raft 算法属于强 Leader 模式，只有 Leader 可以处理客户端的请求，Leader 通过心跳维持自身地位，除非 Leader 故障或网络异常，否则 Leader 保持不变。选举阶段的目的就是为了从集群中选出合适的 Leader 节点。
选举流程如下：
节点初始状态均为 Follower，Follower 只被动接收请求，如果 ElectionTime 到期时仍未收到 Leader 的 AppendEntry RPC，Follower 认为当前没有 Leader，转为 Candidate。
Candidate 在集群中广播 RequestVote RPC，尝试竞选 Leader，其他节点收到后首先判断是否同意本次选举，并将结果返回给 Candidate。如果 Candidate 收到大多数节点的同意响应，转为 Leader。
Leader 接收客户端请求，将其转为 Entry 追加到日志文件，同时通过 AppendEntry RPC 同步日志 Entry 给其他节点。
选举超时值：
在选举时可能会出现两个节点的选举定时器同时到期并发起选举，各自得到一半选票导致选举失败，选举失败意味着系统没有 Leader，不可服务。如果选举定时器是定值，很可能两者再次同时到期。为了降低冲突的概率，选举超时值采用随机值的方式。此外，选举超时值如果过大会导致 Leader 故障会很久才会再次选举。选举超时值通常取 300ms~600ms 之间的随机值。
2.3 日志同步选举阶段完成后，Leader 节点开始接收客户端请求，将请求封装成 Entry 追加到 raft 日志文件末尾，之后同步 Entry 到其他 Follower 节点。当大多数节点写入成功后，该 Entry 被标记为 committed，raft 算法保证了 committed 的 Entry 一定不会再被修改。
日志同步具体流程：
1）Leader 上为每个节点维护 NextIndex、MatchIndex，NextIndex 表示待发往该节点的 Entry index，MatchIndex 表示该节点已匹配的 Entry index，同时每个节点维护 CommitIndex 表示当前已提交的 Entry index。转为 Leader 后会将所有节点的 NextIndex 置为自己最后一条日志 index+1，MatchIndex 全置 0，同时将自身 CommitIndex 置 0。
2）Leader 节点不断将 user_data 转为 Entry 追加到日志文件末尾，Entry 包含 index、term 和 user_data，其中 index 在日志文件中从 1 开始顺序分配，term 为 Leader 当前的 term。
3）Leader 通过 AppendEntry RPC 将 Entry 同步到 Followers，Follower 收到后校验该 Entry 之前的日志是否已匹配。如匹配则直接写入 Entry，返回成功；否则删除不匹配的日志，返回失败。校验是通过在 AppendEntry RPC 中携带待写入 Entry 的前一条 entry 信息完成。
4）当 Follower 返回成功时，更新对应节点的 NextIndex 和 MatchIndex，继续发送后续的 Entry。如果 MatchIndex 更新后，大多数节点的 MatchIndex 已大于 CommitIndex，则更新 CommitIndex。Follower 返回失败时回退 NextIndex 继续发送，直到 Follower 返回成功。
5）Leader 每次 AppendEntry RPC 中会携带当前最新的 LeaderCommitIndex，Follower 写入成功时会将自身 CommitIndex 更新为 Min(LastLogIndex,LeaderCommitIndex)。
同步过程中每次日志的写入均需刷盘以保证宕机时数据不丢失。
日志冲突：
在日志同步的过程中，可能会出现节点之间日志不一致的问题。例如 Follower 写日志过慢、Leader 切换导致旧 Leader 上未提交的脏数据等场景下都会发生。在 Raft 算法中，日志冲突时以 Leader 的日志为准，Follower 删除不匹配部分。
如下图所示，Follower 节点与 Leader 节点的日志都存在不一致问题，其中 (a)、(b) 节点日志不全，(c)、(d)、(e)、(f) 有冲突日志。Leader 首先从 index=11(最后一条 Entry index +1) 开始发送 AppendEntry RPC,Follower 均返回不匹配，Leader 收到后不断回退。(a)、(b) 在找到第一条匹配的日志后正常同步，(c)、(d)、(e)、(f) 在这个过程中会逐步删除不一致的日志，最终所有节点的日志都与 Leader 一致。成为 Leader 节点后不会修改和删除已存在的日志，只会追加新的日志。
Raft 算法中充分考虑了工程化中集群管理问题，支持动态的添加节点到集群，剔除故障节点等。下面详细描述添加和删除节点流程。
添加节点：
如下图所示，集群中包含 A B C，A 为 Leader，现在添加节点 D。
1） 清空 D 节点上的所有数据，避免有脏数据。
2） Leader 将存量的日志通过 AppendEntry RPC 同步到 D，使 D 的数据跟上其他节点。
3） 待 D 的日志追上后，Leader A 创建一条 Config Entry，其中集群信息包含 ABCD。
4） Leader A 将 Config Entry 同步给 B C D，Follower 收到后应用，之后所有节点的集群信息都变为 ABCD，添加完成。
注：在步骤 2 过程中,Leader 仍在不断接收客户请求生成 Entry，所以只要 D 与 A 日志相差不大即认为 D 已追上。
删除节点：
如下图所示，集群中原来包含 A B C D，A 为 Leader，现在剔除节点 D。
1) Leader A 创建一条 Config Entry，其中集群信息为 ABC。
2) A 将日志通过 AppendEntry RPC 同步给节点 B C。
3) A B C 在应用该日志后集群信息变为 ABC，A 不再发送 AppendEntry 给 D，D 从集群中移除。
4) 此时 D 的集群信息依旧为 ABCD，在选举超时到期后，发起选举，为了防止 D 的干扰，引入额外机制：所有节点在正常接收 Leader 的 AppendEntry 时，拒绝其他节点发来的选举请求。
5) 将 D 的数据清空并下线。
2.5 快照管理在节点重启时，由于无法得知 State Matchine 当前 ApplyIndex(除非每次应用完日志都持久化 ApplyIndex，还要保证是原子操作，代价较大)，所以必须清空 State Matchine 的数据，将 ApplyIndex 置为 0,，从头开始应用日志，代价太大，可以通过定期创建快照的方式解决该问题。如下图所示：
1） 在应用完 Entry 5 后，将当前 State Matchine 的数据连同 Entry 信息写入快照文件。
2） 如果节点重启，首先从快照文件中恢复 State Matchine，等价于应用了截止到 Entry 5 为止的所有 Entry，但效率明显提高。
3） 将 ApplyIndex 置为 5，之后从 Entry 6 继续应用日志，数据和重启前一致。
2.6 异常场景及处理Raft 具有很强的容错性，只要大多数节点正常互联，即可保证系统的一致性和可用性，下面是一些常见的异常情况，以及他们的影响及处理：
可以看到异常情况对系统的影响很小，即使是 Leader 故障也可以在极短的时间内恢复，任何情况下系统都一直保持强一致性，为此牺牲了部分可用性（大多数节点故障时，概率极低）。不过，Leader 故障时新的 Leader 可能会包含旧 Leader 未提交或已提交但尚未通知客户端的日志。由于算法规定成为 Leader 后不允许删除日志，所以这部分日志会被新 Leader 同步并提交，但由于连接信息丢失，客户端无法得知该情况。当发起重试后会出现重复数据，需要有幂等性保证。此外，raft 的核心算法都是围绕 Leader 展开，网络分区时可能出现伪 Leader 问题，也需要特殊考虑。
三 Raft 在 CMQ 中的应用和性能优化3.1 Raft 算法在 CMQ 中的应用我们用 State Matchine 统一表示业务模块，其通过 ApplyIndex 维护已应用的日志 index。以下为 Raft 与状态机交互的流程：
1）客户端请求发往 Leader 节点。
2）Leader 节点的 Raft 模块将请求转为 Entry 并同步到 Followers。
3）大多数节点写入成功后 Raft 模块更新 CommitIndex。
4）各节点的 State Machine 顺序读取 ApplyIndex+1 到 CimmitIndex 之间的 Entry，取出其中的 user_data 并应用，完成后更新 ApplyIndex。
5）Leader 上的 State Machine 通知客户端操作成功。
6）如此循环。
下面介绍 CMQ 详细的生产消费流程：
生产流程：
1）生产者将生产消息的请求发往 Leader 的 Raft 模块。
2）Raft 模块完成 Entry 的创建和同步。
3）大多数节点上持久化并返回成功后 Entry 标记为 Committed。
4）所有节点的 State Machine 应用该日志，取出实际的生产请求，将消息内容写入磁盘，更新 ApplyIndex。该步骤不需要刷盘。
5）Leader 回复客户端 Confirm，通知生产成功。
6）如果此后机器重启，通过 raft 日志恢复生产消息，保证了已 Confirm 的消息不丢失。
消费流程：
1）消费者从 Leader 节点拉取消息。
2）Leader 收到后从磁盘加载未删除的消息投递给客户端。
3）客户端处理完成后 Ack 消息，通知服务器删除消息。
4）Ack 请求经 Raft 同步后标记为 Committed。
5）各节点状态机应用该日志，将消息对应的 bit 置位，将其设置为已删除并更新 ApplyIndex。
6）通知客户端删除成功。
7）如果机器重启，通过 Raft 日志恢复 Ack 请求，保证了已删除的消息不会再投递。
快照管理：
快照管理与业务紧密相关，不同系统快照制作的成本差异很大，CMQ 中快照的内容十分轻量，一次快照的耗时在毫秒级，平均 5min 创建一次，各节点独立完成。实现上内存中维护了一份动态的快照，制作快照时首先拷贝出动态快照的副本，之后处理流继续更新动态快照，用拷贝出的副本创建快照文件，不影响实际的处理流。快照具体内容包括：
1）term：快照对应 Entry 的 term （参照算法）
2）index：快照对应 Entry 的 index （参照算法）
3）node_info：Entry 时的集群配置信息。
4）topic info：每个队列一项。CMQ 中同一队列生产的消息顺序写入，分片存储，因此只需记录最后一个分片的状态（分片文件名，文件偏移量）。
5）queue info：每个队列一项。CMQ 中采用 bitmap 记录消息的删除情况，在内存中维护，在制作快照时 dump 到快照文件。
Raft 算法的性能瓶颈主要有两方面：
1）每次日志写入后都需要刷盘才能返回成功，而刷盘是一个比较耗时的操作。
2）由于算法限制，所有的请求都由 Leader 处理，不能做到所有节点皆可提供服务。
针对以上两个问题，我们做了以下优化：
Batch Processing：在请求量较大时，并不是每一条日志写入都刷盘，还是累积一定量的日志后集中刷盘，从而减少刷盘次数。对应的，在同步到 Follower 时也采用批量同步的方式，Follower 接收后将日志批量写盘。
Multi-Raft： 进程中同时运行多个 raft 实例，机器之间组建多 raft 组，客户端请求路由到不同的 group 上，从而实现多主读写，提高并发性能。通过将 leader 分布在不同机器上，提高了系统的整体利用率。
Async-rpc： 在日志同步过程中采用同步 rpc 方式，在一端处理时另一端只能等待，性能较差。我们采用异步的方式使得 leader 端发送和 Follower 端处理并发进行。发送过程中 leader 端维持一个发送窗口，当待确认的 rpc 数达到上限停止发送，窗口值上限：
在与同属于高可靠 (多副本同步刷盘) 的 Rabbitmq 性能对比中，相同压测场景下 CMQ 速度可以达到 RabbitMQ 的四倍左右。
以下为在 E5-2620*2/8G*8/2T*12/SSD-80G*1/10GE*2 配置机型测试 1KB 消息大小时性能数据：
测试中 CMQ 采用单 Raft 组方式以保证测试公平性。监控显示 CPU、内存和网卡均未达到瓶颈，系统瓶颈在磁盘 IO，iostat 显示 w_await 远大于 svctm。主要原因在于刷盘耗时，造成写操作排队等待。
实际生产环境 CMQ 中我们将 raft 组和磁盘进行绑定，实现 raft 组之间磁盘的隔离，一方面保证了磁盘的顺序读写，另一方面充分利用机器的 cpu 、内存、网卡等资源。
四 总结Raft 算法具备强一致、高可靠、高可用等优点， 消息中间件通常分为高可靠版本和高性能版本两种。腾讯云 CMQ 是一款金融级的高可靠分布式消息中间件，通过 raft 保证了消息的可靠不丢失。同时在性能和可用性方面相比竞品都有显著提高。
