当前位置：首页 > news >正文

C++实现Raft算法

news 来源：原创 2025/7/27 16:37:06

概念部分

Raft 算法是一种用于实现分布式系统中的一致性的算法。它是为了容易理解而设计的，其目标是实现和 Paxos 算法相同的功能，但更加容易理解和实现。Raft 算法在分布式系统中尤其关键，因为它帮助系统中的多个节点就其数据的准确状态达成一致。

Raft 算法主要通过以下几个关键的概念和步骤来实现一致性：

领导者选举（Leader Election）:
- 在 Raft 中，任何时候都有一个明确的领导者节点，它负责处理所有的客户端请求（读写操作）。
- 当现有的领导者失效或网络分割导致领导者与其他节点失去联系时，Raft 会自动触发新的领导者选举。
日志复制（Log Replication）:
- 领导者节点负责将客户端的操作作为日志条目复制到集群中的其他节点（称为跟随者）。
- 只有当大多数节点都已经存储了日志条目，这个条目才会被提交，从而确保了数据的一致性和持久性。
安全性:
- Raft 确保所有的已提交日志条目在持久化后对于所有客户端都是一致的。
- 即使在领导者变更或网络故障的情况下，已经被提交的日志也不能被覆盖。
心跳机制（Heartbeats）:
- 领导者定期向所有跟随者发送心跳信号（空的日志条目），以维持其领导地位和防止新的领导者选举。

Q：领导者如何选举？

A：在Raft算法中，每个节点开始都是Follower状态。如果Follower在设定的超时时间内没有收到Leader的心跳（通常包括日志条目的复制请求），它会变成Candidate并发起新的选举。Candidate向其他节点请求投票，如果获得了大多数节点的支持，就会成为新的Leader。Leader负责接收客户端的请求并将请求作为日志条目发送给Follower们，以此维护集群的一致性。

Q：节点之间通过网络通信，其他节点（follower）如何知道leader出现故障？

A：leader会定时向集群中剩下的节点（follower）发送AppendEntry（作为心跳，hearbeat ）以通知自己仍然存活。可以推知，如果follower在一段时间内没有接收leader发送的AppendEntry，那么follower就会认为当前的leader 出现故障，从而发起选举。这里 “follower在一段时间内没有接收leader发送的AppendEntry”，在实现上可以用一个定时器和一个标志位来实现，每到定时时间检查这期间有无AppendEntry 即可。

Q：follower知道leader出现故障后如何选举出leader？

当follower知道leader出故障后，只能通过term增加，变成candidate，向其他节点发起RequestVoteRPC申请其他follower的选票，过一段时间后会发生如下情况，赢得选举，马上成为leader，此时的term已经增加了，其他follower发现有符合要求的leader之后，自己马上变成follower，这个符合要求包括leader的term>=自己的term；如果一轮选举之后无人变成leader，那么就会循环这个过程。

Q：符合什么条件的节点可以成为leader？

A：至于什么样的节点可以成为leader，我们要先明确选举leader的目的，目的是为了选举出的leader一定包含了整个集群中目前已committed的所有日志，当candidate发送RequestVoteRPC时，会带上最后一个entry的信息。所有的节点收到该请求后，都会对比自己的日志，如果发现自己的日志更新，就会拒绝投票给该candidate，即自己的日志必须要"不旧于"该candidate。判断日志老旧的方法，需要比较两个东西，最新日志entry的term和对应的index，如果两个节点最新日志entry的term不同，则term大的日志更新，如果两个节点最新日志entry的term相同，那么最新日志entry的index大的更新。这样的限制可以保证，成为leader的节点，其日志已经是多数节点中最完备的，即包含了整个集群的所有committed entries

Q：AppendEntry 有什么作用？

A：具体来说有两种主要的作用和一个附带的作用：

主要作用是协调和携带日志entry及其辅助信息，以控制日志的同步和日志向状态机提交。附带作用是通告leader的index和term等关键信息以便follower对比确认follower自己或者leader是否过期。

