精解计算机系统课程
MIT 分布式系统（二）C++ 并发编程基础 #
C++17 与并发编程概述
raft_cpp 项目使用 C++17 实现 MIT 6.824 分布式系统 Labs。C++17 在 C++11/14 的基础上进一步完善了标准库，提供了丰富的并发编程支持。与 Go 语言内置的 goroutine 和 channel 不同，C++ 使用标准库的 std::thread、std::mutex、std::condition_variable 等原语来实现并发编程。
在 raft_cpp 中，Raft 算法需要多个后台线程同时运行（选举超时检测、日志复制、日志应用等），这些线程之间需要通过共享状态和同步机制进行协作。本章将介绍 raft_cpp 中使用到的 C++ 并发编程基础知识，所有代码示例均来自 raft_cpp 的实际代码。
std::thread：线程创建与管理
C++11 引入了 std::thread，它是对操作系统线程的封装。在 raft_cpp 中，Raft 库的每个后台任务都使用 std::thread 创建。对比 Go 语言中的 goroutine，std::thread 是操作系统级别的线程，创建成本更高，但对于 Raft 算法这种少量长期运行的后台任务来说完全够用。
// 创建线程的基本方式
std::thread t(&Raft::ticker, this); // 成员函数指针 + 对象指针
std::thread t2(&Raft::applier, this);
std::thread t3(&Raft::replicator, this, peerId);
// join：等待线程结束
t.join();
// detach：让线程在后台运行
std::thread([]() {
// 清理逻辑
}).detach();
在 raft_cpp 的 Raft::Make() 工厂方法中，启动了三个后台线程：
// raft.cpp std::shared_ptrRaft::Make(...) { // ... raft->tickerThread_ = std::thread(&Raft::ticker, raft.get()); raft->applierThread_ = std::thread(&Raft::applier, raft.get()); // 每个 peer 一个 replicator 线程 for (int i = 0; i < numPeers; ++i) { if (i != raft->me_) { raft->replicatorThreads_.emplace_back(&Raft::replicator, raft.get(), i); } } return raft; }
析构时需要先停止 Raft 的后台线程，再 join 自己的线程。raft_cpp 在析构函数中先调用 Kill() 设置停止标志并关闭 apply 通道，然后 join 所有线程：
// 析构函数
Raft::~Raft() {
Kill();
tickerThread_.join();
applierThread_.join();
for (auto& t : replicatorThreads_) {
if (t.joinable()) t.join();
}
}
与 Go 的 go 关键字相比，std::thread 需要手动管理生命周期（join 或 detach），但这给了开发者更精细的控制。在 raft_cpp 中，detach 主要用于异步清理任务（如清理 notifyChan），这些任务不需要等待完成。
std::mutex 与 std::lock_guard：互斥访问共享状态
多个线程访问同一共享状态时必须加锁保护。C++ 标准库提供了 std::mutex 和 std::lock_guard（RAII 封装），对应 Go 中的 sync.Mutex。
// raft.h — Raft 类中的互斥锁
class Raft {
private:
mutable std::mutex mu_;
// ... 受 mu_ 保护的状态
};
// 使用 lock_guard 自动加锁和解锁
void Raft::HandleRequestVote(const RequestVoteRequest& req,
RequestVoteResponse& resp) {
std::lock_guard<std::mutex> lk(mu_);
// 在锁的保护下访问共享状态
if (req.term < currentTerm_) {
resp.term = currentTerm_;
resp.voteGranted = false;
return;
}
// ...
}
// lk 离开作用域时自动解锁
std::lock_guard 是 RAII（Resource Acquisition Is Initialization）模式的封装：构造时加锁，析构时解锁。这保证了即使函数中抛出异常，锁也能正确释放。对应 Go 中的 defer mu.Unlock() 模式，但 C++ 的方式更加自动化。
在 Persister 类中，互斥锁用于保护持久化状态的读写：
// persister.h
class Persister {
public:
void SaveRaftState(const std::vector<uint8_t>& state) {
std::lock_guard<std::mutex> lk(mu_);
raftstate_ = state;
saveToDisk();
}
std::vector<uint8_t> ReadRaftState() const {
std::lock_guard<std::mutex> lk(mu_);
return raftstate_;
}
private:
mutable std::mutex mu_; // mutable 允许 const 方法加锁
std::vector<uint8_t> raftstate_;
std::vector<uint8_t> snapshot_;
std::string path_;
};
注意 mu_ 被声明为 mutable，这是因为 ReadRaftState() 是 const 方法但需要加锁修改 mutex 的状态。这是 C++ 中常见的模式。
std::condition_variable：线程间同步
std::condition_variable 用于线程间的等待-通知机制，对应 Go 中的 channel 或 sync.Cond。在 raft_cpp 中，replicator 线程使用 condition_variable 来等待需要复制的新日志条目：
// raft.h — replicator 的同步机制
class Raft {
private:
std::mutex replicatorMu_;
std::condition_variable replicatorCv_;
// ...
};
// replicator 线程等待新日志
void Raft::replicator(int peerId) {
while (!killed()) {
// 等待需要复制的信号
std::unique_lock<std::mutex> lk(replicatorMu_);
replicatorCv_.wait(lk, [&] {
return killed() || /* 有新日志需要复制 */;
});
if (killed()) return;
// 执行日志复制
replicateOneRound(peerId);
}
}
// 当有新日志时通知 replicator
void Raft::Start(const std::vector<uint8_t>& command) {
// ...
{
std::lock_guard<std::mutex> lk(replicatorMu_);
// 追加日志
logs_.push_back(entry);
}
replicatorCv_.notify_all(); // 唤醒所有 replicator 线程
}
condition_variable 的典型使用模式：
1. 等待线程获取 unique_lock（不是 lock_guard，因为 wait 需要释放和重新获取锁）。
2. 调用 wait()，传入谓词（lambda）。wait() 会原子地释放锁并阻塞线程；当被唤醒时，重新获取锁并检查谓词。
3. 通知线程修改共享状态后调用 notify_one() 或 notify_all()。
这种模式对应 Go 中的：
// Go 的 cond 用法（对比）
cond.L.Lock()
for !hasWork {
cond.Wait()
}
cond.L.Unlock()
// 通知方
cond.L.Lock()
cond.Signal() // 或 cond.Broadcast()
cond.L.Unlock()
C++ 的 condition_variable 配合 lambda 谓词更加简洁，避免了 Go 中手动写 for 循环的样板代码。
智能指针：自动内存管理
C++ 没有垃圾回收（GC），但 C++11 引入的智能指针提供了自动的引用计数式内存管理。raft_cpp 中大量使用了三种智能指针：
std::shared_ptr：共享所有权的智能指针，多个 shared_ptr 可以指向同一对象，引用计数降为 0 时自动销毁对象。在 raft_cpp 中，Raft 实例、Persister、BlockingQueue 等都通过 shared_ptr 管理生命周期：
// 创建 shared_ptr
auto applyCh = std::make_shared<BlockingQueue<ApplyMsg>>();
auto persister = std::make_shared<Persister>(path);
auto raft = raft::Raft::Make(peers, me, persister, applyCh);
// 作为成员变量
class KVServer {
std::shared_ptr<Raft> rf_;
std::shared_ptr<BlockingQueue<ApplyMsg>> applyCh_;
};
std::unique_ptr：独占所有权的智能指针，不能拷贝，只能移动。适用于不需要共享的场景，如 KVStateMachine：
// kvserver.h
class KVServer {
std::unique_ptr<KVStateMachine> stateMachine_;
};
// 构造时创建
kv->stateMachine_ = std::make_unique<RocksDBKV>(dbPath);
std::weak_ptr：弱引用，不增加引用计数，用于打破循环引用。在 ShardKV 中，后台线程通过 weak_ptr 引用 ShardKV 实例，避免线程持有 shared_ptr 导致对象无法销毁：
// shardkv.cpp — Make() 中创建后台线程
std::weak_ptr<ShardKV> weakKv = kv; // 弱引用
kv->configureThread_ = std::thread(&ShardKV::monitorLoop, kv.get(),
[weakKv]() {
auto kv = weakKv.lock(); // 尝试提升为 shared_ptr
if (kv) kv->configureAction(); // 如果对象还活着，执行操作
}, kConfigureMonitorTimeout);
std::enable_shared_from_this：允许在成员函数中安全地获取自身的 shared_ptr。KVServer 和 ShardKV 都继承自它：
// kvserver.h
class KVServer : public std::enable_shared_from_this<KVServer> {
// ...
};
// 在成员函数中使用
void KVServer::HandleCommand(...) {
// ...
// 使用 shared_from_this() 获取自身的 shared_ptr
std::weak_ptr<KVServer> weakSelf = shared_from_this();
std::thread([weakSelf, index = result.index]() {
auto self = weakSelf.lock();
if (!self) return;
std::lock_guard<std::mutex> lk(self->mu_);
self->removeOutdatedNotifyChan(index);
}).detach();
}
注意：使用 enable_shared_from_this 时，对象必须通过 shared_ptr 管理（即通过 new + shared_ptr 创建，而非栈对象或普通 new）。这就是为什么 raft_cpp 的工厂方法使用 new 而非 make_shared：
// 正确：使用 new 创建，确保 enable_shared_from_this 可用
std::shared_ptr<KVServer> kv(new KVServer());
// 错误：make_shared 无法在此场景使用（控制块与对象一体）
// auto kv = std::make_shared<KVServer>(); // 不可用
BlockingQueue<T>：线程安全队列
Go 语言中的 channel 是并发编程的核心原语，用于线程间传递数据。C++ 标准库没有内置的 channel，但 raft_cpp 在 util.h 中实现了一个 BlockingQueue 模板类，完美模拟了 Go channel 的功能：
// util.h
template <typename T>
class BlockingQueue {
public:
// 向队列中推送元素（对应 Go: ch <- val）
void push(T val) {
std::lock_guard<std::mutex> lk(mu_);
if (closed_) return;
q_.push(std::move(val));
cv_.notify_one();
}
// 阻塞式弹出元素（对应 Go: val := <-ch）
// 返回 false 表示队列已关闭且为空
bool pop(T& out) {
std::unique_lock<std::mutex> lk(mu_);
cv_.wait(lk, [&] { return !q_.empty() || closed_; });
if (q_.empty()) return false;
out = std::move(q_.front());
q_.pop();
return true;
}
// 带超时的弹出（对应 Go: select + time.After）
template <typename Duration>
bool pop_for(T& out, Duration timeout) {
std::unique_lock<std::mutex> lk(mu_);
if (!cv_.wait_for(lk, timeout, [&] { return !q_.empty() || closed_; })) {
return false; // 超时
}
if (q_.empty()) return false;
out = std::move(q_.front());
q_.pop();
return true;
}
// 非阻塞式弹出（对应 Go: select { case val := <-ch: ... default: ... }）
bool try_pop(T& out) {
std::lock_guard<std::mutex> lk(mu_);
if (q_.empty()) return false;
out = std::move(q_.front());
q_.pop();
return true;
}
// 关闭队列（对应 Go: close(ch)）
void close() {
std::lock_guard<std::mutex> lk(mu_);
closed_ = true;
cv_.notify_all(); // 唤醒所有等待的线程
}
private:
mutable std::mutex mu_;
std::condition_variable cv_;
std::queue<T> q_;
bool closed_ = false;
};
BlockingQueue 在 raft_cpp 中的应用非常广泛：
- applyCh：Raft 库向 applyCh 推送已提交的 ApplyMsg，KVServer/ShardKV 的 applier 线程从 applyCh 中 pop 消息。
- notifyChans：applier 线程通过 notifyChan 向等待中的 HandleCommand 调用推送响应，HandleCommand 使用 pop_for 带超时等待。
- close() 方法：在 Kill() 中调用 applyCh 的 close()，优雅地唤醒所有阻塞在 pop() 上的线程，使它们能够检查 killed 标志并退出。
// 实际使用示例
// Raft 库的 applier 线程
void Raft::applier() {
ApplyMsg msg;
while (applyCh_->pop(msg)) { // 阻塞等待
if (killed()) break;
// 应用日志...
}
}
// KVServer 的 HandleCommand 等待通知
CommandResponse reply;
bool got = ch->pop_for(reply, kExecuteTimeout); // 500ms 超时
// Kill 时关闭队列
void KVServer::Kill() {
dead_.store(true);
applyCh_->close(); // 唤醒 applier 线程
rf_->Kill();
}
Timer 类：定时器
Go 语言提供了 time.Timer 和 time.After，用于定时操作。在 Raft 算法中，选举超时和心跳超时都需要定时器。raft_cpp 在 util.h 中实现了 Timer 类：
// util.h
class Timer {
public:
Timer() = default;
~Timer() { Stop(); }
// 启动/重启定时器（对应 Go: timer.Reset(duration)）
void Reset(std::chrono::milliseconds duration) {
std::lock_guard<std::mutex> lk(mu_);
deadline_ = std::chrono::steady_clock::now() + duration;
fired_ = false;
running_ = true;
cv_.notify_all();
}
// 取消定时器（对应 Go: timer.Stop()）
void Stop() {
std::lock_guard<std::mutex> lk(mu_);
running_ = false;
cv_.notify_all();
}
// 阻塞等待定时器触发或被停止
// 返回 true 表示触发，false 表示被停止
bool Wait() {
std::unique_lock<std::mutex> lk(mu_);
cv_.wait(lk, [&] {
return !running_ ||
(!fired_ &&
std::chrono::steady_clock::now() >= deadline_);
});
if (running_ && !fired_) {
fired_ = true;
running_ = false;
return true; // 定时器触发
}
return false; // 被停止或已触发
}
// 非阻塞检查是否已触发
bool Fired() {
std::lock_guard<std::mutex> lk(mu_);
if (!running_ && !fired_) return false;
if (fired_) return true;
if (running_ && std::chrono::steady_clock::now() >= deadline_) {
fired_ = true;
running_ = false;
return true;
}
return false;
}
private:
mutable std::mutex mu_;
std::condition_variable cv_;
std::chrono::steady_clock::time_point deadline_{};
bool running_ = false;
bool fired_ = true; // 初始为已触发，Wait() 会阻塞直到 Reset()
};
在 raft_cpp 的 Raft ticker 线程中，Timer 用于检测选举超时和心跳超时：
// raft.cpp
void Raft::ticker() {
while (!killed()) {
// 选举超时检测
if (electionTimer_.Fired()) {
if (state_ != Leader) {
StartElection();
}
electionTimer_.Reset(RandomizedElectionTimeout());
}
// 心跳超时检测
if (heartbeatTimer_.Fired()) {
if (state_ == Leader) {
BroadcastHeartbeat();
}
heartbeatTimer_.Reset(StableHeartbeatTimeout());
}
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
}
其中，RandomizedElectionTimeout() 返回 1000ms~2000ms 的随机超时时间，StableHeartbeatTimeout() 返回 125ms 的固定心跳间隔。随机化选举超时是为了避免多个节点同时发起选举导致活锁。
二进制序列化
Go 版本使用 json.Marshal/Unmarshal 或 protobuf 进行序列化。raft_cpp 选择了自定义二进制序列化方式，这种方式更轻量，不依赖外部库，适合在 Raft 日志内部使用。
序列化的基本模式是：为每种类型定义 writeXxx 和 readXxx 函数，使用简单的二进制格式（长度前缀 + 数据）。以 kvcommon.h 中的 kvser 命名空间为例：
// kvcommon.h — kvser 命名空间
namespace kvser {
// 基础类型写入
inline void writeInt32(std::vector<uint8_t>& buf, int32_t v) {
const auto* p = reinterpret_cast<const uint8_t*>(&v);
buf.insert(buf.end(), p, p + sizeof(v));
}
inline void writeInt64(std::vector<uint8_t>& buf, int64_t v) {
const auto* p = reinterpret_cast<const uint8_t*>(&v);
buf.insert(buf.end(), p, p + sizeof(v));
}
inline void writeString(std::vector<uint8_t>& buf, const std::string& s) {
writeInt32(buf, static_cast<int32_t>(s.size())); // 先写长度
buf.insert(buf.end(), s.begin(), s.end()); // 再写内容
}
// 基础类型读取
inline int32_t readInt32(const uint8_t*& p) {
int32_t v;
std::memcpy(&v, p, sizeof(v));
p += sizeof(v);
return v;
}
inline std::string readString(const uint8_t*& p) {
int32_t len = readInt32(p);
std::string s(reinterpret_cast<const char*>(p), static_cast<size_t>(len));
p += len;
return s;
}
// 复合类型序列化
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 或 json 库，编译后零依赖。
- 高效：二进制格式比 JSON 更紧凑，解析速度更快。
- 类型安全：编译器在编译期检查类型转换。
- 可读性：代码结构清晰，容易理解每种类型的序列化方式。
- 可扩展：添加新的结构体只需编写对应的 serialize/deserialize 函数。
raft_cpp 中有多个序列化命名空间：kvser（KV 操作）、scser（配置操作）、skvser（分片操作），它们共享相同的基础 read/write 函数模式。快照的序列化也是同样的方式，只是包含更多的字段（KV 数据、去重表、配置信息等）。
本章总结
本章介绍了 raft_cpp 项目中使用到的 C++ 并发编程基础知识：
- std::thread：创建后台线程，对应 Go 的 goroutine。
- std::mutex + std::lock_guard：保护共享状态，对应 Go 的 sync.Mutex + defer Unlock。
- std::condition_variable：线程间等待-通知，对应 Go 的 channel/sync.Cond。
- 智能指针：shared_ptr（共享所有权）、unique_ptr（独占所有权）、weak_ptr（弱引用避免循环引用）、enable_shared_from_this（安全获取自身 shared_ptr）。
- BlockingQueue<T>：线程安全队列，完美模拟 Go channel 的 push/pop/close 语义。
- Timer：定时器，模拟 Go 的 time.Timer，用于选举超时和心跳超时检测。
- 二进制序列化：自定义的轻量序列化模式，替代 Go 的 json/protobuf。
掌握这些基础知识后，你就能更好地理解后续章节中 Raft 算法库的 C++ 实现。
捐赠
整理这本书耗费了我们大量的时间和精力。如果你觉得有帮助，一瓶矿泉水的价格支持我们继续输出优质的分布式存储知识体系，2.99¥，感谢大家的支持。
遵循MIT协议开源。
感谢 「赫蹏」 提供如此优秀的中文排版系统