好的，没问题！我们现在就开始进入 Paxos 和 Raft 的奇妙世界，看看如何用 C++ 实现分布式系统的一致性。准备好了吗？让我们开始吧！
大家好！今天我们要聊聊分布式系统中的一个非常重要的概念——一致性，以及实现一致性的两个著名算法：Paxos 和 Raft。它们就像分布式系统的“大脑”，确保集群中的所有节点对同一个事情达成共识，避免出现“各说各话”的混乱局面。
为什么需要一致性？
想象一下，你有一个银行账户，存在一个分布式数据库中。如果数据库中的不同节点对你的账户余额有不同的记录，那会发生什么？你取钱的时候，一个节点说你有 1000 元，另一个节点说你有 10 元，银行岂不是要破产了？
这就是一致性的重要性。它确保分布式系统中的数据是可靠、一致的，即使在出现故障的情况下也能正常工作。
Paxos：一致性算法的“祖师爷”
Paxos 是 Leslie Lamport 在 1990 年代提出的，被认为是“一致性算法之母”。它的理论非常精妙，但理解起来也比较困难，被戏称为“最难理解的算法之一”。
Paxos 的核心思想是：通过多轮投票和协商，让所有节点对一个值达成共识。
Paxos 的角色
在 Paxos 中，有三个角色：
- Proposer（提议者）： 提出一个提议值，希望被大多数节点接受。
- Acceptor（接受者）： 接收提议者的提议，并决定是否接受。
- Learner（学习者）： 从 Acceptor 那里学习最终被选定的值。
Paxos 的基本流程
Paxos 的基本流程分为两个阶段：
-
Prepare 阶段：
- Proposer 选择一个提案编号（Proposal ID），这个编号必须是全局唯一的，并且要大于之前所有 Proposer 提出的编号。
- Proposer 向所有 Acceptor 发送 Prepare 请求，请求中包含提案编号。
- Acceptor 收到 Prepare 请求后，如果提案编号大于自己已经接受过的所有提案编号，则回复一个 Prepare 响应，其中包含自己已经接受过的最高编号的提案（如果之前没有接受过任何提案，则返回空）。
-
Accept 阶段：
- 如果 Proposer 收到了大多数 Acceptor 的 Prepare 响应，并且每个响应都表明 Acceptor 没有接受过更高编号的提案，或者响应中包含了之前接受过的最高编号的提案，那么 Proposer 就选择一个提案值。
- 如果 Acceptor 之前没有接受过任何提案，或者响应中包含了之前接受过的最高编号的提案，那么 Proposer 就选择 Acceptor 之前接受过的最高编号的提案的值。否则，Proposer 可以选择自己提出的值。
- Proposer 向所有 Acceptor 发送 Accept 请求，请求中包含提案编号和提案值。
- Acceptor 收到 Accept 请求后，如果提案编号大于自己已经接受过的所有提案编号，则接受这个提案，并回复一个 Accept 响应。
-
Learn 阶段:
- Learner 从 Acceptor 那里学习最终被选定的值。当大多数 Acceptor 接受了同一个提案后，Learner 就认为这个值已经被选定。
Paxos 的 C++ 实现 (简化版)
由于 Paxos 算法比较复杂，这里我们提供一个简化版的 C++ 实现，主要演示 Prepare 和 Accept 阶段。
#include <iostream>
#include <vector>
#include <algorithm>
#include <mutex>
using namespace std;
// 提案
struct Proposal {
int id;
int value;
};
// Acceptor
class Acceptor {
public:
Acceptor(int id) : acceptorId(id), promisedProposalId(0), acceptedProposal({0, 0}) {}
// 处理 Prepare 请求
bool handlePrepare(int proposalId) {
lock_guard<mutex> lock(mtx);
if (proposalId > promisedProposalId) {
promisedProposalId = proposalId;
cout << "Acceptor " << acceptorId << ": Promised proposal " << proposalId << endl;
return true;
}
cout << "Acceptor " << acceptorId << ": Rejected proposal " << proposalId << " (already promised " << promisedProposalId << ")" << endl;
return false;
}
// 处理 Accept 请求
bool handleAccept(Proposal proposal) {
lock_guard<mutex> lock(mtx);
if (proposal.id >= promisedProposalId) {
promisedProposalId = proposal.id;
acceptedProposal = proposal;
cout << "Acceptor " << acceptorId << ": Accepted proposal " << proposal.id << " with value " << proposal.value << endl;
return true;
}
cout << "Acceptor " << acceptorId << ": Rejected accept proposal " << proposal.id << " (already promised " << promisedProposalId << ")" << endl;
return false;
}
Proposal getAcceptedProposal() {
lock_guard<mutex> lock(mtx);
return acceptedProposal;
}
private:
int acceptorId;
int promisedProposalId; // 承诺的提案 ID
Proposal acceptedProposal; // 接受的提案
mutex mtx;
};
// Proposer
class Proposer {
public:
Proposer(int id, vector<Acceptor*>& acceptors) : proposerId(id), acceptors(acceptors), nextProposalId(1) {}
// 提出提案
bool propose(int value) {
int proposalId = generateProposalId();
cout << "Proposer " << proposerId << ": Proposing value " << value << " with proposal ID " << proposalId << endl;
// Prepare 阶段
int prepareSuccessCount = 0;
for (Acceptor* acceptor : acceptors) {
if (acceptor->handlePrepare(proposalId)) {
prepareSuccessCount++;
}
}
if (prepareSuccessCount < acceptors.size() / 2 + 1) {
cout << "Proposer " << proposerId << ": Prepare phase failed." << endl;
return false;
}
// Accept 阶段
Proposal proposal = {proposalId, value};
int acceptSuccessCount = 0;
for (Acceptor* acceptor : acceptors) {
if (acceptor->handleAccept(proposal)) {
acceptSuccessCount++;
}
}
if (acceptSuccessCount < acceptors.size() / 2 + 1) {
cout << "Proposer " << proposerId << ": Accept phase failed." << endl;
return false;
}
cout << "Proposer " << proposerId << ": Proposal " << proposalId << " with value " << value << " accepted." << endl;
return true;
}
private:
int proposerId;
vector<Acceptor*> acceptors;
int nextProposalId;
int generateProposalId() {
return nextProposalId++;
}
};
int main() {
// 创建 Acceptor 节点
vector<Acceptor*> acceptors;
acceptors.push_back(new Acceptor(1));
acceptors.push_back(new Acceptor(2));
acceptors.push_back(new Acceptor(3));
// 创建 Proposer 节点
Proposer proposer(1, acceptors);
// 提出一个值
proposer.propose(100);
// 查看 Acceptor 接受的值
for (Acceptor* acceptor : acceptors) {
Proposal acceptedProposal = acceptor->getAcceptedProposal();
cout << "Acceptor " << acceptor->getId() << " accepted proposal ID: " << acceptedProposal.id << ", value: " << acceptedProposal.value << endl;
}
// 清理内存
for (Acceptor* acceptor : acceptors) {
delete acceptor;
}
return 0;
}注意： 这只是一个非常简化的 Paxos 实现，没有考虑很多实际情况，比如网络故障、节点宕机等。在实际应用中，需要使用更复杂的 Paxos 变种，比如 Multi-Paxos。
Raft：更易理解的一致性算法
Raft 算法由 Diego Ongaro 和 John Ousterhout 在 2014 年提出，旨在解决 Paxos 难以理解的问题。Raft 的设计目标是“易于理解”，它将一致性问题分解成几个相对独立的子问题，使得算法更容易理解和实现。
Raft 的角色
在 Raft 中，有三种角色：
- Leader（领导者）： 负责接收客户端的请求，并将日志复制到其他节点。
- Follower（跟随者）： 被动地接受 Leader 的日志复制，并响应 Leader 的心跳。
- Candidate（候选者）： 在 Leader 宕机后，发起选举，尝试成为新的 Leader。
Raft 的基本流程
Raft 的基本流程包括：
-
Leader Election（领导者选举）：
- 当 Follower 在一段时间内没有收到 Leader 的心跳时，就会变成 Candidate。
- Candidate 会增加自己的任期号（Term），并向所有其他节点发送 RequestVote 请求，请求其他节点投票给自己。
- 如果 Candidate 收到大多数节点的投票，就成为新的 Leader。
- 每个节点在一个任期内只能投一票。
-
Log Replication（日志复制）：
- Leader 接收客户端的请求，并将请求作为一个新的日志条目添加到自己的日志中。
- Leader 向所有 Follower 发送 AppendEntries 请求，将新的日志条目复制到 Follower 的日志中。
- 如果 Follower 成功地将日志条目添加到自己的日志中，就回复 Leader 一个成功的响应。
- 如果 Leader 收到了大多数 Follower 的成功响应，就认为这个日志条目已经被提交（committed），并将这个日志条目应用到自己的状态机中。
-
Safety（安全性）：
- Raft 保证在任何时候，只有一个 Leader。
- Raft 保证已经提交的日志条目不会被覆盖。
- Raft 保证所有节点的状态机最终会达到一致。
Raft 的 C++ 实现 (简化版)
同样，这里我们提供一个简化版的 Raft C++ 实现，主要演示 Leader 选举和日志复制过程。
#include <iostream>
#include <vector>
#include <algorithm>
#include <random>
#include <chrono>
#include <thread>
#include <mutex>
using namespace std;
// 日志条目
struct LogEntry {
int term;
int command;
};
// Raft 节点
class RaftNode {
public:
RaftNode(int id, int numNodes) : nodeId(id), numNodes(numNodes), currentTerm(0), votedFor(-1), state(Follower),
electionTimeout(generateRandomTimeout()), heartbeatInterval(50),
lastApplied(0) {}
// 启动节点
void start() {
thread t(&RaftNode::run, this);
t.detach();
}
private:
enum State {
Follower,
Candidate,
Leader
};
// 运行节点
void run() {
while (true) {
switch (state) {
case Follower:
runAsFollower();
break;
case Candidate:
runAsCandidate();
break;
case Leader:
runAsLeader();
break;
}
}
}
// 作为 Follower 运行
void runAsFollower() {
unique_lock<mutex> lock(mtx);
electionTimeoutCv.wait_for(lock, chrono::milliseconds(electionTimeout), [&]{ return receivedHeartbeat; });
if (!receivedHeartbeat) {
cout << "Node " << nodeId << ": Election timeout, becoming candidate." << endl;
state = Candidate;
} else {
receivedHeartbeat = false;
electionTimeout = generateRandomTimeout();
}
}
// 作为 Candidate 运行
void runAsCandidate() {
currentTerm++;
votedFor = nodeId;
int votesReceived = 1;
cout << "Node " << nodeId << ": Starting election for term " << currentTerm << endl;
// 发送 RequestVote 请求
for (int i = 0; i < numNodes; ++i) {
if (i != nodeId) {
thread t(&RaftNode::sendRequestVote, this, i, currentTerm, log.size() -1, log.empty() ? -1 : log.back().term);
t.detach();
}
}
unique_lock<mutex> lock(mtx);
electionTimeout = generateRandomTimeout();
electionTimeoutCv.wait_for(lock, chrono::milliseconds(electionTimeout), [&]{ return receivedHeartbeat || state == Leader || votesReceived > numNodes / 2; });
if (votesReceived > numNodes / 2) {
cout << "Node " << nodeId << ": Received majority votes, becoming leader." << endl;
state = Leader;
sendHeartbeats(); // 立即发送心跳
} else if (receivedHeartbeat) {
cout << "Node " << nodeId << ": Received heartbeat from leader, reverting to follower." << endl;
state = Follower;
} else {
cout << "Node " << nodeId << ": Election failed, starting new election." << endl;
}
votedFor = -1; // 重置 votedFor
}
// 作为 Leader 运行
void runAsLeader() {
while (state == Leader) {
sendHeartbeats();
this_thread::sleep_for(chrono::milliseconds(heartbeatInterval));
}
}
// 发送 RequestVote 请求
void sendRequestVote(int targetNode, int term, int lastLogIndex, int lastLogTerm) {
lock_guard<mutex> lock(nodeMutexes[targetNode]); // 使用目标节点的互斥锁
if (term < currentTerm) {
return;
}
// 模拟网络延迟
this_thread::sleep_for(chrono::milliseconds(generateRandomDelay()));
bool voteGranted = nodes[targetNode]->handleRequestVote(term, nodeId, lastLogIndex, lastLogTerm);
lock_guard<mutex> lock2(mtx);
if (voteGranted && state == Candidate && term == currentTerm) {
cout << "Node " << nodeId << ": Received vote from node " << targetNode << " for term " << term << endl;
votesReceived++;
electionTimeoutCv.notify_one();
}
}
// 处理 RequestVote 请求
bool handleRequestVote(int term, int candidateId, int lastLogIndex, int lastLogTerm) {
lock_guard<mutex> lock(mtx);
if (term < currentTerm) {
cout << "Node " << nodeId << ": Rejected vote request from node " << candidateId << " for term " << term << " (current term is " << currentTerm << ")" << endl;
return false;
}
if (term > currentTerm) {
cout << "Node " << nodeId << ": Detected higher term " << term << " from node " << candidateId << ", becoming follower." << endl;
currentTerm = term;
state = Follower;
votedFor = -1; // 重置 votedFor
}
if ((votedFor == -1 || votedFor == candidateId) && (lastLogTerm > (log.empty() ? -1 : log.back().term) || (lastLogTerm == (log.empty() ? -1 : log.back().term) && lastLogIndex >= (int)log.size() -1))) {
votedFor = candidateId;
receivedHeartbeat = true; // 避免立即发起选举
electionTimeout = generateRandomTimeout();
cout << "Node " << nodeId << ": Voted for node " << candidateId << " in term " << term << endl;
return true;
}
cout << "Node " << nodeId << ": Rejected vote request from node " << candidateId << " for term " << term << " (already voted for " << votedFor << ")" << endl;
return false;
}
// 发送心跳
void sendHeartbeats() {
for (int i = 0; i < numNodes; ++i) {
if (i != nodeId) {
thread t(&RaftNode::sendAppendEntries, this, i, currentTerm);
t.detach();
}
}
}
// 发送 AppendEntries 请求
void sendAppendEntries(int targetNode, int term) {
lock_guard<mutex> lock(nodeMutexes[targetNode]); // 使用目标节点的互斥锁
if (term < currentTerm) {
return;
}
// 模拟网络延迟
this_thread::sleep_for(chrono::milliseconds(generateRandomDelay()));
bool success = nodes[targetNode]->handleAppendEntries(term, nodeId);
lock_guard<mutex> lock2(mtx);
if (!success && state == Leader) {
// 处理 AppendEntries 失败的情况，例如回退日志等
}
}
// 处理 AppendEntries 请求
bool handleAppendEntries(int term, int leaderId) {
lock_guard<mutex> lock(mtx);
if (term < currentTerm) {
cout << "Node " << nodeId << ": Rejected heartbeat from node " << leaderId << " for term " << term << " (current term is " << currentTerm << ")" << endl;
return false;
}
cout << "Node " << nodeId << ": Received heartbeat from node " << leaderId << " for term " << term << endl;
receivedHeartbeat = true;
electionTimeout = generateRandomTimeout();
if (term > currentTerm) {
cout << "Node " << nodeId << ": Detected higher term " << term << " from node " << leaderId << ", becoming follower." << endl;
currentTerm = term;
state = Follower;
} else if (state == Candidate) {
cout << "Node " << nodeId << ": Received heartbeat from leader, reverting to follower." << endl;
state = Follower;
}
votedFor = -1; // 重置 votedFor
return true;
}
// 生成随机超时时间
int generateRandomTimeout() {
random_device rd;
mt19937 gen(rd());
uniform_int_distribution<> distrib(2 * heartbeatInterval, 4 * heartbeatInterval);
return distrib(gen);
}
// 生成随机延迟
int generateRandomDelay() {
random_device rd;
mt19937 gen(rd());
uniform_int_distribution<> distrib(0, heartbeatInterval / 2);
return distrib(gen);
}
public:
int nodeId;
private:
int numNodes;
int currentTerm;
int votedFor;
State state;
vector<LogEntry> log;
int electionTimeout;
int heartbeatInterval;
bool receivedHeartbeat = false;
int votesReceived = 0;
int lastApplied;
mutex mtx;
condition_variable electionTimeoutCv;
static vector<RaftNode*> nodes; // 存储所有节点
static vector<mutex> nodeMutexes; // 存储每个节点的互斥锁
public:
static void setNodes(vector<RaftNode*> &n) {
nodes = n;
nodeMutexes.resize(n.size());
}
};
vector<RaftNode*> RaftNode::nodes;
vector<mutex> RaftNode::nodeMutexes;
int main() {
int numNodes = 5;
vector<RaftNode*> nodes(numNodes);
for (int i = 0; i < numNodes; ++i) {
nodes[i] = new RaftNode(i, numNodes);
}
RaftNode::setNodes(nodes); // 设置静态成员变量
// 启动所有节点
for (int i = 0; i < numNodes; ++i) {
nodes[i]->start();
}
// 运行一段时间
this_thread::sleep_for(chrono::seconds(10));
// 清理内存
for (int i = 0; i < numNodes; ++i) {
delete nodes[i];
}
return 0;
}注意： 这只是一个非常简化的 Raft 实现，没有考虑日志压缩、成员变更等复杂情况。在实际应用中，需要使用更完善的 Raft 库。
Paxos vs Raft：选择哪个？
| 特性 | Paxos | Raft |
|---|---|---|
| 易理解性 | 非常难 | 相对容易 |
| 实现难度 | 非常难 | 相对容易 |
| 性能 | 理论上更高，但实际实现可能受影响 | 稍逊，但足够满足大多数场景 |
| 适用场景 | 对性能要求非常高的底层基础设施 | 大多数分布式系统，尤其是需要快速开发和迭代的场景 |
总的来说，如果你的团队对一致性算法有深入的理解，并且对性能有极致的追求，可以选择 Paxos。否则，Raft 是一个更明智的选择，因为它更容易理解、实现和维护。
总结
Paxos 和 Raft 都是非常重要的分布式一致性算法，它们保证了分布式系统中的数据一致性，使得系统更加可靠和稳定。虽然它们的实现都比较复杂，但是理解它们的核心思想，可以帮助我们更好地设计和构建分布式系统。
希望今天的讲座能帮助大家更好地理解 Paxos 和 Raft。谢谢大家！