线程间共享数据

从一张共享订单开始

前两章你学会了喊帮厨、管帮厨。可一旦两个厨师在同一个后厨干活，新麻烦就来了：他们要共用东西——同一块砧板、同一张订单。两人各自埋头干没事，一旦同时去改同一张订单，就乱套了。

想象订单上写着「已做 0 道菜」。厨师甲看了一眼是 0，准备改成 1；与此同时厨师乙也看了一眼，还是 0，也准备改成 1。两人各自把「1」写上去——明明做了两道菜，订单上却只记了 1。少记的那道，账面上凭空蒸发了。问题不在某一个厨师，而在两人同时动同一张订单、谁先谁后没人管。

这一章要解决的就是：当多个线程共用同一份数据时，怎么让他们有序地碰它，而不是一拥而上把数据改乱。最常用的办法，是给后厨配一把「唯一的厨刀」——谁拿到刀谁才能切，别人只能等。

竞态条件与数据竞争：先分清逻辑错误和语言规则

竞态条件（race condition） 是一个逻辑层概念：只要正确性依赖“谁先谁后”，时序改变就可能暴露错误。它不等于“同时写一个变量”；后面 empty() 与 pop() 的组合即使内部都加锁，仍然可能有竞态条件。

数据竞争（data race） 是更严格的 C++ 语言规则：两个线程冲突访问同一内存位置，至少一个写，并且没有原子操作或同步建立顺序，程序行为就是未定义。因此 counter++ 无锁并发既有时序上的竞态条件，也有数据竞争；不能把真实 C++ 程序的可能结果限定为 1 或 2。

一段共享数据通常带着某种约定——书里叫 不变量（invariant）。比如计数器 counter 的不变量是「它的值等于已经发生的 ++ 次数」。counter++ 看起来是一个动作，在机器层面其实是三步：读出当前值 → 加一 → 写回。这三步之间，不变量是暂时破着的。

如果一个线程刚读完、还没写回，另一个线程也读了旧值，两次写回在一个简化机器模型里会形成“丢更新”。下面的交互演示用于理解这种交错，不是 C++ 未定义行为的完整结果枚举：

猜一猜：两个线程各 ++counter 一次（初值 0），最后 counter 一定是 2 吗？先选一种交错走一遍，再下结论。

玩过你就明白了：需要把一次完整的“读、改、写”作为不可被另一条同类操作穿插的单元。互斥量可以建立这个顺序；若只是独立计数，也可以使用满足所需内存顺序的原子类型，后者会在第 5 章展开。

互斥量：唯一的那把厨刀

让操作「不重叠」的标准工具，是 互斥量（mutex）。把它想成后厨那把唯一的厨刀：要切砧板（碰共享数据），先得 lock 拿到刀；切完 unlock 放下刀。同一时刻只有一个人能攥着这把刀，别的线程想拿就只能等。

被这把刀保护起来的那段代码——也就是「拿到刀才能进、任一刻只容一个线程」的代码段——叫 临界区（critical section）。下面这张动画把两个线程争这把刀、临界区被强行串行化的过程演给你看：

看清楚了：A 持刀切砧板时，B 只能空等；A 放下刀，B 才拿到。两段临界区在时间轴上永不重叠——这正是竞态消失的原因。手动 lock / unlock 容易忘记解锁（尤其中途抛异常时），所以 C++ 给了一个 RAII 包装 std::lock_guard——它构造时上锁、析构时自动解锁，具体写法第六节细说。

互斥量不会“绑定”某个变量，正确性来自一条共同协议：所有可能触碰同一不变量的路径都使用同一保护机制，并且锁的作用域覆盖完整操作。漏掉一次读、提前返回内部引用，或把检查与执行拆到两个临界区，都会让协议失效。

接口竞态：每步都加锁，仍可能翻车