Raft 通过以上机制确保了系统的高可用性和一致性，同时也相对易于实现和理解。这种算法通常用于各种分布式数据存储和服务系统中，如分布式数据库、日志服务和配置管理系统等。

相比较Paxos，Raft算法通过拆解共识过程，引入Leader election等机制简化了公式过程，因此Raft算法已经是方便入门的共识算法了。

Q：什么是共识过程？

A：共识是容错分布式系统中的一个基本问题。共识涉及多个服务器对状态机状态（对本项目而言就是上层的k-v数据库）达成一致。一旦他们对状态机状态做出决定，这个决定就是最终决定（已经被集群共识的值可以保证后面不会被覆盖，Raft的安全性）。

共识用于确保系统中的多个节点即使在部分节点失败的情况下也能对某个值或状态达成一致意见。共识算法主要解决的问题是如何在没有中心协调者的情况下，通过网络中的各个节点之间的通信来达到整个系统的数据一致性。

实际使用系统中的共识算法一般满足以下特性：

在非拜占庭条件下保证共识的一致性。非拜占庭条件就是可信的网络条件，即与你通信的节点的信息都是真实的，不存在欺骗。
在多数节点存活时，保持可用性。“多数”永远指的是配置文件中所有节点的多数，而不是存活节点的多数。多数等同于超过半数的节点，多数这个概念概念很重要，贯穿Raft算法的多个步骤。
不依赖于绝对时间。理解这点要明白共识算法是要应对节点出现故障的情况，在这样的环境中网络报文也很可能会受到干扰从而延迟，如果完全依靠于绝对时间，会带来问题，Raft用自定的Term（任期）作为逻辑时钟来代替绝对时间。
在多数节点一致后就返回结果，而不会受到个别慢节点的影响。这点与第二点联合理解，只要“大多数节点同意该操作”就代表整个集群同意该操作。对于raft来说，”操作“是储存到日志log中，一个操作就是log中的一个entry。

Q：raft的Term是怎么工作的？

A：

Raft将Term作为内部的逻辑时钟，使用Term的对比来比较日志、身份、心跳的新旧而不是用绝对时间。Term与Leader的身份绑定，即某个节点是Leader更严谨一点的说法是集群某个Term的Leader。Term用连续的数字进行表示。Term会在follower发起选举（成为Candidate从而试图成为Leader ）的时候加1，对于一次选举可能存在两种结果：

1.胜利当选：胜利的条件是超过半数的节点认为当前Candidate有资格成为Leader，即超过半数的节点给当前Candidate投了选票。

2.失败：如果没有任何Candidate（一个Term的Leader只有一位，但是如果多个节点同时发起选举，那么某个Term的Candidate可能有多位）获得超半数的选票，那么选举超时之后又会开始另一个Term（Term递增）的选举。

Q：Raft是如何保证一个Term只有一个Leader的？

A：因为Candidate变成Leader的条件是获得超过半数选票，一个节点在一个Term内只有一个选票（投给了一个节点就不能再投递给另一个节点），因此不可能有两个节点同时获得超过半数的选票。

发生故障时，一个节点无法知道当前最新的Term是多少，在故障恢复后，节点就可以通过其他节点发送过来的心跳中的Term信息查明一些过期信息。

当发现自己的Term小于其他节点的Term时，这意味着“自己已经过期”，不同身份的节点的处理方式有所不同：

leader、Candidate：退回follower并更新term到较大的那个Term
follower：更新Term信息到较大的那个Term

这里解释一下为什么自己的Term小于其他节点的Term时leader、Candidate会退回follower 而不是延续身份，因为通过Term信息知道自己过期，意味着自己可能发生了网络隔离等故障，那么在此期间整个Raft集群可能已经有了新的leader、提交了新的日志，此时自己的日志是有缺失的，如果不退回follower，那么可能会导致整个集群的日志缺失，不符合安全性。

相反，如果发现自己的Term大于其他节点的Term，那么就会忽略这个消息中携带的其他信息。

安全性：

Election Safety ：每个 term 最多只会有一个 leader；集群同时最多只会有一个可以读写的 leader。

