精解计算机系统课程
MIT 分布式系统（五）基于 Raft 库实现分布式 KV 系统 #
系统架构概览
上一章中，我们已经完整介绍了 raft_cpp 中的 Raft 库实现。现在我们要使用这个库来构建一个高可用的分布式 KV 存储系统。
如上图所示，客户端通过 RPC 将 Put/Get/Append 请求发送到 KVServer。KVServer 将请求序列化为 Command 后通过 Raft 库的 Start() 接口提交到 Raft 共识层。Raft 库在日志条目被提交后，通过 applyCh（一个 BlockingQueue<ApplyMsg>）通知 KVServer 的 applier 线程。applier 线程将日志反序列化并应用到状态机中，然后通过 notifyChan 通知正在等待的 HandleCommand 调用，最终返回结果给客户端。
对外接口定义
第一步我们来定义系统与客户端的交互接口。客户端可以发送 Put、Append 和 Get 操作将 KV 数据写到系统中。我们将所有操作的请求和响应定义在 kvcommon.h 中：
// kvcommon.h
// 客户端操作类型
enum OperationOp : uint8_t {
OpPut = 0,
OpAppend = 1,
OpGet = 2
};
// 错误码
enum Err : uint8_t {
OK = 0,
ErrNoKey = 1,
ErrWrongLeader = 2,
ErrTimeout = 3
};
// 客户端命令请求
struct CommandRequest {
std::string key;
std::string value;
OperationOp op = OpGet;
int64_t clientId = 0;
int64_t commandId = 0;
};
// 客户端命令响应
struct CommandResponse {
Err err = OK;
std::string value;
};
// 包装在 Raft 日志中的 Command
struct Command {
CommandRequest request;
};
其中 OperationOp 定义了支持的操作类型：OpPut（写入）、OpAppend（追加）、OpGet（读取）。CommandRequest 中包含 key、value、操作类型，以及用于去重的 clientId 和 commandId。CommandResponse 包含错误码 err 和返回值 value。当节点不是 Leader 时返回 ErrWrongLeader，客户端会切换到其他节点重试。
在 Go 版本中，这些结构通过 protobuf 定义为 RPC 消息。而在 raft_cpp 中，我们使用自定义的二进制序列化方式，由 kvser 命名空间中的 serializeCommand / deserializeCommand 函数完成 Command 与字节数组之间的转换：
// kvcommon.h — kvser 命名空间
namespace kvser {
inline std::vector<uint8_t> serializeCommand(const Command& cmd) {
std::vector<uint8_t> buf;
writeString(buf, cmd.request.key);
writeString(buf, cmd.request.value);
writeUint8(buf, static_cast<uint8_t>(cmd.request.op));
writeInt64(buf, cmd.request.clientId);
writeInt64(buf, cmd.request.commandId);
return buf;
}
inline Command deserializeCommand(const uint8_t* data, size_t size) {
const uint8_t* p = data;
Command cmd;
cmd.request.key = readString(p);
cmd.request.value = readString(p);
cmd.request.op = static_cast<OperationOp>(readUint8(p));
cmd.request.clientId = readInt64(p);
cmd.request.commandId = readInt64(p);
return cmd;
}
} // namespace kvser
这种自定义序列化方式比 protobuf 更轻量，适合在 Raft 日志内部使用。序列化格式为：key（长度+内容）+ value（长度+内容）+ op（1 字节）+ clientId（8 字节）+ commandId（8 字节）。
状态机接口与实现
KVServer 需要维护一个状态机来存储实际的 KV 数据。我们定义了一个抽象接口 KVStateMachine，以便将来替换不同的存储引擎：
// kvstatemachine.h
class KVStateMachine {
public:
virtual ~KVStateMachine() = default;
virtual Err Get(const std::string& key, std::string& value) = 0;
virtual Err Put(const std::string& key, const std::string& value) = 0;
virtual Err Append(const std::string& key, const std::string& value) = 0;
virtual void Close() = 0;
virtual int64_t Size() = 0;
// 快照支持：导出所有 KV 和批量写入
virtual std::vector<std::pair<std::string, std::string>> DumpAll() = 0;
virtual void BulkPut(const std::vector<std::pair<std::string, std::string>>& pairs) = 0;
};
raft_cpp 提供了基于 RocksDB 的具体实现 RocksDBKV：
// kvstatemachine.h
class RocksDBKV : public KVStateMachine {
public:
explicit RocksDBKV(const std::string& path);
~RocksDBKV() override;
Err Get(const std::string& key, std::string& value) override;
Err Put(const std::string& key, const std::string& value) override;
Err Append(const std::string& key, const std::string& value) override;
void Close() override;
int64_t Size() override;
std::vector<std::pair<std::string, std::string>> DumpAll() override;
void BulkPut(const std::vector<std::pair<std::string, std::string>>& pairs) override;
private:
std::unique_ptr<rocksdb::DB> db_;
std::unique_ptr<rocksdb::Options> opts_;
std::string path_;
bool closed_ = false;
};
RocksDBKV 使用 RocksDB 作为底层存储引擎，所有 KV 数据持久化到磁盘。DumpAll() 和 BulkPut() 方法用于快照的导出和恢复。通过抽象接口的设计，未来可以轻松替换为其他存储引擎（如内存哈希表），只需实现 KVStateMachine 接口即可。
KVServer 类设计
有了状态机接口和 Raft 库之后，我们来定义 KVServer 的核心结构。KVServer 继承自 enable_shared_from_this，以便在异步回调中安全地获取自身的 shared_ptr：
// kvserver.h
class KVServer : public std::enable_shared_from_this<KVServer> {
public:
static std::shared_ptr<KVServer> Make(
std::vector<std::shared_ptr<RaftPeer>> peers,
int me,
std::shared_ptr<Persister> persister,
int maxRaftState,
const std::string& dbPath);
void HandleCommand(const CommandRequest& req, CommandResponse& resp);
void Kill();
bool killed() const { return dead_.load(std::memory_order_relaxed); }
private:
KVServer() = default;
// ── 成员变量 ──
mutable std::mutex mu_;
std::atomic<bool> dead_{false};
std::shared_ptr<Raft> rf_; // Raft 实例
std::shared_ptr<BlockingQueue<ApplyMsg>> applyCh_; // apply 通道
int maxRaftState_ = -1; // 快照阈值
int lastApplied_ = 0; // 已应用日志索引
std::unique_ptr<KVStateMachine> stateMachine_; // 状态机
std::unordered_map<int64_t, OperationContext> lastOperations_; // 去重表
// notifyChans_[index] = 通知 HandleCommand 的通道
std::unordered_map<int, std::shared_ptr<BlockingQueue<CommandResponse>>> notifyChans_;
std::thread applierThread_; // applier 线程
};
各成员的作用如下：
rf_：Raft 实例的 shared_ptr，是 KVServer 与共识层交互的桥梁。
applyCh_：BlockingQueue<ApplyMsg> 类型的 apply 通道，Raft 库在日志提交后向此队列推送 ApplyMsg。
stateMachine_：状态机的 unique_ptr，存储实际的 KV 数据（RocksDBKV 实现）。
lastOperations_：去重表，记录每个 clientId 最近一次执行的命令信息（OperationContext）。
notifyChans_：通知通道映射表，key 为日志索引，value 为 BlockingQueue<CommandResponse>。当 applier 线程处理完某条日志后，通过对应通道通知正在等待的 HandleCommand 调用。
maxRaftState_：Raft 状态大小阈值，超过此值时触发快照。
对比 Go 版本，Go 中使用 sync.RWMutex 和 channel 实现，C++ 版本使用 std::mutex 和 BlockingQueue
构造流程：KVServer::Make()
KVServer 使用工厂方法 Make() 进行构造，流程如下：
// kvserver.cpp
std::shared_ptr<KVServer> KVServer::Make(
std::vector<std::shared_ptr<RaftPeer>> peers,
int me,
std::shared_ptr<Persister> persister,
int maxRaftState,
const std::string& dbPath)
{
auto applyCh = std::make_shared<BlockingQueue<ApplyMsg>>();
// 1. 创建 KVServer 实例（使用 new 而非 make_shared，确保 enable_shared_from_this 生效）
std::shared_ptr<KVServer> kv(new KVServer());
kv->maxRaftState_ = maxRaftState;
kv->applyCh_ = applyCh;
// 2. 构造 Raft 实例
kv->rf_ = raft::Raft::Make(peers, me, persister, applyCh);
// 3. 构造 RocksDB 状态机
kv->stateMachine_ = std::make_unique<RocksDBKV>(dbPath);
// 4. 从快照恢复状态
auto snap = persister->ReadSnapshot();
if (!snap.empty()) {
kv->restoreSnapshot(snap);
}
// 5. 启动 applier 线程
kv->applierThread_ = std::thread(&KVServer::applier, kv.get());
return kv;
}
构造流程对应了 Go 版本中的几个步骤：创建 apply 通道、构造 Raft 实例、创建状态机、从快照恢复、启动 applier 协程。在 C++ 版本中，这些步骤完全对应，只是将 Go 的 goroutine 替换为 std::thread，将 channel 替换为 BlockingQueue。值得注意的是使用 new 而非 make_shared 创建 KVServer，这是为了让 enable_shared_from_this 在后续的异步回调中能正确工作。
请求处理：HandleCommand()
客户端命令到来后，首先调用的是 HandleCommand 函数。它的工作流程是：先做去重检查，然后将请求序列化并提交到 Raft，最后等待 applier 线程通过 notifyChan 通知结果。
// kvserver.cpp
void KVServer::HandleCommand(const CommandRequest& req, CommandResponse& resp) {
// 1. 去重检查（非 Get 操作）
{
std::lock_guard<std::mutex> lk(mu_);
if (req.op != OpGet && isDuplicateRequest(req.clientId, req.commandId)) {
auto it = lastOperations_.find(req.clientId);
if (it != lastOperations_.end()) {
resp = it->second.lastResponse; // 直接返回上次的结果
return;
}
}
}
// 2. 序列化命令并提交到 Raft
struct Command cmd;
cmd.request = req;
auto cmdBytes = kvser::serializeCommand(cmd);
auto result = rf_->Start(cmdBytes);
if (!result.isLeader) {
resp.err = ErrWrongLeader;
return;
}
// 3. 等待 applier 线程的通知（带超时）
std::shared_ptr<BlockingQueue<CommandResponse>> ch;
{
std::lock_guard<std::mutex> lk(mu_);
ch = getNotifyChan(result.index);
}
CommandResponse reply;
bool got = ch->pop_for(reply, kExecuteTimeout); // 超时 500ms
if (got) {
resp = reply;
} else {
resp.err = ErrTimeout;
}
// 4. 异步清理通知通道
std::thread([this, index = result.index]() {
std::lock_guard<std::mutex> lk(mu_);
removeOutdatedNotifyChan(index);
}).detach();
}
HandleCommand 的核心逻辑包括：
1. 去重检查：对于 Put 和 Append 操作，检查是否为重复请求。如果是重复请求，直接返回上次的响应结果，避免重复执行。
2. 提交到 Raft：将 CommandRequest 序列化为字节数组，调用 rf_->Start() 提交到 Raft 共识层。如果当前节点不是 Leader，返回 ErrWrongLeader。
3. 等待通知：获取该日志索引对应的 notifyChan，调用 pop_for() 带超时地等待结果。如果超时（kExecuteTimeout = 500ms），返回 ErrTimeout。
4. 异步清理：使用 detach 的线程清理已完成的 notifyChan，避免内存泄漏。
对比 Go 版本中的 DoCommand 函数，C++ 版本使用了 BlockingQueue 的 pop_for() 方法实现带超时的等待，替代了 Go 中的 select + time.After 模式。异步清理通道的逻辑也类似，Go 使用 goroutine，C++ 使用 detach 的 std::thread。
Applier 线程
applier 线程是 KVServer 的核心后台线程，负责从 applyCh 中读取 Raft 库提交的 ApplyMsg，将其反序列化并应用到状态机中，然后通知正在等待的 HandleCommand 调用。
// kvserver.cpp
void KVServer::applier() {
ApplyMsg message;
while (applyCh_->pop(message)) { // 阻塞等待 apply 通道
if (killed()) break;
int lastIndex = -1;
applyOneMessage(message, lastIndex);
// 排空所有待处理消息后再检查快照
ApplyMsg next;
while (!killed() && applyCh_->try_pop(next)) {
applyOneMessage(next, lastIndex);
}
// 检查是否需要快照
if (lastIndex >= 0) {
std::lock_guard<std::mutex> lk(mu_);
if (needSnapshot()) {
takeSnapshot(lastIndex);
}
}
}
}
applier 线程会先处理一条消息，然后通过 try_pop() 排空所有剩余的待处理消息，最后统一检查是否需要快照。这种批量处理方式可以防止日志在应用速度跟不上追加速度时无限增长。对比 Go 版本中的 ApplingToStm 协程，C++ 版本使用 BlockingQueue 的 pop()（阻塞）和 try_pop()（非阻塞）替代了 Go 的 select-case-channel 模式。
applyOneMessage() 是处理单条 ApplyMsg 的核心函数：
// kvserver.cpp
void KVServer::applyOneMessage(const ApplyMsg& message, int& lastIndex) {
if (message.command_valid) {
std::lock_guard<std::mutex> lk(mu_);
// 跳过已应用的消息
if (message.command_index <= lastApplied_) {
return;
}
lastApplied_ = message.command_index;
lastIndex = message.command_index;
// 反序列化 Command
struct Command command = kvser::deserializeCommand(message.command);
CommandResponse response;
// 去重检查
if (command.request.op != OpGet &&
isDuplicateRequest(command.request.clientId, command.request.commandId))
{
response = lastOperations_[command.request.clientId].lastResponse;
} else {
// 应用到状态机
response = applyLogToStateMachine(command);
// 更新去重表（非 Get 操作）
if (command.request.op != OpGet) {
OperationContext ctx;
ctx.maxAppliedCommandId = command.request.commandId;
ctx.lastResponse = response;
lastOperations_[command.request.clientId] = ctx;
}
}
// 通知等待的 HandleCommand（仅当当前是 Leader 且任期匹配）
auto [currentTerm, isLeader] = rf_->GetState();
if (isLeader && message.command_term == currentTerm) {
auto ch = getNotifyChan(message.command_index);
ch->push(response);
}
} else if (message.snapshot_valid) {
std::lock_guard<std::mutex> lk(mu_);
if (rf_->CondInstallSnapshot(
message.snapshot_term, message.snapshot_index, message.snapshot))
{
restoreSnapshot(message.snapshot);
lastApplied_ = message.snapshot_index;
lastIndex = message.snapshot_index;
}
}
}
applyOneMessage 的处理逻辑：
1. 跳过过期消息：如果 command_index 小于等于 lastApplied_，说明是重复消息，直接跳过。
2. 反序列化：将 ApplyMsg 中的 command 字段反序列化为 Command 结构。
3. 去重检查：再次检查是否为重复请求（因为从 HandleCommand 提交到 applier 处理之间可能有时间差）。如果是重复请求，从 lastOperations_ 中取回上次的响应。
4. 应用到状态机：调用 applyLogToStateMachine() 将操作应用到 RocksDBKV 状态机。
5. 更新去重表：对于非 Get 操作，记录 clientId 到 OperationContext 的映射，供后续去重使用。
6. 通知 Leader：只有当前节点是 Leader 且任期匹配时，才通过 notifyChan 通知 HandleCommand。这是因为在 Leader 切换期间，旧 Leader 提交的日志可能仍被提交，但旧 Leader 不应再响应客户端。
7. 快照处理：如果是 snapshot_valid 的消息，调用 CondInstallSnapshot 安装快照并恢复状态。
applyLogToStateMachine() 是实际执行 KV 操作的函数，根据操作类型调用状态机的对应接口：
// kvserver.cpp
CommandResponse KVServer::applyLogToStateMachine(const struct Command& cmd) {
CommandResponse resp;
switch (cmd.request.op) {
case OpPut:
resp.err = stateMachine_->Put(cmd.request.key, cmd.request.value);
break;
case OpAppend:
resp.err = stateMachine_->Append(cmd.request.key, cmd.request.value);
break;
case OpGet:
resp.err = stateMachine_->Get(cmd.request.key, resp.value);
break;
}
return resp;
}
去重检测机制
在分布式系统中，客户端可能因为网络超时而重试已经发送过的请求。如果不做去重，同一个 Put 操作可能被执行多次，导致数据不一致。raft_cpp 通过 lastOperations_ 表和 OperationContext 结构来实现去重：
// kvcommon.h
struct OperationContext {
int64_t maxAppliedCommandId = 0;
CommandResponse lastResponse;
};
每个 clientId 对应一个 OperationContext，记录了该客户端最近一次执行的 commandId 和对应的响应。去重检查的逻辑非常简单：
// kvserver.cpp
bool KVServer::isDuplicateRequest(int64_t clientId, int64_t requestId) {
auto it = lastOperations_.find(clientId);
return it != lastOperations_.end() && requestId <= it->second.maxAppliedCommandId;
}
如果 requestId 小于等于已记录的 maxAppliedCommandId，说明这是一个重复请求，直接返回 lastResponse 中缓存的结果。注意 Get 操作不需要去重，因为读操作是幂等的。
去重检查发生在两个地方：HandleCommand 入口处和 applyOneMessage 中实际应用前。这是因为在 HandleCommand 提交到 Raft 之后、applier 处理之前，可能有其他重复请求也被提交了，所以在 applier 中需要再次检查。
🎬 KV 请求去重检测演示
快照机制
随着系统运行，Raft 日志会不断增长。当日志大小超过阈值 maxRaftState_ 时，KVServer 会触发快照，将状态机的当前状态序列化保存，然后通过 rf_->Snapshot() 通知 Raft 库截断日志。
// kvserver.cpp
bool KVServer::needSnapshot() {
return maxRaftState_ != -1 && rf_->GetRaftStateSize() >= maxRaftState_;
}
void KVServer::takeSnapshot(int index) {
// 1. 导出所有 KV 数据
auto kvPairs = stateMachine_->DumpAll();
// 2. 序列化去重表
std::vector<std::pair<int64_t, OperationContext>> ops(
lastOperations_.begin(), lastOperations_.end());
// 3. 序列化快照并提交给 Raft 库
auto snapshot = kvser::serializeSnapshot(kvPairs, ops);
rf_->Snapshot(index, snapshot);
}
快照内容包含两部分：状态机的所有 KV 数据和去重表 lastOperations_。将去重表也纳入快照是为了确保节点重启后仍能正确去重。快照的序列化格式由 kvser::serializeSnapshot() 和 kvser::deserializeSnapshot() 函数处理，格式为：KV 条目数量 + KV 数据 + 去重条目数量 + 去重数据。
恢复快照时，将快照数据反序列化后恢复状态机和去重表：
// kvserver.cpp
void KVServer::restoreSnapshot(const std::vector<uint8_t>& snap) {
if (snap.empty()) return;
std::vector<std::pair<std::string, std::string>> kvPairs;
std::vector<std::pair<int64_t, OperationContext>> ops;
kvser::deserializeSnapshot(snap.data(), snap.size(), kvPairs, ops);
// 恢复状态机
stateMachine_->BulkPut(kvPairs);
// 恢复去重表
lastOperations_.clear();
for (auto& [cid, ctx] : ops) {
lastOperations_[cid] = ctx;
}
}
快照恢复发生在两个时机：KVServer::Make() 构造时从 Persister 读取持久化的快照数据，以及 applier 线程收到 InstallSnapshot RPC 触发的 snapshot_valid 消息时。两种情况都调用 restoreSnapshot() 完成状态恢复。
🎬 KV 快照机制演示
Clerk 客户端
Clerk 是客户端的封装，提供 Get/Put/Append 三个高级接口，内部实现了自动重试和 Leader 发现逻辑。
// clerk.h
class Clerk {
public:
explicit Clerk(std::vector<std::shared_ptr<KVPeer>> servers);
std::string Get(const std::string& key);
void Put(const std::string& key, const std::string& value);
void Append(const std::string& key, const std::string& value);
private:
std::string doCommand(const std::string& key, const std::string& value, OperationOp op);
std::vector<std::shared_ptr<KVPeer>> servers_;
int leaderId_ = 0; // 缓存的 Leader 节点索引
int64_t clientId_ = 0; // 客户端唯一 ID
int64_t commandId_ = 0; // 递增的命令 ID
};
Clerk 通过 KVPeer 抽象接口与服务端通信。在测试中，InMemKVPeer 直接调用 KVServer::HandleCommand()；在生产环境中，可以使用 gRPC 客户端实现。这种抽象接口的设计使得测试环境可以使用内存模拟网络，而无需启动真实的 gRPC 服务。
doCommand 是核心的重试循环：
// clerk.cpp
std::string Clerk::doCommand(const std::string& key,
const std::string& value, OperationOp op) {
CommandRequest req;
req.key = key;
req.value = value;
req.op = op;
req.clientId = clientId_;
req.commandId = commandId_;
for (;;) {
CommandResponse resp;
bool ok = servers_[leaderId_]->Command(req, resp);
if (!ok || resp.err == ErrWrongLeader || resp.err == ErrTimeout) {
// 网络错误或不是 Leader 或超时，切换到下一个节点重试
leaderId_ = (leaderId_ + 1) % static_cast<int>(servers_.size());
std::this_thread::sleep_for(std::chrono::milliseconds(10));
continue;
}
commandId_++; // 成功后才递增 commandId
return resp.value;
}
}
doCommand 的重试逻辑：
1. 构造 CommandRequest，包含 clientId 和 commandId。
2. 向当前缓存的 leaderId 节点发送请求。
3. 如果返回网络错误（ok=false）、ErrWrongLeader 或 ErrTimeout，则切换到下一个节点重试。
4. 关键点：commandId_ 只在成功后才递增。这保证了如果请求超时但实际被 Raft 提交了，客户端重试时使用相同的 commandId，服务端的去重机制会识别并返回上次的结果。
clientId 在构造函数中通过 nrand() 生成一个随机 64 位整数，确保不同客户端之间不会冲突。commandId 从 0 开始递增，每次成功执行后才加 1。
请求处理全流程总结
让我们总结一下完整的请求处理流程：
1. 客户端：Clerk 调用 Get/Put/Append，内部通过 doCommand 构造 CommandRequest，发送到缓存的 Leader 节点。如果遇到 ErrWrongLeader 或超时，自动切换节点重试。
2. Leader 接收：KVServer::HandleCommand 先做去重检查，然后将请求序列化为 Command 并调用 rf_->Start() 提交到 Raft 共识层。如果不是 Leader，返回 ErrWrongLeader。
3. Raft 共识：Raft 库将 Command 追加到 Leader 日志，通过 AppendEntries RPC 复制到 Follower，获得多数派确认后提交。
4. 应用日志：Raft 库通过 applyCh 将已提交的 ApplyMsg 推送给 KVServer 的 applier 线程。applier 反序列化 Command，应用到状态机，更新去重表。
5. 通知 Leader：applier 通过 notifyChan 将结果通知正在等待的 HandleCommand 调用。
6. 返回客户端：HandleCommand 从 notifyChan 获取结果，返回给 Clerk，Clerk 递增 commandId 并返回结果。
整个流程中，Raft 共识层保证了所有副本以相同顺序应用相同的命令，从而实现强一致性。去重机制保证了客户端重试不会导致重复执行。快照机制保证了日志不会无限增长。这三者共同构成了一个可靠、高效、可扩展的分布式 KV 存储系统。
🎬 KV 请求处理全流程演示
捐赠
整理这本书耗费了我们大量的时间和精力。如果你觉得有帮助，一瓶矿泉水的价格支持我们继续输出优质的分布式存储知识体系，2.99¥，感谢大家的支持。
遵循MIT协议开源。
感谢 「赫蹏」 提供如此优秀的中文排版系统