这里有个反直觉的坑：给每个操作都单独加了锁，组合起来用却仍可能有竞态。 经典例子是 std::stack 的 empty()、top()、pop()。哪怕这三个成员函数各自内部都正确加了锁，下面这种「先查空、再取顶、再弹出」的组合用法仍然是错的：在你 empty() 返回 false 之后、top() 之前，别的线程可能把栈弹空了。问题不在单个操作，而在接口被设计成「查」和「用」分两步——这一节先建立认识，第六节给出根治的改法（把 top + pop 合并成一个返回值的原子操作）。

死锁：两厨师各攥一件对方要的工具

加锁能救竞态，但用多把锁时会引入新麻烦——死锁（deadlock）。设想两把锁 lock1、lock2：线程 A 先锁了 lock1、又伸手要 lock2；线程 B 先锁了 lock2、又伸手要 lock1。于是 A 等 B 手里的 lock2、B 等 A 手里的 lock1——环形等待，谁都动不了。像两个厨师各攥着一件对方要的工具，互不相让，后厨就此僵死。下面这张动画把环一步步画闭合：

避免死锁最直接的办法是让所有线程遵守同一锁层级；也可以用 std::lock / std::scoped_lock 获取一组锁。算法执行期间可能暂时持有其中一部分并回退重试，准确承诺是“不会因为传入锁的获取顺序形成死锁”，不是每个中间瞬间都“全有或全无”。

还要遵守三条组合规则：能不嵌套就不嵌套；持锁时不要调用用户回调或其他未知代码；必须嵌套时，把锁层级写成可审查的不变量。固定顺序应使用稳定业务层级或 std::less 定义的指针全序，不能依赖无关对象裸指针的内建 < 比较。

灵活加锁：更趁手的几种锁

lock_guard 足够应付大多数场景，但它太「死板」——一旦构造就锁到作用域结束、中途不能解开、也不能转移。需要更灵活时，标准库还给了几样：std::unique_lock 支持延迟上锁、提前解锁、移动转移；call_once / once_flag 用于线程安全的惰性初始化（某段初始化只跑一次）；读写锁（shared_mutex） 则区分「读」和「写」——允许多个读线程同时进、但写线程独占，适合读多写少。这些都留到第六节用代码逐个上手。

上手玩一玩这三张图

这三张图都是可操作的教具，配合本章一起玩，比读十遍文字管用：

• 竞态交错器 ``：选不同的「交错顺序」预设，单步走每个微步。重点盯「丢更新」那条——两个线程都先读到旧值 0，最后结果只有 1。换几条交错对比，体会「同样两次 ++，结果由顺序决定」。
• mutex 串行化 <MutexSerializeDiagram />：单步看 A 持刀、B 等待、A 放刀、B 才拿——两段临界区在时间轴上错开不重叠，这就是锁消除竞态的原理。
• 死锁环 <DeadlockCycleDiagram />：单步看四条边怎么接成一个环。盯住第 ③、④ 步——A 等 B 手里的锁、B 等 A 手里的锁，环一闭合就僵死。

玩熟这三张图，下一节的代码每一行 lock / unlock / std::lock 你都能对上图里某个节点。

代码：把共享数据一步步护好

翻车现场：无保护的竞态计数器

先看会出问题的代码。两个线程各对全局 counter 加 100 万次，期望最后是 200 万——但跑出来常常少：

#include <thread>
 
int counter = 0; // 共享数据，无任何保护
 
void add_many() {
    for (int i = 0; i < 1'000'000; ++i)
        ++counter; // 读—改—写，会被另一个线程打断
}
 
int main() {
    std::thread t1(add_many);
    std::thread t2(add_many);
    t1.join();
    t2.join();
    // 数据竞争：C++ 行为未定义，不能只假设“结果偏小”
}

交错器展示了常见的丢更新路径，但这段程序在 C++ 中有数据竞争，编译器无需维持该简化模型。修复目标不是“让结果大概率正确”，而是建立同步关系，彻底移除数据竞争。

加锁：std::mutex + std::lock_guard

给 counter 配一把 std::mutex，每次访问前用 std::lock_guard 圈出临界区：

#include <mutex>
#include <thread>
 