每个Term最多只会有一个leader在前面已经解释过；集群同时最多只会有一个可以读写的 leade 是指集群中由于发生了网络分区，一个分区中的leader会维护分区的运行，而另一个分区会因为没有leader而发生选举产生新的leader，任何情况下，最多只有一个分区拥有绝大部分节点，那么只有一个分区能写入日志，这个在[共识算法要满足的性质]一节中已经介绍过。

Leader Append-Only ：leader 的日志是只增的。
Log Matching ：如果两个节点的日志中有两个 entry 有相同的 index 和 term，那么它们就是相同的 entry。

这是因为在Raft中，每个新的日志条目都必须按照顺序追加到日志中。

在运行过程中会根据这条性质来检查follower的日志与leader的日志是否匹配，如果不匹配的话leader会发送自己的日志去覆盖follower对应不匹配的日志。

Leader Completeness ：一旦一个操作被提交了，那么在之后的 term 中，该操作都会存在于日志中。
State Machine Safety ：状态机一致性，一旦一个节点应用了某个 index 的 entry 到状态机，那么其他所有节点应用的该 index 的操作都是一致的。

在RPC中日志同步和心跳是放在一个RPC函数（AppendEntryRPC）中来实现的，原因为：

对于一个follower，如果leader认为其日志已经和自己匹配了，那么在AppendEntryRPC中不用携带日志（再携带日志属于无效信息了，但其他信息依然要携带），反之如果follower的日志只有部分匹配，那么就需要在AppendEntryRPC中携带对应的日志。心跳RPC 可以看成是没有携带日志的特殊的日志同步RPC。

Q：为什么不直接让follower拷贝leader的日志或者leader发送全部的日志给follower？

A：leader发送日志的目的是让follower同步自己的日志，当然可以让leader发送自己全部的日志给follower，然后follower接收后就覆盖自己原有的日志，但是这样就会携带大量的无效的日志（因为这些日志follower本身就有）。

因此 raft的方式是：先找到日志不匹配的那个点，然后只同步那个点之后的日志。

Q：leader如何知道follower的日志是否与自己完全匹配？

A：在AppendEntryRPC中携带上 entry的index和对应的term（日志的term），可以通过比较最后一个日志的index和term来得出某个follower日志是否匹配。

Q：如果发现不匹配，那么如何知道哪部分日志是匹配的，哪部分日志是不匹配的？

A：leader每次发送AppendEntryRPC后，follower都会根据其entry的index和对应的term来判断某一个日志是否匹配。

在leader刚当选，会从最后一个日志开始判断是否匹配，如果匹配，那么后续发送AppendEntryRPC就不需要携带日志entry了。

如果不匹配，那么下一次就发送倒数第2个日志entry的index和其对应的term来判断匹配，

如果还不匹配，那么依旧重复这个过程，即发送倒数第3个日志entry的相关信息

重复这个过程，知道遇到一个匹配的日志。

raft日志的两个特点：

两个节点的日志中，有两个 entry 拥有相同的 index 和 term，那么它们一定记录了相同的内容/操作，即两个日志匹配
两个节点的日志中，有两个 entry 拥有相同的 index 和 term，那么它们前面的日志entry也相同

Q：如何保证这两点？

A：

保证第一点：仅有 leader 可以生成 entry
保证第二点：leader 在通过 AppendEntriesRPC 和 follower 通讯时，除了带上自己的term等信息外，还会带上entry的index和对应的term等信息，follower在接收到后通过对比就可以知道自己与leader的日志是否匹配，不匹配则拒绝请求。leader发现follower拒绝后就知道entry不匹配，那么下一次就会尝试匹配前一个entry，直到遇到一个entry匹配，并将不匹配的entry给删除（覆盖）。

raft为了避免出现一致性问题，要求 leader 绝不会提交过去的 term 的 entry （即使该 entry 已经被复制到了多数节点上）。leader 永远只提交当前 term 的 entry，过去的 entry 只会随着当前的 entry 被一并提交。

代码部分

demo演示

先看一个小demo，这个demo会实现Raft算法，尤其是选举过程，需要涉及多线程管理、网络通信(模拟)，随机超时等机制。

