mutex 与 lock_guard / unique_lock：共享数据的正确加锁姿势

上一篇讲完 std::thread 的生命周期，接下来必须面对并发编程真正的核心问题：多个线程同时碰同一块内存时会发生什么。答案不是「结果可能慢一点」，而是「结果可以是任意值」。下面从数据竞争的原理讲起，把 std::mutex 家族的接口、RAII 锁的三种选择、以及一次锁多把时的死锁避免说清楚，每个结论都配上能跑的代码。

不加锁的累加，结果随机

下面这段代码看起来很合理：8 个线程各加 10 万次，期望 80 万。

#include <thread>
#include <vector>

long long g_value = 0; // 没有锁保护的共享状态

void unsynchronized_add() {
// 反例，不要这么写：g_value += 1 实际是「读 – 改 – 写」三步，
// 两个线程可能读到同一个旧值，各自加 1 再写回 —— 一次自增被吞掉
for (int i = 0; i < 100000; ++i) g_value += 1;
}

void race_demo() {
std::vector<std::thread> pool;
for (int i = 0; i < 8; ++i) pool.emplace_back(unsynchronized_add);
for (auto& t : pool) t.join();
// g_value 通常小于 800000，而且每次运行结果都不一样 —— 这就是数据竞争
}

这段没有 main()，是纯展示片段。之所以不给它配真实输出，是因为它压根没有确定输出：同一份二进制反复跑，会得到 79 万、80 万附近的随机值，偶尔还能「碰巧」等于 80 万。这种「有时候对」比稳定错误更危险。本地测十次都对，压力一大就出错，而且几乎无法复现。

根本原因在编译器允许的假设：单线程视角下，没有其它执行流会偷偷改这块内存。于是 g_value += 1 被拆成「读入寄存器 → 加 1 → 写回内存」三步，中间随时可以被另一个线程插进来。这属于 C++ 标准定义的数据竞争（data race），行为是未定义的。它不只是算错，理论上编译器可以做任何事。

官方文档：C++ 内存模型与数据竞争 — cppreference（「数据竞争」的正式定义就在这里）

std::mutex：三个接口

std::mutex 提供的最基本契约只有三个动作：

用起来是这样：

线程 A 线程 B mutex 状态
│ │ ┌──────────┐
│ lock() ──────────────────────────────────────────────► │ 被 A 持有 │
│ <临界区: 读 value_ / 改 / 写回> │ └──────────┘
│ │ lock() ─────────────────► 阻塞等待中
│ unlock() ────────────────────────────────────────────► ┌──────────┐
│ │ <临界区开始> │ 被 B 持有 │
│ │ unlock() ──────────────► └──────────┘
空闲

手写 lock() / unlock() 基本必错。看反例：

#include <mutex>
#include <stdexcept>

std::mutex g_mtx;
bool g_should_fail = false;

void manual_lock_unlock() {
g_mtx.lock();
// 反例，不要这么写（违反 Core Guidelines CP.20 / CP.50）：
// 中间任何一步抛异常，unlock() 就被跳过 —— 这把锁永久泄漏，
// 其它线程全部永远阻塞在 lock() 上，程序表现为「莫名卡死」
if (g_should_fail) throw std::runtime_error{"boom"};
g_mtx.unlock();
}

问题不在「忘了写 unlock」，而在**unlock 和 lock 之间隔着一个可能抛异常的调用**。你没法保证每条路径都走到 unlock，除非让析构函数来兜底。这就是 RAII：把锁绑在对象生命周期上，析构即解锁。

官方文档：std::mutex — cppreference、C++ Core Guidelines · CP.20（用 RAII，不要裸 lock/unlock）

lock_guard：最省事的 RAII 锁

std::lock_guard 是最简单的一档：构造即加锁，析构即解锁，中间不给你任何操作空间。

// counter_guard.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread counter_guard.cpp -o counter_guard
#include <cstdio>
#include <mutex>
#include <thread>
#include <vector>

class Counter {
long long value_{0};
mutable std::mutex mtx_; // mutable：value() 是 const 函数，也要能加锁

public:
void add(long long delta) {
const std::lock_guard<std::mutex> lock{mtx_}; // 构造即锁，析构即解锁
value_ += delta;
}

long long value() const {
const std::lock_guard<std::mutex> lock{mtx_};
return value_;
}
};

int main() {
constexpr int kThreads = 8;
constexpr long long kPerThread = 100000;

Counter counter;
std::vector<std::thread> pool;
pool.reserve(kThreads);
for (int i = 0; i < kThreads; ++i) {
pool.emplace_back([&counter] {
for (long long k = 0; k < kPerThread; ++k) counter.add(1);
});
}
for (auto& t : pool) t.join();

const long long expected = kThreads * kPerThread;
const long long actual = counter.value();
std::printf("累加结果 = %lld\\n", actual);
std::printf("期望值 = %lld\\n", expected);
std::printf("是否一致 = %s\\n", actual == expected ? "是" : "否");
}

累加结果 = 800000
期望值 = 800000
是否一致 = 是

