同步并发操作

从等待一份结果开始

前三章你学会了喊帮厨、管帮厨、给共享的砧板配一把刀。可后厨里还有一类活儿：服务员要等厨师把菜做好才能端走。最笨的办法是服务员守在出菜口，每秒探头看一眼「好了没？好了没？」——既累人，又白占着地方。

更聪明的办法是装一只「上菜」铃：服务员先去歇着（不占地方、不空转），厨师把菜做好往台上一摆、顺手按一下铃，服务员听见铃声才过来取。这一章解决的，正是「一个线程要等另一个线程把某件事做完」——怎么让它安静地等、被准点叫醒，而不是傻乎乎地干等。

没有这套机制会怎样？要么服务员死守出菜口空转（浪费 CPU），要么更糟：菜早做好了、铃也按了，可服务员那会儿恰好没在听，铃白响一场，菜就一直摆在台上没人取——漏单。这一章要把「等—通知」这件事做得既不空转、也不漏单。

条件变量：等「上菜」铃，按铃才叫醒

让一个线程「安静地等某个条件成立」的标准工具，是 条件变量（condition_variable）。把它想成出菜口那只铃：消费者线程发现「还没料」时，调 wait 安静睡过去（同时把手里的锁放掉，让别人能改数据）；生产者线程准备好数据后，调 notify 按一下铃，把睡着的消费者叫醒。

但光「被叫醒就往下走」是不够的，这里藏着两个必须知道的坑。其一是 虚假唤醒（spurious wakeup）：操作系统允许 wait 在没人按铃时也自己醒一下。其二是丢失唤醒：如果 notify 发生在你 wait 之前那一瞬，铃响时你还没开始睡，等你睡下就再也等不到那一声了。

根治这两个坑的办法，是给 wait 配一个 谓词（predicate）。谓词和受 mutex 保护的共享状态才是“记忆”，notify 本身不保存历史。生产者必须在同一把锁的协议下修改状态；消费者睡前查一次、醒后重新拿锁再查一次。下面这张动画把完整握手演给你看：

猜一猜：消费者 cv.wait(lk, pred) 时，如果队列是空的（谓词为假），它是「守在那儿每秒查一次」还是「直接睡过去」？睡着时手里那把锁还攥着吗？先想一下，再单步看第一拍。

看清楚了：消费者 wait 时把锁放掉并睡过去（不空转），生产者 push 数据后 notify 按铃叫醒它；消费者醒来先重新抢回锁，再查一次谓词确认真有数据，才取走继续。这「醒来再查」的一步，正是谓词存在的全部意义。

future 与 async：取餐凭证，菜没好就阻塞等

条件变量适合「反复发生」的等待（一个接一个的数据）。但很多时候你只是想「派个活儿出去，过会儿来拿它的结果」——这种「一次性事件 + 一个返回值」用 std::future 更顺手。把它想成点单时换回的「取餐凭证」：你先拿到凭证，等会儿凭凭证去 get() 取菜——菜好了就立刻拿到，没好就在那儿阻塞等着。

最省事的搭配是 std::async。只有显式选择 std::launch::async，才能要求任务在单独线程执行；无策略版本可能被实现选为 deferred。等你需要结果时，f.get() 会等待就绪，并取得值或重新抛出任务异常。

这里有个启动策略必须知道。std::async 的第一个参数可以是 launch 策略：std::launch::async 要求单独线程执行；std::launch::deferred 则推迟到首次 get() 或非定时 wait()，并在调用者线程同步执行。deferred shared state 的定时等待只会报告 future_status::deferred，不会启动任务。

promise 与 packaged_task：手动摆上台，或打包成可搬运的任务

std::async 是「自动雇厨师」，但有时你想自己掌控「谁、何时、怎样」把结果填进那张凭证。标准库另给了两件工具。