int counter = 0;
std::mutex m; // 保护 counter 的「那把刀」
 
void add_many() {
    for (int i = 0; i < 1'000'000; ++i) {
        std::lock_guard<std::mutex> guard(m); // 上锁，进入临界区
        ++counter;
    } // guard 析构 → 自动 unlock，离开临界区
}

std::lock_guard<std::mutex> guard(m); 在构造时锁住 m、在它离开作用域（这里是每轮循环结束）时自动解锁。这样两个线程的 ++counter 被串行化，最终一定是 200 万。

接口竞态：empty / top / pop 的组合陷阱

光给每个操作加锁还不够。看这段「线程安全栈」的危险用法——即便 empty()、top()、pop() 内部各自加了锁：

// 假设 s 是一个「每个成员都加了锁」的线程安全栈
if (!s.empty()) {       // ① 查：此刻非空
    int v = s.top();    // ② 取顶 —— 但 ① 之后可能已被别的线程弹空！
    s.pop();            // ③ 弹出
    use(v);
}

empty() 返回 false 后、top() 之前，另一个线程可能把栈弹空了——于是 top() 操作一个空栈，未定义行为。问题在接口把「查」和「用」拆成两步（check-then-act），两步之间有缝。

根治办法是改接口——把「取顶 + 弹出」合并成一个加了锁的原子操作，让外部无缝可乘：

#include <mutex>
#include <stack>
#include <optional>
 
template <typename T>
class SafeStack {
    std::stack<T> data;
    mutable std::mutex m;
public:
    void push(T v) {
        std::lock_guard<std::mutex> g(m);
        data.push(std::move(v));
    }
    // 合并 top+pop：一把锁内完成，空栈返回 nullopt
    std::optional<T> pop() {
        std::lock_guard<std::mutex> g(m);
        if (data.empty()) return std::nullopt;
        T v = std::move(data.top());
        data.pop();
        return v;
    }
};

现在调用方只用一句 auto v = s.pop();，「查空 + 取顶 + 弹出」在同一把锁里一气呵成，中间没有缝可钻。

这个简化版本还要求 T 的移动构造满足你需要的异常保证：若移动会抛出并修改源对象，栈顶可能保留一个被部分移动的值。生产接口可以约束 T 为 nothrow-move、在可用时先复制，或像原书示例那样返回 std::shared_ptr<T>，先安全构造返回对象再修改容器。

死锁与修法：固定顺序 / std::lock / std::scoped_lock

多把锁是死锁的温床。下面这段「交换两个账户余额」的代码，如果线程甲调 swap(a, b)、线程乙同时调 swap(b, a)，就可能各持一把、互等另一把——死锁（正是上面死锁环演示的样子）：

struct Account { std::mutex m; int balance; };
 
void swap_bad(Account& x, Account& y) {
    std::lock_guard<std::mutex> g1(x.m); // 甲先锁 x
    std::lock_guard<std::mutex> g2(y.m); // 乙先锁 y → 环形等待，死锁
    std::swap(x.balance, y.balance);
}

最干净的修法是用 std::scoped_lock（C++17）一次锁住两把，它内部用「全有或全无」算法避开死锁：

void swap_ok(Account& x, Account& y) {
    if (&x == &y) return; // 同一个 mutex 不能作为两个参数重复获取
    std::scoped_lock guard(x.m, y.m);
    std::swap(x.balance, y.balance);
}

灵活加锁：unique_lock（延迟 / 提前解锁）

lock_guard 不能中途解锁。要在临界区内「先解锁、做点不需要锁的耗时活、再上锁」，或想延迟上锁，就用 std::unique_lock：

#include <mutex>
 
std::mutex m;
 
void process() {
    std::unique_lock<std::mutex> lk(m); // 上锁
    auto data = read_shared();          // 在锁内读共享数据
    lk.unlock();                        // 提前解锁：下面这步不碰共享数据
    auto result = heavy_compute(data);  // 耗时计算，不占着锁
    lk.lock();                          // 重新上锁
    write_shared(result);
}                                       // lk 析构，确保最终解锁