我们将创建一个 Raft 节点类，用于管理状态转换（如 Follower、Candidate、Leader），处理选举过程，发送和接收消息等。为了简化，我们将跳过复杂的分布式网络部分，用本地线程和模拟函数来代替网络通信。

1. 环境准备与头文件引入

#include <iostream>
#include <vector>
#include <thread>
#include <mutex>
#include <atomic>
#include <chrono>
#include <random>
#include <condition_variable>
#include <functional>
#include <algorithm>using namespace std;// 节点状态
enum class ServerState {Follower,Candidate,Leader
};

2. 定义RaftServer类

RaftServer 类包含节点状态、当前任期、已投票对象等成员变量，以及各种控制服务器状态的方法。

class RaftServer {
public:RaftServer(int id, int totalServers);void run(); // 启动服务器的主逻辑void electionTimeoutCheck(); // 检查选举超时void startElection(); // 发起选举void requestVotes(); // 向其他节点请求投票void receiveVote(int term, bool voteGranted); // 处理投票响应void becomeLeader(); // 成为 Leaderprivate:int id; // 服务器 IDatomic<ServerState> state{ServerState::Follower}; // 节点状态atomic<int> currentTerm{0}; // 当前任期atomic<int> votedFor{-1}; // -1 表示未投票atomic<int> votesReceived{0}; // 已收到的选票数mutex mtx; // 互斥锁condition_variable cv; // 条件变量，用于实现选举超时的等待vector<thread> serverThreads; // 用于模拟不同节点的线程int totalServers; // 总共的服务器数量chrono::time_point<chrono::steady_clock> lastHeartbeat; // 最后一次收到心跳的时间
};

解释：

1. atomic类型

atomic类型是为了确保多线程程序中对单个变量的操作是原子的，即这些操作在并发环境下是安全的，不会发生数据竞争。例如atomic<int> currentTerm{0}，这行代码声明了一个名为currentTerm的原子整数，并使用初始化列表{}将其初始化为0。在C++11后可以使用花括号进行列表初始化，这是一种更安全的初始化方式，能够减少某些类型转换的错误。

2. 枚举类 enum class

enum class是C++11引入的一种强类型枚举，提供了更好的类型安全性和作用域控制：以上代码定义了一个名为ServerState的枚举类，并声明了一个原子类型的state变量，初始状态设置为ServerState::Follower。

3. mutex

mutex(互斥量)是C++中用来管理多线程访问共享数据的同步机制的数据类型。属于C++标准库<mutex>头文件下的一个类。使用mutex可以防止所谓的竞态条件，即多个线程同时修改同一数据的情况。mutex mtx，mtx.clock()锁上，mtx.unclock()解锁

4. chrono::time_point

chrono::time_point是一个与时间相关的类模板，用于表示一个具体的时间点。它是C++11标准库<chrono>中时间库的一部分，该库提供了一系列与时间测量和持续时间计算相关的工具。而代码中的chrono::time_point<chrono::steady_clock> lastHeartbeat;，这里使用了 chrono::steady_clock 作为 chrono::time_point 的时钟类型。下面分别解释这些组成部分的含义和用途：

chrono::steady_clock 是一种提供“单调时间”的时钟，意味着它的时间从来不会倒退（即使系统时间被用户或同步协议更改）。
这个时钟主要用于测量时间间隔，适合用于计时器和延迟测量。
它保证从程序启动到程序结束，时间始终向前移动。
chrono::time_point 是一个时间点表示，用于保存某个时钟的一个具体时间点。
它是模板化的，可以与任何 chrono 中定义的时钟类型一起使用，比如 system_clock、steady_clock 或 high_resolution_clock。
时间点可以用于计算两个事件的时间差或确定特定事件发生的时间。

示例用法：

假设你在应用程序中需要测量某个操作的执行时间，或者检查自上一次心跳信号以来过了多久，可以这样使用：