一是 std::promise：它是 future 的「写端」，你拿着它在任意线程手动 set_value 或 set_exception。get_future() 只能取得一次，结果也只能提交一次；promise 未提交就被销毁时，读端不会永久挂起，而会在 get() 收到 broken_promise。

二是 packaged_task：它把一个可调用对象包起来，并关联一张 future。task 可以移动、进入容器或交给工作线程，等被调用时才执行并提交结果。线程池的任务队列常用它把“任务是什么”和“由谁执行”分开。

async / packaged_task / promise 这三条路，最后都换回同一种 future、都用 future.get() 取结果，区别只在「谁来填这张凭证」。下面这张图把三者并排对比：

带超时的等待与 shared_future

最后两件趁手的小工具。一是带超时的等待：wait_for 接收 duration，wait_until 接收 clock 的 time point。二是 shared_future：普通 future 的 get() 只能调一次；多个线程需要同一个只读结果时，先用 share() 转换，再把副本分给各线程。对值类型调用 shared_future::get() 返回 const T&，引用的生存期受共享状态约束。

函数式任务链、消息传递与阶段协作

functional programming with futures 的核心是让阶段接收值并产生新值，尽量不共享可变状态；message passing（消息传递） 则让每个线程拥有自己的状态，通过线程安全队列交换命令和结果。两者都把同步边界从“任意共享变量”收拢为“值何时就绪”或“消息何时到达”，但队列关闭、背压、取消和错误传播仍需显式协议。

continuation（延续任务） 可以表达任务链，但要注意版本边界：书中讨论的 future.then(...) 来自 Concurrency TS，并不是 C++17 标准接口。C++17 只能用库、显式调度器或另一个等待任务组合；不能把 TS 示例原样当成可移植标准代码。

阶段型协作后来在 C++20 标准化为 latch（门闩） 和 barrier（栅栏）。std::latch 适合“一组初始化全部完成后放行”，std::barrier 适合迭代算法每一轮末尾会合。第二版原书讨论的是当时的演进方向；本章主体仍按 C++17，不能在代码中假设这两个标准类型已经可用。

上手玩一玩这三张图

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

• 条件变量握手 <CondVarWaitNotifyDiagram />（主图）：单步走六拍，重点盯第①拍（消费者 wait 时释放锁睡眠，不空转）和第⑥拍（醒来再查一次谓词才取数据）。把这两拍看透，谓词为什么不可省就懂了。
• future 阻塞取餐 <FuturePromiseDiagram />：单步看主线程 f.get() 那一拍——任务没算完时主线程被阻塞停住（不是返回个「还没好」），直到结果摆上取餐台才解除。这就是「凭证—取餐」的阻塞语义。
• 三种拿 future 的方式 <GetFutureWaysDiagram />：横向对比三列，盯「谁来填」那一行——async 自动填、packaged_task 被调用时填、promise 你手动填。下一节三段代码正好一一对应这三列。

玩熟这三张图，下一节每一句 wait / notify / get / set_value 你都能对上图里某个节点。

代码：把「等—通知—取结果」一步步写对

生产者-消费者：condition_variable + 谓词

最经典的用法。一个线程把数据 push 进队列、notify 一下；另一个线程 wait 着取。先看消费者怎么等——关键是那个谓词 lambda：

#include <condition_variable>
#include <mutex>
#include <queue>
 
std::mutex m;
std::condition_variable cv;
std::queue<int> q;
 
int consume() {
    std::unique_lock<std::mutex> lk(m);   // 必须用 unique_lock（wait 要中途解锁）
    cv.wait(lk, [] { return !q.empty(); }); // 谓词为假就接着睡；醒来再查
    int v = q.front();                      // 走到这里，谓词一定成立（队列非空）
    q.pop();
    return v;
}

cv.wait(lk, pred) 做了三件事：进来先查一次 pred()，假就原子地「释放 lk + 睡眠」（让生产者能拿到锁去改队列）；被 notify 叫醒后，先重新锁回 lk，再查一次 pred()，仍假就接着睡。所以它等价于 while (!pred()) cv.wait(lk);——这正是上面那张握手动画演的过程。