unique_lock 还支持 std::defer_lock（构造时先不锁、之后再 lock）和移动（把锁的所有权交给别的 unique_lock 或从函数返回），灵活得多——代价是比 lock_guard 略有运行时开销。条件变量（下一章）必须配它用。

惰性初始化：call_once + once_flag

某个昂贵资源想「用到时才初始化、且只初始化一次」，多线程下别手写双重检查锁定——用 std::call_once：

#include <mutex>
#include <memory>
 
std::once_flag flag;
std::unique_ptr<Resource> res;
 
Resource& get_resource() {
    std::call_once(flag, [] {
        res = std::make_unique<Resource>();
    });
    return *res;                    // 之后每次调用都直接返回，已就绪
}

std::call_once 保证一次成功返回的初始化对其他返回的调用可见。若 callable 抛异常，本次调用是 exceptional，flag 不会被置为完成，后续调用还会重试；所以“最多执行一次”只适用于成功完成的那次初始化。C++11 起，函数内静态变量的初始化也由语言保证线程安全，适合能直接构造并返回的单例资源。

读多写少：shared_mutex 读写锁

如果一份数据被频繁读、偶尔写（如配置表、DNS 缓存），用普通 mutex 会让读线程也互相排斥，浪费并发。C++17 的 std::shared_mutex 区分两种锁：

#include <shared_mutex>
#include <map>
#include <string>
 
std::shared_mutex rw;
std::map<std::string, std::string> dns_cache;
 
// 读：共享锁，多个读线程可同时进
std::string lookup(const std::string& host) {
    std::shared_lock<std::shared_mutex> lk(rw);
    auto it = dns_cache.find(host);
    return it == dns_cache.end() ? "" : it->second;
}
 
// 写：独占锁，写时谁都不能进（连读也挡）
void update(const std::string& host, const std::string& ip) {
    std::lock_guard<std::shared_mutex> lk(rw);
    dns_cache[host] = ip;
}

读用 std::shared_lock（多读并行），写用 std::lock_guard / std::unique_lock（独占）。它是否快于普通 mutex 必须实测：读临界区太短时管理开销可能更高，而且标准不保证读者或写者公平，具体实现可能出现一方长期等待。

容易踩的坑

小结

• 数据竞争是 C++ 的未定义行为条件；竞态条件是更广的逻辑时序错误，接口的 check-then-act 即使各步加锁也可能竞态
• 用 std::mutex + std::lock_guard 把读写共享数据的代码圈成临界区：构造上锁、析构自动解锁，任一刻只有一个线程能进，竞态消失
• 接口竞态：每个操作都加锁，组合用（empty→top→pop）仍可能 race；修法是把需要原子完成的复合逻辑合并成一个加锁的操作
• 死锁：多把锁顺序不一致 → 环形等待僵死；用 std::lock / std::scoped_lock 一次锁多把，或约定全局固定加锁顺序避免
• 灵活加锁：unique_lock 可延迟/提前解锁/移动，call_once 在异常后会重试初始化，shared_mutex 允许多读但性能与公平性由实现和负载决定

long total = 0;
 
void accumulate(const std::vector<int>& v) {
    for (int x : v)
        total += x; // 竞态：读—改—写被打断
}

#include <mutex>
 
long total = 0;
std::mutex m; // 保护 total
 
void accumulate(const std::vector<int>& v) {
    for (int x : v) {
        std::lock_guard<std::mutex> g(m); // 上锁
        total += x;                       // 临界区：唯一线程在改 total
    }                                     // g 析构 → 自动解锁
}

#include <mutex>
 
class Counter {
    long value = 0;
    mutable std::mutex m; // mutable：const 成员函数也能锁它
public:
    void increment() {
        std::lock_guard<std::mutex> g(m);
        ++value;
    }
    long get() const {
        std::lock_guard<std::mutex> g(m); // const 函数里加锁 → mutex 必须 mutable
        return value;                     // 返回值拷贝，别返回引用（否则锁外被读）
    }
};