C++11 异步编程四件套详解
UU 10.6k 字 · 约 39 分钟 C++11 异步编程四件套详解
底层概览：内部操作围绕着「共享状态」
这四个组件不是并列关系，而是层层包装的关系。它们都围绕一个看不见的核心——共享状态（Shared State）工作：
┌─────────────────────────────────────────────────────┐
│ 共享状态（堆上的一块内存） │
│ 包含：结果值 / 异常 / 就绪标志 │
└──────────────┬──────────────────────┬───────────────┘
│ 写入端（生产者） │ 读取端（消费者）
▼ ▼
promise future::get()
packaged_task future::wait()
async future::valid()- 共享状态是堆上的一块匿名内存 (control block)，存着「结果还没好 / 已经好了（值或异常）」这个状态机，内部保证线程安全。
- promise 和 packaged_task 是写端：负责把结果塞进共享状态。
- future 是读端：负责从共享状态取出结果，如果还没好就阻塞等待。
- async 是最上层的一键封装：自动创建共享状态 + 自动启动线程执行 + 自动返回 future。
一、std::promise + std::future：最底层的「生产者-消费者」通道
本质
promise 是结果的写端，future 是结果的读端。两者在创建时通过 promise::get_future() 绑定到同一块共享状态。
适用场景
你已经有一个线程（可能是第三方库创建的、可能是你手动 std::thread 启动的），你想在这个线程里算出一个结果，然后让主线程拿到它——但你无法修改线程启动方式，只能在线程函数内部投递结果。
完整代码
#include <iostream>
#include <thread>
#include <future>
#include <chrono>
// 线程函数：接收一个 promise，计算完毕后把结果 set 进去
void compute(std::promise<int> prom) {
std::cout << "[Worker] 开始计算，3秒后出结果...\n";
std::this_thread::sleep_for(std::chrono::seconds(3));
// 把结果写入共享状态（只能写一次，写第二次会抛 future_error）
prom.set_value(42);
std::cout << "[Worker] 结果已投递\n";
}
int main() {
// ① 创建一个 promise（生产者端）
std::promise<int> prom;
// ② 从 promise 拿到对应的 future（消费者端）
std::future<int> fut = prom.get_future();
// ③ 启动线程，把 promise move 进去（promise 不可拷贝）
std::thread t(compute, std::move(prom));
// ④ 主线程在这里阻塞等待，直到 worker 调用 set_value
std::cout << "[Main] 等待结果中...\n";
int result = fut.get(); // 阻塞约3秒
std::cout << "[Main] 拿到结果: " << result << '\n';
t.join();
return 0;
}关键细节
- prom.set_value(42) 调用后，共享状态的「就绪标志」翻转为 true，所有在 fut.get() 上阻塞的线程被唤醒。
- 如果线程函数里抛异常，可以用 prom.set_exception(std::current_exception()) 把异常也塞进共享状态，fut.get() 会重新抛出这个异常。
- promise 禁止拷贝操作，只能 move——共享状态只有一个写端。
- set_value() / set_exception() 只能调用一次，第二次会抛 std::future_error（错误码 promise_already_satisfied）。
注意：get_future() 只能调一次
对同一个 promise（或 packaged_task）调用第二次 get_future() 会抛 std::future_error，错误码 future_already_retrieved。也就是说读端和写端是对称的一次性绑定：写端只能 set 一次，读端只能取 future 一次。要把结果给多个消费者，请用第五节的 shared_future，而不是反复 get_future()。
std::promise<int> prom;
std::future<int> f1 = prom.get_future();
// std::future<int> f2 = prom.get_future(); // 抛 future_error(future_already_retrieved)注意：broken_promise —— 最常见的崩溃来源
如果 promise 在还没调用 set_value / set_exception 之前就被销毁（典型场景：worker 线程提前 return、抛了未捕获的异常、或任务被取消），它会在析构时把共享状态标记为 broken。此时对面 fut.get() 不会返回，而是抛出 std::future_error，错误码为 broken_promise。
// worker 里如果直接 return 而没有 set_value：
// void compute(std::promise<int> prom) { return; } // prom 析构 -> broken_promise
// 消费端必须这样兜底：
try {
int r = fut.get();
} catch (const std::future_error& e) {
if (e.code() == std::future_errc::broken_promise)
std::cerr << "对方没给结果就挂了\n";
else
std::cerr << "其他 future 错误: " << e.what() << '\n';
}经验法则：worker 线程里要么 set_value、要么 set_exception，绝不能两个都不做就退出。最稳妥的写法是用 try/catch 把整个函数体包起来，异常分支统一 set_exception。
二、std::future 读端的硬约束：get / wait / valid
这几个约束是文档最容易忽略、又最容易写出 bug 的地方，单独列一节。
| 成员 | 调用次数 | 行为 |
|---|---|---|
| get() | 只能调一次 | 阻塞到就绪；把结果 move 出来；调用后 future 变为 invalid（valid() 返回 false）。第二次 get() 是未定义行为或抛异常。 |
| wait() | 可多次 | 只阻塞等待就绪，不取值、不消费，调用后 future 仍然 valid。 |
| wait_for(d) / wait_until(t) | 可多次 | 带超时等待，返回 future_status（ready / timeout / deferred），同样不消费。 |
| valid() | 任意 | 返回该 future 是否仍关联一块共享状态。被 move 走、或 get() 之后，会变成 false。 |
std::future<int> fut = /* ... */;
fut.wait(); // OK，可以反复调用，只等待不取值
int a = fut.get(); // 第一次 get：拿到值，fut 变为 invalid
// int b = fut.get();// fut 已 invalid，行为未定义
std::cout << fut.valid(); // get 之后输出 false
std::future<int> fut2 = std::move(fut); // move 后 fut.valid() == false要点：get() 是一次性消费，想多次读同一结果用 shared_future；只想等不等值用 wait()；想超时轮询用 wait_for()。
三、std::packaged_task：把函数「打包」成带 future 的可调用对象
本质
packaged_task = 一个可调用对象（函数/lambda）+ 一个内置的 promise。你调用它时，它自动执行函数，把返回值塞进内置的 promise，然后你就能通过 get_future() 拿到 future。
和 promise 的区别
| | promise | packaged_task |
|---|---|---|
| 谁来写入结果 | 你手动 set_value() | 执行完函数后自动把返回值塞进去 |
| 适用场景 | 结果不是函数返回值，而是你在某处算出来的 | 你有一个函数，只想把它的返回值异步拿回来 |
| 灵活性 | 高：可以在任意时刻、任意地点 set | 低：只能和绑定的函数绑定在一起 |
完整代码
#include <iostream>
#include <thread>
#include <future>
#include <chrono>
// 一个普通的同步函数
int heavy_compute(int x, int y) {
std::cout << "[Worker] 开始计算 " << x << " + " << y << "...\n";
std::this_thread::sleep_for(std::chrono::seconds(2));
return x + y;
}
int main() {
// ① 打包函数：模板参数写函数的签名
std::packaged_task<int(int, int)> task(heavy_compute);
// ② 在任务执行前，先拿到 future（get_future 同样只能调一次）
std::future<int> fut = task.get_future();
// ③ 把 task move 到线程里执行（task 本身是可调用对象）
std::thread t(std::move(task), 10, 20);
// ④ 主线程可以干别的...
std::cout << "[Main] 任务已交给线程，主线程继续做别的事\n";
// ⑤ 需要结果时再 get（会阻塞到线程执行完毕）
std::cout << "[Main] 等待结果...\n";
int result = fut.get();
std::cout << "[Main] 结果: " << result << '\n';
t.join();
return 0;
}进阶用法：延迟执行 / 线程池
packaged_task 的强大之处在于：它只是包装，不决定在哪执行。你可以：
std::packaged_task<int()> task([]{ return 42; });
std::future<int> fut = task.get_future();
// 方式1：当前线程同步执行（future 立刻就绪）
task();
// 方式2：丢给线程池
// thread_pool.enqueue(std::move(task));
// 方式3：丢给新线程
std::thread t(std::move(task));
t.detach();四、std::async：一行代码搞定异步调用
本质
async = 自动创建 packaged_task + 自动选择线程执行 + 自动返回 future。它是最高层的封装。
完整代码
#include <iostream>
#include <future>
#include <chrono>
int heavy_compute(int x) {
std::this_thread::sleep_for(std::chrono::seconds(2));
return x * 2;
}
int main() {
// 一行搞定：启动异步任务 + 返回 future
std::future<int> fut = std::async(std::launch::async, heavy_compute, 21);
std::cout << "[Main] 任务已启动，做别的事...\n";
// 需要时取结果（阻塞等待）
int result = fut.get();
std::cout << "[Main] 结果: " << result << '\n';
return 0;
}启动策略和自动推导
返回值的类型推导： std: : async 会自动推导返回值类型，而且会在异步任务完成后把结果存储在std: :future 对象里。这种类型推导可以让代码更加简洁，不需要明确指定返回值类型。
// 策略1：强制新线程（和 std::thread 行为类似，但多了 future）
auto f1 = std::async(std::launch::async, func);
// 策略2：延迟执行（lazy），直到你调用 fut.get() 或 fut.wait() 时才在当前线程同步执行
auto f2 = std::async(std::launch::deferred, func);
// 策略3：系统决定（默认），可能新线程也可能延迟执行
auto f3 = std::async(func); // 等价于 std::launch::async | std::launch::deferred注意：async 的任务的生命周期问题
异步任务的生命周期与std: :future 对象绑定。如果future 对象被销毁，那么异步任务也会被取消。所以，确保future 对象的生命周期覆盖任务的执行时间。
// 错误写法：返回的 future 被丢弃！
std::async(std::launch::async, heavy_compute, 21); // 临时 future 被销毁
// 后果：析构函数会阻塞，直到任务完成！
// 原因：std::async 返回的 future 析构时会隐式 join
// 这意味着上面的代码本质上是同步的，完全失去了异步的意义正确写法：必须把 future 存起来。
析构阻塞的精确范围
future 析构会阻塞这件事只对 std::async 创建的那块共享状态成立，因为只有它内部封装了thread，自动join等待结果阻塞线程。
promise / packaged_task 因为需要手动启动thread，产生的 future 析构时绝不阻塞和 join。
// 会阻塞：async 的 future 析构隐式 join
{
std::async(std::launch::async, heavy_compute, 21); // 析构 = 同步等待
}
// 循环踩坑
{
for (int i = 0; i < 10; ++i)
std::async(std::launch::async, work, i); // 每次迭代末尾临时 future 析构 → 串行！
}
// 你以为启动了 10 个并行任务，实际是跑完一个再跑下一个
// 正确做法
{
vector<std::future<int>> futs;
for (int i = 0; i < 10; ++i)
futs.push_back(std::async(std::launch::async, work, i));
}
// 不会阻塞：packaged_task 的 future 析构什么也不做
{
std::packaged_task<int(int)> task(heavy_compute);
auto fut = task.get_future();
// fut 析构：不阻塞；但 task 若未执行会触发 broken_promise（见第二节）
}记忆：只有 async 的 future 析构会隐式 join；promise / packaged_task 的 future 析构不阻塞。
并发工程实践
虽然std: : async提供了一种简单实用的并发机制，但在实际应用中，你可能还需要使用其他并发容器如std: :mutex、 std: :lock_guard、 std: :atomic等来处理复杂的共享数据访问问题。
五、std::shared_future：一个生产者、多个消费者
前面所有示例都是一个写端、一个读端。但真实场景经常是多个消费者等待同一个结果（例如多个线程都要等同一个配置加载完成）。这时把普通 future 转成 std::shared_future：
- shared_future 可以拷贝，可以多线程同时 get()，每个调用者都拿到值的拷贝（对左值则拿到 const 引用）。
- 由 std::future::share() 得到；也可以直接把 future move 构造进 shared_future。
#include <future>
#include <thread>
#include <vector>
#include <iostream>
int load_config() { /* 耗时初始化 */ return 100; }
int main() {
std::future<int> fut = std::async(std::launch::async, load_config);
// 转成 shared_future：可被多个线程共享读取
std::shared_future<int> sf = fut.share(); // fut 之后变 invalid
std::vector<std::thread> ts;
for (int i = 0; i < 4; ++i) {
ts.emplace_back([&sf, i] {
int cfg = sf.get(); // 每个线程都能 get 一次，拿到同一份值
std::cout << "thread " << i << " got " << cfg << '\n';
});
}
for (auto& t : ts) t.join();
}要点：普通 future 是独占、一次性的读端；shared_future 是可共享、可重复读的读端。需要广播一个结果给多方时用它，而不是反复 get_future()（那是非法的）。
六、四种不同使用方式的完整实践案例
模拟一个真实场景：主线程派发多个独立任务 → 工作线程各自计算 → 主线程收集结果。分别用四种方式实现同一件事，对比写法差异。
#include <iostream>
#include <vector>
#include <thread>
#include <future>
#include <numeric>
#include <chrono>
// 模拟耗时任务
int task(int id) {
std::this_thread::sleep_for(std::chrono::milliseconds(500));
return id * 10;
}
int main() {
const int N = 5;
// ===== 方式1：async（最简，推荐日常使用）=====
{
std::vector<std::future<int>> futs;
futs.reserve(N);
for (int i = 0; i < N; ++i)
futs.push_back(std::async(std::launch::async, task, i));
int sum = 0;
for (auto& f : futs) sum += f.get();
std::cout << "async 结果: " << sum << '\n';
}
// ===== 方式2：packaged_task + thread =====
{
std::vector<std::future<int>> futs;
std::vector<std::thread> threads;
futs.reserve(N); threads.reserve(N);
for (int i = 0; i < N; ++i) {
std::packaged_task<int(int)> pt(task);
futs.push_back(pt.get_future());
threads.emplace_back(std::move(pt), i);
}
int sum = 0;
for (auto& f : futs) sum += f.get();
for (auto& t : threads) t.join();
std::cout << "packaged_task 结果: " << sum << '\n';
}
// ===== 方式3：promise + thread（手写投递 + 异常兜底）=====
{
std::vector<std::future<int>> futs;
std::vector<std::thread> threads;
futs.reserve(N); threads.reserve(N);
auto worker = [](int id, std::promise<int> p) {
try {
p.set_value(task(id)); // promise 按值传入（move 进 lambda）
} catch (...) {
p.set_exception(std::current_exception()); // 防止 broken_promise
}
};
for (int i = 0; i < N; ++i) {
std::promise<int> prom;
futs.push_back(prom.get_future());
threads.emplace_back(worker, i, std::move(prom));
}
int sum = 0;
for (auto& f : futs) {
try { sum += f.get(); }
catch (const std::future_error& e) { std::cerr << "任务异常: " << e.what() << '\n'; }
}
for (auto& t : threads) t.join();
std::cout << "promise 结果: " << sum << '\n';
}
// ===== 方式4：future 的其他成员（wait_for / valid）=====
{
auto f = std::async(std::launch::async, task, 99);
while (f.valid()) {
auto st = f.wait_for(std::chrono::milliseconds(100));
if (st == std::future_status::ready) {
std::cout << "结果就绪: " << f.get() << '\n';
break;
} else if (st == std::future_status::timeout) {
std::cout << "还没好，先干点别的...\n";
}
}
}
}注：方式3 中 promise 必须使用移动语意传进 worker（std::move(prom)），lambda 不能按值捕获 promise，否则编译失败；worker 内 try/catch 保证异常分支也 set_exception，避免对面读到 broken_promise。另外 sum += f.get() 是串行等待——第一个任务最慢时，后面早就绪的任务也得干等，生产环境可先 wait_for(0s) 过滤已就绪的，或用 C++20 when_all。
七、一张表总结选型
| 需求 | 选谁 |
|---|---|
| 只要异步拿个返回值，懒得管线程 | std::async（记得存 future） |
| 要把任务丢进线程池 / 自己控制何时何地执行 | packaged_task |
| 结果不是函数返回值（回调、事件、中途算出的值） | promise |
| 一个结果要广播给多个消费者 | shared_future（fut.share()） |
| 只需要等待、不需要值 | std::future\ 或直接用条件变量 |
| 要等多个任务中任意一个完成 | C++20 std::when_any / 自己封装 |
| 要等全部完成并汇总 | C++20 std::when_all |
一句话记忆：async 包 packaged_task，packaged_task 内置 promise，promise 和 future 共享一块状态；读端默认独占且一次性，要共享就 share()。
八、五个常见异常速查（future_error 错误码）
| 错误码 | 触发条件 | 常见场景 |
|---|---|---|
| broken_promise | promise 析构时从未 set_value / set_exception | worker 提前 return 或抛未捕获异常 |
| future_already_retrieved | 对同一 promise / packaged_task 调第二次 get_future() | 想给多个消费者 → 应改用 shared_future |
| promise_already_satisfied | set_value / set_exception 调了第二次 | 写端重复投递 |
| no_state | 在没有关联共享状态的 promise / future 上操作 | move 走之后又用原对象；二次 get() |
| 二次 get() | 同一 future get() 两次 | 第二次属未定义行为或抛异常，应改用 shared_future |
编译提示：以上均需 -pthread；std::when_all / std::when_any 为 C++20 特性，各标准库实现进度不一。