生产者一端则是：拿锁、改数据、解锁、按铃：

void produce(int v) {
    {
        std::lock_guard<std::mutex> lk(m);
        q.push(v);            // 在锁的保护下改共享队列
    }                         // 先解锁，再 notify（避免被叫醒的线程又立刻卡在锁上）
    cv.notify_one();          // 按铃：叫醒一个在 wait 的消费者
}

起异步任务取结果：std::async + future.get

不想自己摆弄锁和条件变量、只想「派个活儿、过会儿拿结果」，用 std::async 最省事：

#include <future>
 
int heavy_compute() {
    // ……一段耗时计算，返回 42
    return 42;
}
 
void caller() {
    auto f = std::async(std::launch::async, heavy_compute);
    do_other_work();      // 策略明确，主线程可与任务重叠
    int result = f.get(); // 未就绪就等待；异常在这里重抛
    // 此后 result == 42；f 已被取空，不能再 get
}

显式 launch::async 后，任务在单独线程执行；无策略版本则不能承诺与调用者重叠。f.get() 在结果就绪前会阻塞，任务异常会存入共享状态并在这里重新抛出。

std::async 的启动策略默认由实现选，但你常常想明确指定：

// 强制立刻另起线程异步执行
auto f1 = std::async(std::launch::async, heavy_compute);
 
// 推迟执行：任务不动，直到 f2.get()/wait() 时才在「当前线程」同步跑
auto f2 = std::async(std::launch::deferred, heavy_compute);

打包可搬运的任务：packaged_task + get_future

想「先把任务打包、之后再决定在哪个线程、什么时候跑」，用 std::packaged_task。它包装一个可调用对象、关联一张 future，被调用时执行并把结果填进 future：

#include <future>
#include <thread>
 
int task_fn(int x) { return x * x; }
 
void run_packaged() {
    std::packaged_task<int(int)> task(task_fn); // 打包成「带凭证的任务」
    std::future<int> f = task.get_future();     // 先拿到关联的凭证
    std::thread t(std::move(task), 7);          // 把 task 搬到另一个线程去跑
    t.join();                                    // 先完成线程所有权收尾
    int result = f.get();                        // 再取值或重抛任务异常
}

packaged_task 本身可移动（std::move 搬进线程、存进容器都行），这正是它比 async 灵活的地方——你能把一堆 task 排进一个队列，让线程池里的工作线程一个个取出来跑。调用 task(7) 就像调用普通函数，只是返回值被悄悄存进了 f。

跨线程传值/传异常：promise + set_value

如果「结果」不是某个函数的返回值，而是在一段逻辑中途才算出来，用 std::promise 手动摆上台：

#include <future>
#include <thread>
 
void worker(std::promise<int> p) {
    try {
        int v = compute_something();
        p.set_value(v);          // 手动把值摆上取餐台 → 配对的 future 就绪
    } catch (...) {
        p.set_exception(std::current_exception()); // 出错就把异常摆上台
    }
}
 
void caller() {
    std::promise<int> p;
    std::future<int> f = p.get_future();   // 先从 promise 取出配对的凭证
    std::thread t(worker, std::move(p));   // promise 搬进工作线程
    t.join();
    int result = f.get();                  // 在线程已收尾后取值或重抛异常
}

promise 和它的 future 是一对：set_value / set_exception 在写端提交，get() 在读端取。示例先 join 再 get，避免 get() 重抛时跳过线程收尾。若生产者路径既未提交值也未提交异常，promise 析构会让 future 以 broken_promise 就绪。

带超时等待 + shared_future

不想无限阻塞，就用带超时的等待：

#include <chrono>
#include <future>
 