和开头那段被吞掉自增的代码相比，唯一的区别就是那句 lock_guard。结果永远确定，这正是加锁要买的东西：把随机的错误换成确定的正确。

三个实践细节：

官方文档：std::lock_guard — cppreference

lock_guard 的局限也很明确：不能手动 unlock、不能延迟加锁、不能转移所有权、不能配条件变量。需要这些就得换下一档。

unique_lock：更贵，但什么都能做

std::unique_lock 是「全功能版」。它内部多存了一个「当前是否持有锁」的标志，因此比 lock_guard 略大、略慢一点点。很多人把它理解成「能手动解锁的 lock_guard」，其实多出来的那点开销正是这个标志带来的；换来的是四件 lock_guard 做不到的事：

最有用的是 std::defer_lock + std::lock 这个组合，它能一次原子地锁住多把互斥量，从而避免经典的「A 锁 B、B 锁 A」死锁：

// unique_lock_demo.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread unique_lock_demo.cpp -o unique_lock_demo
#include <cstdio>
#include <mutex>
#include <thread>
#include <utility>

class Account {
long long balance_;
mutable std::mutex mtx_;

public:
explicit Account(long long init) : balance_{init} {}

long long balance() const {
const std::lock_guard<std::mutex> lock{mtx_};
return balance_;
}

friend void transfer(Account& from, Account& to, long long amount);
};

// 一次锁两把：用 std::lock 的死锁避免算法，而不是手写「先锁 from 再锁 to」
void transfer(Account& from, Account& to, long long amount) {
std::unique_lock<std::mutex> lock_from{from.mtx_, std::defer_lock}; // 只包装，不加锁
std::unique_lock<std::mutex> lock_to{to.mtx_, std::defer_lock};
std::lock(lock_from, lock_to); // 要么两把都拿到，要么一把都不持有
from.balance_ -= amount;
to.balance_ += amount;
}

int main() {
Account a{1000};
Account b{500};
std::printf("转账前: a = %lld, b = %lld\\n", a.balance(), b.balance());

// 两个方向相反的并发转账：若手写「先锁 A 再锁 B」，这里就会死锁
std::thread t1{[&a, &b] { transfer(a, b, 100); }};
std::thread t2{[&a, &b] { transfer(b, a, 30); }};
t1.join();
t2.join();

std::printf("转账后: a = %lld, b = %lld\\n", a.balance(), b.balance());
std::printf("总额守恒: %s\\n", (a.balance() + b.balance()) == 1500 ? "是" : "否");

// unique_lock 可以不拥有锁，也可以把所有权转移给别人
std::mutex gate;
std::unique_lock<std::mutex> lock{gate, std::defer_lock};
std::printf("defer_lock 之后 owns_lock() = %s\\n", lock.owns_lock() ? "true" : "false");
lock.lock();
std::printf("手动 lock 之后 owns_lock() = %s\\n", lock.owns_lock() ? "true" : "false");
std::unique_lock<std::mutex> handed_over{std::move(lock)};
std::printf("move 之后: 原对象 %s, 新对象 %s\\n",
lock.owns_lock() ? "true" : "false", handed_over.owns_lock() ? "true" : "false");
}

转账前: a = 1000, b = 500
转账后: a = 930, b = 570
总额守恒: 是
defer_lock 之后 owns_lock() = false
手动 lock 之后 owns_lock() = true
move 之后: 原对象 false, 新对象 true

两个关键点：

• std::lock(l1, l2) 内部用的是「全拿或全不拿」的算法，标准要求它必须避免死锁，所以两个方向相反的转账不会互相卡住。自己写「先锁 A 再锁 B」在单线程时永远正确，一上并发就是定时炸弹。
• 转移所有权后，原对象的 owns_lock() 变成 false，解锁责任一并转移。这正是「把锁交给别人管理」的基础，也是 condition_variable 能工作的前提：wait 需要在等待期间释放锁，醒来后再重新获取。

官方文档：std::unique_lock — cppreference、std::lock（多锁死锁避免）、C++ Core Guidelines · CP.21 / CP.22 / CP.50

try_lock 与 timed_mutex：等不到就走人

有些场景不能无限等锁，比如「拿不到就跳过这一帧」的渲染循环，或者「超时就直接报错」的服务端。这时候用 std::timed_mutex：

// try_lock.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread try_lock.cpp -o try_lock
#include <atomic>
#include <chrono>
#include <cstdio>
#include <mutex>
#include <thread>