#include <iostream>
#include <chrono>
#include <thread>using namespace std;int main() {// 记录开始时间auto start = chrono::steady_clock::now();// 模拟耗时操作this_thread::sleep_for(chrono::seconds(2));// 记录结束时间auto end = chrono::steady_clock::now();// 计算持续时间auto duration = chrono::duration_cast<chrono::seconds>(end - start);cout << "Operation took " << duration.count() << " seconds." << endl;
}

3. 构造函数和选举超时逻辑

构造函数初始化服务器的属性，并启动主逻辑。选举超时逻辑通过条件变量来实现。

RaftServer::RaftServer(int id, int totalServers): id(id), totalServers(totalServers) {lastHeartbeat = chrono::steady_clock::now();
}void RaftServer::electionTimeoutCheck() {while (true) {unique_lock<mutex> lock(mtx);// 使用随机的选举超时时间，150 到 300 毫秒之间auto randomTimeout = chrono::milliseconds(rand() % 150 + 150);cv.wait_for(lock, randomTimeout);// 如果选举超时，且节点仍是 Follower，就转变为 Candidate 开始选举if (chrono::steady_clock::now() - lastHeartbeat >= randomTimeout) {cout << "Node " << id << " election timeout. Starting election." << endl;startElection();}}
}

4. 发起选举

当选举超时时触发选举时，节点变为Candidate，并请求其他节点投票

void RaftServer::startElection() {lock_guard<mutex> lock(mtx);state = ServerState::Candidate;currentTerm++;votedFor = id; // 自己给自己投票votesReceived = 1; // 自己的票cout << "Node " << id << " started election for term " << currentTerm << "." << endl;requestVotes();
}void RaftServer::requestVotes() {for (int i = 0; i < totalServers; ++i) {if (i != id) {// 模拟向其他节点请求投票thread([this, i]() {this_thread::sleep_for(chrono::milliseconds(50)); // 模拟网络延迟// 假设其他节点都同意投票（简单起见）receiveVote(currentTerm, true);}).detach();}}
}

5. 处理投票响应

处理投票响应，如果获得大多数节点的支持，则成为 Leader。

void RaftServer::receiveVote(int term, bool voteGranted) {lock_guard<mutex> lock(mtx);if (term == currentTerm && voteGranted) {votesReceived++;if (votesReceived > totalServers / 2) {becomeLeader();}}
}void RaftServer::becomeLeader() {state = ServerState::Leader;cout << "Node " << id << " became Leader for term " << currentTerm << "." << endl;// 成为 Leader 后，定期发送心跳while (state == ServerState::Leader) {this_thread::sleep_for(chrono::milliseconds(100));cout << "Node " << id << " sends heartbeat to followers." << endl;}
}

6. 启动服务器的主逻辑

主逻辑运行时，根据状态执行不同的操作：

Follower：等待心跳并检查选举超时。
Candidate：发起选举。
Leader：发送心跳。

void RaftServer::run() {thread([this]() { electionTimeoutCheck(); }).detach(); // 启动选举超时检查线程while (true) {this_thread::sleep_for(chrono::milliseconds(100)); // 每 100 毫秒检查一次状态if (state == ServerState::Leader) {// 作为 Leader，不断发送心跳cout << "Node " << id << " is sending heartbeats." << endl;}}
}

7. 主函数启动多个节点

int main() {srand(time(0)); // 用于随机超时const int totalServers = 5; // 集群中总共 5 个节点vector<RaftServer> servers;// 创建并启动每个节点for (int i = 0; i < totalServers; ++i) {servers.emplace_back(i, totalServers);}vector<thread> threads;for (int i = 0; i < totalServers; ++i) {threads.emplace_back(&RaftServer::run, &servers[i]);}for (auto& t : threads) {t.join();}return 0;
}

解释

节点状态管理：每个节点有三种状态：Follower、Candidate、Leader。通过状态之间的转换来实现选举、心跳维护。
选举触发：Follower 会在选举超时时变为 Candidate 并发起选举，这里使用随机的超时时间防止出现 "分裂投票" 的问题。
投票请求和处理：Candidate 会向其他节点请求投票，收到足够多的投票后成为 Leader。
成为 Leader 后的工作：Leader 不断向其他节点发送心跳，保持领导地位。