void poll(std::future<int>& f) {
    // 等最多 100 毫秒；到点还没就绪就先去干别的，下次再来等
    if (f.wait_for(std::chrono::milliseconds(100)) ==
        std::future_status::ready) {
        int v = f.get();   // 已就绪，安全取
        use(v);
    } else {
        do_other_work();   // 超时：别傻等，回头再 wait_for
    }
}

wait_for 返回一个 std::future_status：ready（已就绪）/ timeout（超时）/ deferred（任务被推迟、还没跑）。condition_variable 的 wait_for(lk, dur, pred) 同理——配谓词时返回 pred() 的结果（true 表示条件成立、false 表示超时仍不成立）。

多个线程要等同一个结果时，普通 future（get() 只能调一次）不够用，改 shared_future：

std::promise<Config> p;
std::shared_future<Config> sf = p.get_future().share(); // 转成可共享的凭证
 
// 把 sf 拷给多个工作线程，每个都能 get() 到同一份 Config
auto t1 = std::thread([sf] { auto c = sf.get(); use(c); });
auto t2 = std::thread([sf] { auto c = sf.get(); use(c); });
p.set_value(loaded_config);
t1.join();
t2.join();

future::share() 把一张普通凭证转成可复印的 shared_future，每个线程各持一份副本、都能 get() 到那同一个结果——像同一张取餐凭证复印多份发给一桌人，每人凭副本都能取到那道菜。

容易踩的坑

小结

• 条件变量 wait 时释放锁并睡眠（不忙等），生产者 notify_one/notify_all 把它叫醒——配一把 mutex + unique_lock 用
• cv.wait(lk, pred) 必须带谓词：睡前查一次挡丢失唤醒、醒来再查一次挡虚假唤醒，等价于 while(!pred()) wait(lk);
• std::async + future：一句话派异步任务、换回「取餐凭证」，get() 阻塞取回结果（任务的异常也照样重新抛出）；策略 launch::async（必并发）和 launch::deferred（推迟到 get 才在本线程跑）差别很大
• packaged_task 把任务打包成可搬运/延后执行的对象（线程池常用），promise 让你在任意线程手动 set_value/set_exception 跨线程传值或传异常
• 带超时用 wait_for/wait_until，多方等同一结果用 shared_future；C++17 的 future 没有标准 .then，latch / barrier 是 C++20 才标准化的阶段工具

int consume() {
    std::unique_lock<std::mutex> lk(m);
    cv.wait(lk);          // ❌ 不带谓词：被唤醒就往下走
    int v = q.front();    // 队列可能仍是空的！
    q.pop();
    return v;
}

int consume() {
    std::unique_lock<std::mutex> lk(m);
    cv.wait(lk, [] { return !q.empty(); }); // ✅ 谓词为假就接着睡，醒来再查
    int v = q.front();                      // 走到这里队列一定非空
    q.pop();
    return v;
}

#include <condition_variable>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <utility>
 
template <typename T>
class BoundedBuffer {
    std::queue<T> q;
    size_t cap;
    std::mutex m;
    std::condition_variable not_full;   // 等「不满」的铃（生产者等）
    std::condition_variable not_empty;  // 等「非空」的铃（消费者等）
public:
    explicit BoundedBuffer(size_t c) : cap(c) {
        if (cap == 0) throw std::invalid_argument("capacity must be positive");
    }
 
    void put(T x) {
        std::unique_lock<std::mutex> lk(m);
        not_full.wait(lk, [&] { return q.size() < cap; }); // 满则等
        q.push(std::move(x));
        lk.unlock();
        not_empty.notify_one();         // 通知一个等「非空」的消费者
    }
 
    T take() {
        std::unique_lock<std::mutex> lk(m);
        not_empty.wait(lk, [&] { return !q.empty(); });    // 空则等
        T v = std::move(q.front());
        q.pop();
        lk.unlock();
        not_full.notify_one();          // 通知一个等「不满」的生产者
        return v;
    }
};