int main() {
std::timed_mutex gate;

// ① 没人竞争时，try_lock 立刻成功（它是「试一下」，不是「等下去」）
const bool idle_ok = gate.try_lock();
std::printf("没人竞争: try_lock() -> %s\\n", idle_ok ? "拿到" : "没拿到");
if (idle_ok) gate.unlock();

// ② 有人持锁时，try_lock_for 到点就放弃，绝不无限等
std::atomic<bool> holder_ready{false};
std::thread holder{[&gate, &holder_ready] {
gate.lock();
holder_ready.store(true);
std::this_thread::sleep_for(std::chrono::milliseconds{500}); // 故意占住 500ms
gate.unlock();
}};

while (!holder_ready.load()) std::this_thread::yield(); // 等持锁者就绪
const bool busy_ok = gate.try_lock_for(std::chrono::milliseconds{50});
std::printf("有人持锁: try_lock_for(50ms) -> %s\\n", busy_ok ? "拿到" : "超时放弃");
if (busy_ok) gate.unlock();

holder.join();
const bool after_ok = gate.try_lock_for(std::chrono::milliseconds{50});
std::printf("持锁者退出后: try_lock_for(50ms) -> %s\\n", after_ok ? "拿到" : "超时放弃");
if (after_ok) gate.unlock();
}

没人竞争: try_lock() -> 拿到
有人持锁: try_lock_for(50ms) -> 超时放弃
持锁者退出后: try_lock_for(50ms) -> 拿到

注意第三行：超时不是永久失败，只是「这 50ms 内没拿到」。try_lock_for 一定要检查返回值再决定是否 unlock，否则会对着没持有的锁调用 unlock，那是 UB。

std::mutex 家族该选哪个：

关于 std::recursive_mutex，说句务实的：它允许同一个线程重复加锁（内部记一个计数），看起来能救活「递归函数里顺手加锁」的代码。但递归锁几乎总是设计问题的信号。它掩盖了「临界区边界没划清楚」这件事，并且会让不变量（invariant）在递归中途处于半成品状态，别的线程看不到，你自己也容易写错。真正需要它的场合通常是「要兼容一个已经写好的、内部也会加锁的第三方接口」。其余情况请改成：只在最外层加一次锁，内部函数换成「假设调用方已持锁」的私有实现。

官方文档：std::timed_mutex、std::recursive_mutex、std::shared_mutex（C++17）

三种 RAII 锁怎么选

选型只有一句话：能用 lock_guard 就用 lock_guard，一次要锁多把就用 scoped_lock，只有需要 defer_lock、转移、或者配条件变量时才上 unique_lock。不要因为 unique_lock 功能多就无脑用它，多出来的灵活性是有成本的。

官方文档：std::scoped_lock — cppreference、C++ Core Guidelines · CP.20 / CP.50

线程安全的关键字统计器

把前面的东西串起来，做一个多线程上报事件的次数统计：

// stats.cpp — 编译: g++ -std=c++17 -Wall -O2 -pthread stats.cpp -o stats
#include <cstdio>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

class Stats {
std::map<std::string, long long> counts_;
mutable std::mutex mtx_;

public:
void record(const std::string& key) {
const std::lock_guard<std::mutex> lock{mtx_};
++counts_[key]; // 临界区：读改写一个可能触发扩容的容器
}

std::map<std::string, long long> snapshot() const {
const std::lock_guard<std::mutex> lock{mtx_};
return counts_; // 拷一份出去，锁内不做耗时操作
}
};

int main() {
constexpr int kThreads = 4;
constexpr int kPerThread = 10000;
const std::vector<std::string> keys{"cpu", "mem", "disk"};

Stats stats;
std::vector<std::thread> pool;
pool.reserve(kThreads);
for (int i = 0; i < kThreads; ++i) {
pool.emplace_back([&stats, &keys] {
for (int k = 0; k < kPerThread; ++k) {
stats.record(keys[static_cast<std::size_t>(k) % keys.size()]);
}
});
}
for (auto& t : pool) t.join();

long long total = 0;
for (const auto& [key, n] : stats.snapshot()) { // C++17 结构化绑定
std::printf("%s = %lld\\n", key.c_str(), n);
total += n;
}
std::printf("总计 = %lld（期望 %d）\\n", total, kThreads * kPerThread);
}

cpu = 13336
disk = 13332
mem = 13332
总计 = 40000（期望 40000）

这段有两个值得抄走的写法：

延伸阅读

• std::mutex — cppreference：三大接口的正式语义，注意 lock 对已持锁者调用是 UB
• std::lock_guard / std::unique_lock / std::scoped_lock：三种 RAII 锁的成员清单，选型前对着看一遍
• std::lock：多锁死锁避免算法的官方说明，defer_lock 组合的搭档
• C++ 内存模型 — cppreference：数据竞争、happens-before、顺序一致性的权威定义，理解「为什么锁能修好它」的必读项
• C++ Core Guidelines · 并发篇（CP）：CP.20、CP.21、CP.22、CP.50 都是加锁相关的硬规则

收个尾

数据竞争的根源是「读-改-写」不是原子操作，锁把它变成了原子。三条规矩值得背下来：锁一律用 RAII（lock_guard 默认、scoped_lock 锁多把、unique_lock 只在需要延迟、转移或条件变量时才用），临界区尽可能短，绝不在临界区里做 IO 或长耗时操作。

真正的功夫在划边界，不在挑锁型。锁型选错顶多慢一点，边界划错就是死锁和数据竞争，后者比前者难查得多。