C++ 并发编程完全指南：从 std::thread 到 C++26 Executors 🚀

2026-05-29 C/C++ 1 次阅读

一篇覆盖 C++ 并发编程全图谱的实战教程——从基础线程管理到协程与 Sender/Receiver 模型，帮你从「能用」进阶到「精通」。全文万字深度解析，附完整可运行代码。

背景与概念：为什么要学并发编程
核心基石：std::thread 与线程管理
同步与互斥：锁、原子操作与内存序
异步编程：Future、Promise 与 async
协程：C++20 的异步革命
并发数据结构：从屏障到无锁队列
C++26 Executors：Sender/Receiver 模型前瞻
实战：手写一个工业级线程池
常见问题 FAQ
总结与展望

一、背景与概念：为什么要学并发编程

1.1 摩尔定律的尽头，并发才是未来

曾几何时，提升 CPU 性能只需要提高主频。但物理定律的围墙——功耗墙和散热墙——让「单核狂奔」走到了尽头。今天的 CPU 已经不再追求更高的单核频率，而是堆砌更多的核心。即便是你手里那台轻薄本，也可能有 8 核 16 线程；服务器端的 EPYC 和 Xeon 更是轻松破百核。

这意味着一个冷酷的事实：不学并发编程，你的程序就只能用上 CPU 八分之一的算力。对于 C++ 开发者来说，这个问题的紧迫性更加突出——游戏引擎、高频交易系统、数据库内核、浏览器渲染引擎，这些 C++ 的传统强项，无一不是并发的重度使用者。

并发编程不是可选的「加分项」，而是现代 C++ 开发者必须掌握的「生存技能」。

1.2 C++ 并发模型的演进之路

C++ 的并发支持经历了从「一无所有」到「应有尽有」的华丽蜕变。理解这段历史有助于你把握整个体系的设计脉络：

C++ 版本	并发特性	里程碑意义
C++98/03	无标准支持	依赖平台 API（pthread / Win32），代码不可移植
C++11	`std::thread`, `std::mutex`, `std::atomic`, `std::future`, `std::condition_variable`	🎉 并发正式进入标准库，C++ 有了统一的内存模型
C++14	`std::shared_lock`, `std::shared_timed_mutex`	读写锁支持，读多写少场景性能飞跃
C++17	`std::scoped_lock`, `std::shared_mutex`, 并行算法 (`std::execution::par`)	更安全的锁管理 + 数据并行初探
C++20	协程 (`co_await`/`co_yield`/`co_return`), `std::jthread`, `std::latch`, `std::barrier`, `std::counting_semaphore`	🚀 真正的异步革命 + 线程协作原语爆发
C++26	`std::execution` Sender/Receiver 模型 (P2300)	统一的异步编程范式，终结碎片化

💡 关键洞察：C++ 的并发演进不是「打补丁」，而是在逐步构建一套完整的异步编程范式。从 C++11 的「裸线程」到 C++26 的「结构化并发」，背后是对「正确性」和「可组合性」的不懈追求。当你理解了这一脉络，每个新特性的设计动机就变得一目了然。

1.3 并发 vs 并行：一个必须澄清的概念

很多开发者把「并发」和「并行」混为一谈，但这其实是两个不同的概念：

维度	并发 (Concurrency)	并行 (Parallelism)
定义	多个任务交替执行（逻辑上同时）	多个任务同时执行（物理上同时）
硬件要求	单核即可	需要多核
典型场景	响应用户交互的同时处理后台任务	矩阵乘法分块多核计算
C++ 工具	协程、异步 IO	`std::thread`、`std::execution::par`

简单说：并发是结构，并行是执行。一个设计良好的并发程序可以在单核上正确运行（只是慢），但一个没有并发结构的程序永远无法利用多核。

1.4 本文阅读指南

目标读者：有 C++ 基础，想系统学习并发编程的开发者
前置知识：C++11 基础语法、基本的操作系统概念（什么是线程）
你将学到：线程管理、同步机制、协程、无锁编程、线程池设计、C++26 最新进展
代码环境：所有示例基于 C++17/20，使用 GCC 12+ 或 Clang 16+ 编译，编译选项 -std=c++20 -pthread

图1：C++ 并发编程技术全景图

二、核心基石：std::thread 与线程管理

2.1 创建你的第一个线程

std::thread 是 C++ 并发编程的起点。它封装了操作系统原生线程，用起来比 pthread_create 友好得多——你不再需要手动管理 void* 参数和函数指针：

#include <iostream>
#include <thread>
#include <chrono>
#include <string>

void worker(int id, const std::string& message) {
    std::cout << "[线程 " << id << "] " << message << "\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    std::cout << "[线程 " << id << "] 工作完成\n";
}

int main() {
    std::thread t1(worker, 1, "处理任务A");
    std::thread t2(worker, 2, "处理任务B");
    
    // 🔴 关键：必须 join 或 detach，否则线程对象析构时调用 std::terminate
    t1.join();
    t2.join();
    
    std::cout << "所有线程已完成\n";
    return 0;
}

运行结果：

[线程 1] 处理任务A
[线程 2] 处理任务B
[线程 1] 工作完成
[线程 2] 工作完成
所有线程已完成

2.2 传递参数给线程：你可能不知道的陷阱

std::thread 的构造函数会拷贝或移动参数到新线程的存储中，然后以右值形式传递给可调用对象。这个机制有两个常见陷阱：

陷阱一：引用参数被拷贝

#include <thread>
#include <iostream>

void update(int& value) {
    value += 100;  // 你期望修改的是原始变量...
}

int main() {
    int x = 0;
    // ❌ 错误：x 被拷贝了，update 修改的是拷贝！
    std::thread t(update, x);
    t.join();
    std::cout << "x = " << x << "\n";  // 输出 0，而不是 100
    return 0;
}

正确做法——用 std::ref 包装引用参数：

std::thread t(update, std::ref(x));  // ✅ x 以引用方式传递
t.join();
std::cout << "x = " << x << "\n";  // 输出 100

陷阱二：临时对象的生命周期

void process(const std::string& str) {
    std::cout << str << "\n";
}

int main() {
    // ⚠️ 危险："临时字符串" 是 const char*，转换为 string 的时机在新线程中
    // 如果转换发生在线程上下文切换之后，可能访问已释放的指针
    std::thread t(process, "临时字符串");
    t.join();
    
    // ✅ 安全：提前构造 string
    std::thread t2(process, std::string("临时字符串"));
    t2.join();
    return 0;
}

2.3 join vs detach：你必须知道的分岔路口

这是每个 C++ 并发新手都会遇到的困惑。join() 和 detach() 分别代表了两种完全不同的线程管理哲学：

操作	行为	适用场景	风险等级
`join()`	阻塞等待线程结束	有明确的生命周期依赖	🟢 安全
`detach()`	线程与 thread 对象解绑	后台守护任务	🔴 高风险

detach 的经典陷阱——use-after-free：

#include <thread>
#include <string>
#include <chrono>

void dangerous() {
    std::string local = "我在栈上";
    
    std::thread t([&local]() {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        // 💥 此时 dangerous() 已返回，local 已析构——未定义行为！
        std::cout << local << "\n";
    });
    
    t.detach();
} // local 在此析构，但线程还在跑

🛑 规则一：除非你非常清楚自己在做什么，否则永远使用 join()。

2.4 C++20 的礼物：std::jthread

C++20 引入的 std::jthread（joining thread）解决了 std::thread 最让人头疼的两个问题：

析构时自动 join()——永远不会因为忘记 join 而触发 std::terminate
内置中断机制——通过 std::stop_token 实现协作式的线程取消

#include <thread>
#include <iostream>
#include <chrono>

void worker(std::stop_token token, int id) {
    while (!token.stop_requested()) {
        std::cout << "[Worker " << id << "] 工作中...\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    std::cout << "[Worker " << id << "] 收到停止请求，善后处理...\n";
    // 这里可以做清理工作
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    std::cout << "[Worker " << id << "] 优雅退出\n";
}

int main() {
    {
        std::jthread t1(worker, 1);
        std::jthread t2(worker, 2);
        std::this_thread::sleep_for(std::chrono::milliseconds(300));
        // 离开作用域时自动请求停止并 join，不需要手动操作
    }
    std::cout << "所有 jthread 已安全销毁\n";
    return 0;
}

std::stop_token 的设计理念值得学习——它是协作式的，线程自己决定何时检查停止信号，如何做清理。这比 pthread_cancel 那种异步杀死线程的粗暴方式安全得多。

2.5 线程局部存储（thread_local）

当一个全局变量需要在每个线程中拥有独立副本时，用 thread_local：

#include <thread>
#include <iostream>
#include <random>

thread_local std::mt19937 rng(std::random_device{}());

void worker(int id) {
    // 每个线程有自己的随机数生成器，无需加锁
    std::uniform_int_distribution<int> dist(1, 100);
    for (int i = 0; i < 3; ++i) {
        std::cout << "[T" << id << "] " << dist(rng) << "\n";
    }
}

int main() {
    std::thread t1(worker, 1);
    std::thread t2(worker, 2);
    t1.join();
    t2.join();
    return 0;
}

📝 使用场景：随机数生成器、日志上下文、请求级别的缓存。thread_local 变量在 C++11 就可用，但直到今天仍然是被低估的特性。

2.6 线程数量的黄金法则

一个常见的问题是：「我应该开多少个线程？」

#include <thread>
#include <iostream>

unsigned int optimal_threads(bool io_bound = false) {
    unsigned int n = std::thread::hardware_concurrency();
    if (n == 0) n = 4;  // 无法检测时的保守默认
    
    if (io_bound) {
        // IO 密集型：线程多数时间在等待，可以超额分配
        return n * 2;
    } else {
        // 计算密集型：每个核心一个线程，避免上下文切换开销
        return n;
    }
}

int main() {
    std::cout << "硬件并发数: " << std::thread::hardware_concurrency() << "\n";
    std::cout << "计算密集型推荐: " << optimal_threads(false) << "\n";
    std::cout << "IO 密集型推荐: " << optimal_threads(true) << "\n";
    return 0;
}

📊 经验法则：计算密集型 = 核心数；IO 密集型 = 核心数 × 2。但这只是起点，实际调优需要用 profiling 工具测量。对于混合型负载（如 Web 服务器），可以考虑用线程池 + 动态调整。

三、同步与互斥：锁、原子操作与内存序

图2：线程同步机制流程图

3.1 数据竞争：并发世界的头号公敌

当两个线程同时访问同一块内存，且至少有一个是写操作，就发生了数据竞争（data race）。在 C++ 标准中，数据竞争 = 未定义行为，编译器可以做任何事——包括优化掉你的代码。

#include <thread>
#include <iostream>

int counter = 0;  // 共享变量，没有任何保护

void increment() {
    for (int i = 0; i < 100000; ++i) {
        ++counter;  // 💥 数据竞争！三条汇编指令不是原子的
        // 实际上 counter++ 是: load → add → store
        // 两个线程的指令可能交错执行，导致丢失更新
    }
}

int main() {
    std::thread t1(increment);
    std::thread t2(increment);
    t1.join();
    t2.join();
    std::cout << "counter = " << counter << "\n";  // 几乎不会是 200000
    return 0;
}

运行 10 次，你可能得到 10 个不同的结果。这就是并发最可怕的地方：bug 不是必现的，但它就在那里。

3.2 互斥锁全家桶

C++ 标准库提供了丰富的锁机制，覆盖了从基础到高级的各种需求：

锁类型	特点	典型场景
`std::mutex`	基础互斥锁，不可重入	保护临界区（99% 的场景）
`std::timed_mutex`	支持超时尝试加锁	需要超时逻辑，避免死等
`std::recursive_mutex`	同一线程可重复加锁	递归函数中的临界区保护
`std::shared_mutex` (C++17)	读写锁，多读单写	读多写少的缓存场景

3.3 RAII 锁守卫：永远不要手动 unlock

C++ 的锁管理遵循 RAII 原则——获取资源即初始化，离开作用域即释放。这能彻底杜绝「忘了 unlock」的 bug：

#include <mutex>
#include <vector>
#include <thread>
#include <iostream>

class ThreadSafeCounter {
    mutable std::mutex mtx;  // mutable 允许在 const 函数中加锁
    int value = 0;
    
public:
    void increment() {
        // lock_guard: 构造时 lock，析构时 unlock
        // 即使中间抛出异常，也能保证解锁
        std::lock_guard<std::mutex> lock(mtx);
        ++value;
    }
    
    void add(int n) {
        // unique_lock 更灵活：可以延迟锁定、提前解锁、转移所有权
        std::unique_lock<std::mutex> lock(mtx);
        value += n;
        // 可以在这里提前 unlock，释放锁去做其他事
        lock.unlock();
        // ... 做一些不需要锁的操作 ...
    }
    
    int get() const {
        std::lock_guard<std::mutex> lock(mtx);
        return value;
    }
};

lock_guard vs unique_lock 的选择指南：

特性	`std::lock_guard`	`std::unique_lock`
大小	最小（1 个指针）	较大（有额外状态）
灵活性	不可 unlock / relock	可随时 unlock、relock
所有权转移	❌ 不可移动	✅ 可移动（配合条件变量）
使用建议	优先使用，满足 90% 需求	需要灵活控制锁生命周期时用

3.4 条件变量：线程间的信号通信

条件变量是解决「等待某个条件成立」的经典工具。它让线程高效休眠，直到被其他线程唤醒：

#include <mutex>
#include <condition_variable>
#include <queue>
#include <thread>
#include <iostream>

template<typename T>
class BoundedQueue {
    std::queue<T> queue_;
    mutable std::mutex mtx_;
    std::condition_variable not_empty_;
    std::condition_variable not_full_;
    size_t capacity_;
    
public:
    explicit BoundedQueue(size_t cap) : capacity_(cap) {}
    
    void push(T item) {
        std::unique_lock<std::mutex> lock(mtx_);
        // 等待直到队列不满，防止虚假唤醒所以要循环检查
        not_full_.wait(lock, [this] { return queue_.size() < capacity_; });
        queue_.push(std::move(item));
        // 解锁后通知消费者（在解锁后通知性能更好）
        lock.unlock();
        not_empty_.notify_one();
    }
    
    T pop() {
        std::unique_lock<std::mutex> lock(mtx_);
        not_empty_.wait(lock, [this] { return !queue_.empty(); });
        T item = std::move(queue_.front());
        queue_.pop();
        lock.unlock();
        not_full_.notify_one();
        return item;
    }
};

⚠️ 条件变量的三大陷阱：

必须配合 mutex 使用——条件变量的 wait 操作会原子地 unlock mutex 并休眠

必须用 while/带谓词的 wait 防止虚假唤醒——操作系统可能无故唤醒线程

notify 在 unlock 之前还是之后——推荐在 unlock 之后调用 notify，避免「匆忙等待」

3.5 死锁与 std::scoped_lock

多锁场景下，死锁（deadlock）是最常见的并发 bug：

// ❌ 死锁示例
void transfer_bad(Account& a, Account& b, int amount) {
    std::lock_guard lock_a(a.mtx);  // 线程1: 锁定 A
    std::lock_guard lock_b(b.mtx);  // 等待 B...
    // 线程2 同时: 锁定 B，等待 A → 互相等待，死锁！
}

// ✅ C++17 安全方案：std::scoped_lock 原子锁定多个 mutex
void transfer_good(Account& a, Account& b, int amount) {
    std::scoped_lock lock(a.mtx, b.mtx);  // 使用死锁避免算法同时锁两个
    a.balance -= amount;
    b.balance += amount;
}

3.6 std::call_once：保证只执行一次

单例模式、惰性初始化等场景的标准方案：

#include <mutex>
#include <memory>

class DatabaseConnection {
    static std::unique_ptr<DatabaseConnection> instance_;
    static std::once_flag init_flag_;
    
    DatabaseConnection() { /* 初始化连接池 */ }
    
public:
    static DatabaseConnection& get() {
        std::call_once(init_flag_, [] {
            instance_.reset(new DatabaseConnection());
        });
        return *instance_;
    }
};

std::call_once 比「双检锁」模式更安全、更简单——它完全消除了指令重排带来的隐患。

3.7 原子操作：无锁并发的基石

std::atomic 提供了无需互斥锁的线程安全操作。对于简单的计数器、标志位，原子操作比锁快 10 到 100 倍：

#include <atomic>
#include <thread>
#include <iostream>
#include <vector>

std::atomic<int> counter{0};

void increment_many() {
    for (int i = 0; i < 100000; ++i) {
        // fetch_add 是原子操作，保证每个线程看到正确的值
        counter.fetch_add(1, std::memory_order_relaxed);
    }
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back(increment_many);
    }
    for (auto& t : threads) t.join();
    
    std::cout << "counter = " << counter << "\n";  // 永远是 400000
    return 0;
}

原子操作的优势不仅在于性能——它们还是无锁数据结构的基础构建块。

3.8 内存序：并发编程的终极挑战

std::memory_order 是 C++ 并发模型中最晦涩也最强大的部分。它控制原子操作之间的可见性顺序：

内存序	保证	相对性能	典型场景
`memory_order_seq_cst`	全局顺序一致性（最强）	1x（基准）	默认，适合绝大多数场景
`memory_order_acquire/release`	获取-释放语义	~1.5x	生产者-消费者模式
`memory_order_relaxed`	仅保证原子性（最弱）	~3x	简单计数器，无同步需求

生产者-消费者模式的 acquire/release：

#include <atomic>
#include <thread>
#include <string>
#include <iostream>

std::atomic<bool> ready{false};
std::string data;  // 非原子变量，但被原子操作保护

void producer() {
    data = "重要的数据 (大小: 1MB)";
    // release: 保证在此之前的写操作（data="..."）对后续的 acquire 可见
    ready.store(true, std::memory_order_release);
}

void consumer() {
    // acquire: 保证在此之后的读操作能看见 release 之前的所有写入
    while (!ready.load(std::memory_order_acquire))
        ; // 自旋等待
    std::cout << "收到: " << data << "\n";  // 一定能看到 "重要的数据"
}

⚠️ 重要提醒：除非你在做性能关键的低延迟系统（如高频交易），否则使用默认的 seq_cst 就足够了。过早优化内存序是万恶之源。记住：先正确，再优化。

四、异步编程：Future、Promise 与 async

4.1 std::async：最简单的异步入口

std::async 让你只需一行代码就能将任务抛到后台，然后在需要结果时获取：

#include <future>
#include <iostream>
#include <chrono>

int heavy_computation(int n) {
    std::cout << "开始计算 " << n << "...\n";
    std::this_thread::sleep_for(std::chrono::seconds(1));  // 模拟耗时计算
    int result = n * n;
    std::cout << "计算完成: " << result << "\n";
    return result;
}

int main() {
    // 启动异步任务
    std::future<int> result = std::async(std::launch::async, heavy_computation, 42);
    
    // 主线程可以做其他事——这就是异步的价值
    std::cout << "主线程继续工作...\n";
    
    // get() 会阻塞直到结果就绪
    int value = result.get();
    std::cout << "最终结果: " << value << "\n";
    return 0;
}

4.2 launch 策略：async vs deferred

std::async 有三种启动策略，理解它们的区别至关重要：

策略	行为	何时使用
`std::launch::async`	立即在新线程中执行	真正的并发需求
`std::launch::deferred`	延迟到调用 `get()`/`wait()` 时才在当前线程执行	懒计算，避免线程开销
`async \| deferred` (默认)	由标准库实现决定	🚫 不推荐——行为不确定

📝 生产建议：永远显式指定 std::launch::async，除非你非常清楚 deferred 就是你需要的。默认策略的行为在 GCC 和 MSVC 之间可能不同。

4.3 Promise-Future 通信：一对一的结果传递

std::promise 和 std::future 是一对搭档——Promise 是「写端」（生产者设置结果），Future 是「读端」（消费者获取结果）：

#include <future>
#include <thread>
#include <iostream>

void download_file(std::promise<std::string>&& prom) {
    try {
        std::cout << "开始下载...\n";
        std::this_thread::sleep_for(std::chrono::seconds(2));
        prom.set_value("下载完成：file.zip (128MB, MD5: a1b2c3)");
    } catch (...) {
        // 将异常传递给 future，消费者可以通过 get() 重新抛出
        prom.set_exception(std::current_exception());
    }
}

int main() {
    std::promise<std::string> prom;
    std::future<std::string> fut = prom.get_future();
    
    std::thread t(download_file, std::move(prom));
    
    std::cout << "等待下载...\n";
    try {
        std::string result = fut.get();  // 阻塞直到得到结果
        std::cout << result << "\n";
    } catch (const std::exception& e) {
        std::cout << "下载失败: " << e.what() << "\n";
    }
    
    t.join();
    return 0;
}

4.4 std::shared_future：一对多的结果广播

std::future 只能被 get() 一次（移动语义）。如果需要多个线程等待同一个结果，用 std::shared_future：

#include <future>
#include <thread>
#include <iostream>
#include <vector>

int main() {
    std::promise<int> prom;
    std::shared_future<int> sf = prom.get_future().share();
    
    std::vector<std::thread> consumers;
    for (int i = 0; i < 3; ++i) {
        consumers.emplace_back([sf, i]() {
            std::cout << "消费者 " << i << " 等待...\n";
            int value = sf.get();  // 每个线程都能 get()
            std::cout << "消费者 " << i << " 收到: " << value << "\n";
        });
    }
    
    prom.set_value(42);  // 广播给所有消费者
    
    for (auto& t : consumers) t.join();
    return 0;
}

4.5 std::packaged_task：包装可调用对象

std::packaged_task 将任意可调用对象包装成一个「自带 future」的异步任务：

#include <future>
#include <thread>
#include <iostream>

int main() {
    // 包装一个 lambda
    std::packaged_task<int(int, int)> task([](int a, int b) {
        return a + b;
    });
    
    std::future<int> result = task.get_future();
    
    // 在新线程中执行
    std::thread t(std::move(task), 10, 20);
    
    std::cout << "10 + 20 = " << result.get() << "\n";  // 30
    t.join();
    return 0;
}

std::packaged_task 是实现线程池的核心组件——它把「任务」和「返回值」捆绑在一起，是 void() 风格的 std::function 无法做到的。

五、协程：C++20 的异步革命

5.1 协程是什么？为什么需要它？

传统异步编程最大的痛点是什么？不是性能，而是可读性和可维护性。回调地狱（callback hell）和手写状态机让异步代码比同步代码复杂一个数量级。

来看看同一个异步读文件的逻辑，回调和协程的对比：

回调风格（难以阅读和维护）：

void read_config(const std::string& path,
                 std::function<void(std::string)> on_success,
                 std::function<void(std::error_code)> on_error) {
    async_read(path, [on_success, on_error](auto data) {
        async_parse(data, [on_success, on_error](auto config) {
            async_validate(config, [on_success, on_error](auto valid) {
                on_success(valid);
            }, on_error);
        }, on_error);
    }, on_error);
}
// 每增加一步就多一层嵌套，错误处理分散在各处

协程风格（线性逻辑，清晰直观）：

Task<std::string> read_config(const std::string& path) {
    auto data = co_await async_read(path);
    auto config = co_await async_parse(data);
    auto valid = co_await async_validate(config);
    co_return valid;
}
// 同步风格的书写方式，编译时自动转换为状态机

这就是协程的魔力——用写同步代码的方式写异步逻辑。编译器帮你把线性的代码自动转换成高效的状态机，同时保留了直观的错误处理（try-catch 照样能用）。

5.2 C++20 协程三大关键字

关键字	含义	函数类型	典型场景
`co_await`	挂起协程，等待异步操作完成	异步任务	等待 IO、网络、计算
`co_yield`	挂起并产出一个值，可恢复	生成器	惰性序列、流处理
`co_return`	返回最终值，协程结束	任何协程	返回异步计算结果

📝 C++ 协程的一大特点是无栈协程（stackless coroutine）——协程的状态存储在堆上而不是栈上。这意味着协程的开销很低（约等于一次 new），但局部变量不能跨挂起点保存（需要在 promise_type 中管理）。

5.3 手写一个协程 Task 类型

C++20 只提供了协程的「底层原语」——编译器可以处理 co_await/co_yield/co_return，但没有提供像 Python asyncio 那样的高层框架。你需要自己定义 promise_type 和 awaitable。

这是一个最小但完整的 Task 实现：

#include <coroutine>
#include <iostream>
#include <thread>
#include <exception>

template<typename T>
struct Task {
    struct promise_type {
        T value;
        std::exception_ptr exception;
        
        Task get_return_object() {
            return Task{std::coroutine_handle<promise_type>::from_promise(*this)};
        }
        std::suspend_never initial_suspend() { return {}; }  // 创建后立即开始
        std::suspend_never final_suspend() noexcept { return {}; }  // 结束后自动清理
        void return_value(T v) { value = std::move(v); }
        void unhandled_exception() { exception = std::current_exception(); }
    };
    
    std::coroutine_handle<promise_type> handle;
    
    explicit Task(std::coroutine_handle<promise_type> h) : handle(h) {}
    ~Task() { if (handle) handle.destroy(); }
    
    // 禁止拷贝，允许移动
    Task(const Task&) = delete;
    Task& operator=(const Task&) = delete;
    Task(Task&& other) noexcept : handle(std::exchange(other.handle, nullptr)) {}
    
    T get() {
        if (handle.promise().exception)
            std::rethrow_exception(handle.promise().exception);
        return std::move(handle.promise().value);
    }
};

// 一个简单的异步等待器——在后台线程中恢复协程
struct AsyncOp {
    bool await_ready() { return false; }  // 不立即就绪，需要挂起
    void await_suspend(std::coroutine_handle<> h) {
        std::thread([h]() {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            h.resume();  // 在后台线程中恢复协程
        }).detach();
    }
    std::string await_resume() { return "异步操作结果"; }
};

Task<std::string> process_data() {
    auto data = co_await AsyncOp{};
    co_return "处理后的: " + data;
}

5.4 生成器模式：co_yield 实战

co_yield 让你能写出像 Python 生成器一样优雅的惰性序列。这在处理大数据流时特别有用——不需要一次性将所有数据加载到内存：

#include <coroutine>
#include <iostream>

template<typename T>
struct Generator {
    struct promise_type {
        T current_value;
        
        std::suspend_always yield_value(T v) {
            current_value = std::move(v);
            return {};  // 挂起，把控制权还给调用者
        }
        std::suspend_always initial_suspend() { return {}; }
        std::suspend_always final_suspend() noexcept { return {}; }
        Generator get_return_object() {
            return Generator{
                std::coroutine_handle<promise_type>::from_promise(*this)
            };
        }
        void return_void() {}
        void unhandled_exception() { std::terminate(); }
    };
    
    std::coroutine_handle<promise_type> handle;
    
    // 迭代器支持——可以直接用 range-for
    struct iterator {
        std::coroutine_handle<promise_type> handle;
        bool operator!=(std::default_sentinel_t) const {
            return !handle.done();
        }
        iterator& operator++() {
            handle.resume();
            return *this;
        }
        T operator*() const { return handle.promise().current_value; }
    };
    
    iterator begin() { handle.resume(); return {handle}; }
    std::default_sentinel_t end() { return {}; }
    
    ~Generator() { if (handle) handle.destroy(); }
};

// 使用协程生成斐波那契数列
Generator<int> fibonacci(int n) {
    int a = 0, b = 1;
    for (int i = 0; i < n; ++i) {
        co_yield a;
        int next = a + b;
        a = b;
        b = next;
    }
}

int main() {
    for (int f : fibonacci(10)) {
        std::cout << f << " ";  // 0 1 1 2 3 5 8 13 21 34
    }
    std::cout << "\n";
    return 0;
}

生成器的美妙之处在于：每次迭代时才计算下一个值，而不是一次性生成整个数列。当你需要生成前 100 万个素数时，这种惰性计算能节省大量内存。

5.5 协程的实战使用模式

在生产环境中，协程最常见的三种使用模式：

模式	协程关键字	典型库	场景
异步任务	`co_await` + `co_return`	ASIO, cppcoro	网络 IO、文件 IO
生成器	`co_yield`	ranges-v3, cppcoro	惰性序列、流处理
异步生成器	`co_await` + `co_yield`	ASIO (实验性)	分块读取流数据

六、并发数据结构：从屏障到无锁队列

6.1 C++20 新增的同步原语

C++20 一口气引入了四个新的同步设施，填补了多年来的空白：

设施	功能	可重用	典型场景
`std::latch`	一次性倒计时计数器	❌	等待 N 个线程完成初始化
`std::barrier`	可重用的线程屏障，支持分阶段	✅	迭代式并行计算的分阶段同步
`std::counting_semaphore`	轻量级计数信号量（0~max）	✅	限流、有界队列、资源池
`std::binary_semaphore`	互斥信号量（只有 0/1）	✅	轻量互斥、事件通知

6.2 std::latch：等待所有线程就绪

std::latch 是最简单的线程同步原语——像一个只能往下数的计数器：

#include <latch>
#include <thread>
#include <vector>
#include <iostream>

void worker(int id, std::latch& init_latch, std::latch& done_latch) {
    // 第一阶段：每个线程独立初始化
    std::cout << "Worker " << id << " 初始化中...\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(id * 50));
    std::cout << "Worker " << id << " 初始化完成\n";
    init_latch.count_down();  // 通知：我准备好了
    
    init_latch.wait();  // 阻塞，直到所有 worker 就绪
    
    // 所有线程同时开始工作——真正的「起跑线」
    std::cout << "Worker " << id << " 开始处理！\n";
    done_latch.count_down();
}

int main() {
    const int N = 5;
    std::latch init_latch(N);
    std::latch done_latch(N);
    
    std::vector<std::thread> workers;
    for (int i = 0; i < N; ++i) {
        workers.emplace_back(worker, i, std::ref(init_latch), std::ref(done_latch));
    }
    
    done_latch.wait();
    std::cout << "所有 worker 任务完成\n";
    for (auto& t : workers) t.join();
    return 0;
}

6.3 std::barrier：分阶段同步

std::barrier 与 latch 类似，但可以重复使用——适合迭代式的并行算法：

#include <barrier>
#include <thread>
#include <vector>
#include <iostream>
#include <random>

void phase_worker(int id, std::barrier<>& sync) {
    std::mt19937 rng(id);
    for (int phase = 0; phase < 3; ++phase) {
        int work = std::uniform_int_distribution<int>(10, 50)(rng);
        std::cout << "[T" << id << "] Phase " << phase 
                  << " working " << work << "ms\n";
        std::this_thread::sleep_for(std::chrono::milliseconds(work));
        
        // 到达屏障，等待所有线程完成当前阶段
        sync.arrive_and_wait();
        std::cout << "[T" << id << "] Phase " << phase << " done!\n";
    }
}

int main() {
    const int N = 4;
    std::barrier sync(N);
    std::vector<std::thread> threads;
    for (int i = 0; i < N; ++i) {
        threads.emplace_back(phase_worker, i, std::ref(sync));
    }
    for (auto& t : threads) t.join();
    return 0;
}

6.4 信号量：轻量级的并发控制

C++20 的信号量比条件变量更轻量——没有 mutex 的束缚，语义更纯粹：

#include <semaphore>
#include <thread>
#include <vector>
#include <iostream>

// 限制同时只有 3 个线程访问数据库
std::counting_semaphore<3> db_connections{3};

void query_database(int id) {
    db_connections.acquire();  // 获取一个连接槽位
    std::cout << "[查询 " << id << "] 获得数据库连接\n";
    
    std::this_thread::sleep_for(std::chrono::milliseconds(200));  // 模拟查询
    
    std::cout << "[查询 " << id << "] 释放连接\n";
    db_connections.release();  // 归还连接槽位
}

int main() {
    std::vector<std::thread> threads;
    for (int i = 0; i < 10; ++i) {
        threads.emplace_back(query_database, i);
    }
    for (auto& t : threads) t.join();
    return 0;
}
// 输出会显示：最多同时只有 3 个查询在执行

6.5 无锁队列：高性能并发的银弹？

对于超低延迟场景（如高频交易、游戏引擎），互斥锁的上下文切换开销可能无法接受。无锁（lock-free）数据结构使用 CAS（Compare-And-Swap）等原子操作来避免mutex：

#include <atomic>
#include <memory>

template<typename T>
class LockFreeQueue {
    struct Node {
        T data;
        std::atomic<Node*> next{nullptr};
        Node(T val) : data(std::move(val)) {}
    };
    
    std::atomic<Node*> head_;
    std::atomic<Node*> tail_;
    
public:
    LockFreeQueue() {
        Node* dummy = new Node(T{});
        head_.store(dummy, std::memory_order_relaxed);
        tail_.store(dummy, std::memory_order_relaxed);
    }
    
    void enqueue(T value) {
        Node* node = new Node(std::move(value));
        while (true) {
            Node* last = tail_.load(std::memory_order_acquire);
            Node* next = last->next.load(std::memory_order_acquire);
            
            if (last == tail_.load(std::memory_order_acquire)) {
                if (next == nullptr) {
                    // 尝试将新节点链接到队尾
                    if (last->next.compare_exchange_weak(
                            next, node, 
                            std::memory_order_release,
                            std::memory_order_relaxed)) {
                        // 尝试推进 tail 指针
                        tail_.compare_exchange_strong(
                            last, node,
                            std::memory_order_release,
                            std::memory_order_relaxed);
                        return;
                    }
                } else {
                    // 帮助推进 tail（其他线程可能卡住了）
                    tail_.compare_exchange_weak(
                        last, next,
                        std::memory_order_release,
                        std::memory_order_relaxed);
                }
            }
        }
    }
    
    bool dequeue(T& result) {
        while (true) {
            Node* first = head_.load(std::memory_order_acquire);
            Node* last = tail_.load(std::memory_order_acquire);
            Node* next = first->next.load(std::memory_order_acquire);
            
            if (first == head_.load(std::memory_order_acquire)) {
                if (first == last) {
                    if (next == nullptr) return false;  // 队列空
                    tail_.compare_exchange_weak(last, next);
                } else {
                    result = std::move(next->data);
                    if (head_.compare_exchange_weak(first, next)) {
                        delete first;
                        return true;
                    }
                }
            }
        }
    }
};

⚠️ 忠告：无锁编程是并发编程的「黑带」领域。上面的 Michael-Scott 队列实现看似简洁，但隐藏了内存序细节、ABA 问题和安全内存回收的无尽陷阱。在生产环境中，优先使用成熟的无锁库（如 boost::lockfree 或 folly::MPMCQueue）。

七、C++26 Executors：Sender/Receiver 模型前瞻

7.1 为什么需要 Executors？碎片化的现状

C++ 异步编程的现状是碎片化的。放眼望去，各大框架各有各的模型：

框架/库	异步模型	调度器
`std::async`	Future/Promise	隐式创建线程，不可控
ASIO	CompletionToken	`io_context` 事件循环
TBB	`task_group`	工作窃取线程池
libunifex	Sender/Receiver	可插拔调度器
HPX	`hpx::future`	内置调度器

这导致了什么？你在 ASIO 里写的优雅异步代码，到了 TBB 里就得重写。协程的调度策略依赖底层框架的选择。简单说：C++ 缺少一个统一的异步编程模型。

C++26 的 std::execution（P2300 提案）试图用一套统一的 Sender/Receiver 抽象终结这个乱局。它不是凭空设计的——其核心思想直接来源于 Facebook 的 libunifex 和 NVIDIA 的 stdexec，已经在生产环境中经受了大规模验证。

7.2 核心概念三部曲

P2300 模型建立在三个核心抽象之上，它们形成了异步编程的「三段论」：

[Sender] ──(调度到 Scheduler)──> [Receiver]
   |              |                   |
 描述操作       控制在哪执行        处理结果
 (懒执行)      (线程池/事件循环)    (回调风格)

概念	角色	通俗类比
Sender	描述一个异步操作，但不执行	写好的信（还没寄出）
Scheduler	控制异步操作在哪个执行上下文运行	邮局（负责投递）
Receiver	接收异步操作的结果（值/错误/取消）	收信人

关键设计原则——懒执行：Sender 只是对异步操作的「描述」。在你调用 sync_wait 或 start_detached 之前，什么都不会发生。这跟 std::async 立即创建线程的行为截然不同。

7.3 C++26 异步编程新范式（代码预览）

#include <execution>  // C++26

// 定义一个异步操作管线——像 Unix 管道一样组合异步操作
auto work = std::execution::schedule(pool)           // 在线程池上调度
          | std::execution::then([] { return 42; })   // 步骤1：计算
          | std::execution::then([](int x) {          // 步骤2：变换
                return x * x; 
            })
          | std::execution::then([](int result) {     // 步骤3：输出
                std::cout << "结果: " << result << "\n";
            });

// 提交并同步等待完成（调试用）
std::this_thread::sync_wait(std::move(work));

这个管道的威力在于：

可组合性：用管道操作符 | 串联异步操作，类比 Unix 管道和 ranges
懒执行：不提交就不运行，可以做复杂的条件分支和循环组合
调度器无关：同一个管线可以跑在不同调度器上（线程池、GPU 流、事件循环）
零开销抽象：编译期展开整个管线，运行时没有虚函数开销

7.4 错误处理与取消传播

P2300 提供了结构化的错误处理和取消机制：

auto robust_work = 
    std::execution::schedule(pool)
    | std::execution::then([] { 
        return risky_operation(); 
      })
    | std::execution::upon_error([](std::error_code ec) {
        // 错误恢复路径——返回备用值或重新抛出
        return fallback_value();
      })
    | std::execution::upon_stopped([] {
        // 取消路径——清理资源
        cleanup();
      });

取消是自动传播的：当你取消最外层的 Sender，它会层层传递到最内层的异步操作。这解决了以往异步编程中「取消地狱」的经典难题。

7.5 与现有方案的对比总结

维度	C++11 async	C++20 协程	C++26 Executors
可组合性	❌ 低	⚠️ 中（需手写 awaitable）	✅ 高（管道组合）
调度控制	❌ 不可控	⚠️ 需自定义 Scheduler	✅ 内置 Scheduler 抽象
错误处理	try-catch	co_await try-catch	`upon_error` / `let_error`
取消支持	❌ 无	⚠️ stop_token	✅ 内置取消传播
学习曲线	低	高	中高
标准化程度	✅ C++11	✅ C++20	🔜 C++26

八、实战：手写一个工业级线程池

图3：线程池架构设计图

8.1 线程池为什么重要？

如果你还记得前面说的——线程创建和销毁的开销很大——你就会理解线程池的价值。线程池通过预先创建一组工作线程并复用它们，将任务提交的成本从「创建线程」降低到「入队 + 条件变量通知」。

一个优秀的线程池需要解决以下问题：

✅ 任务提交与结果获取（Future/Promise 模型）
✅ 优雅关闭（处理完队列中剩余任务再退出）
✅ 避免惊群效应（每次只唤醒一个线程，而非全部）
✅ 可选的优先级调度（某些任务需要优先执行）

8.2 完整实现

以下是一个生产就绪的线程池实现，约 100 行代码：

#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <iostream>
#include <atomic>
#include <type_traits>

class ThreadPool {
public:
    explicit ThreadPool(size_t num_threads) : stop_(false) {
        for (size_t i = 0; i < num_threads; ++i) {
            workers_.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    
                    {
                        std::unique_lock<std::mutex> lock(queue_mutex_);
                        // 等待直到有任务或线程池停止
                        condition_.wait(lock, [this] {
                            return stop_ || !tasks_.empty();
                        });
                        
                        // 停止且没任务了 → 退出
                        if (stop_ && tasks_.empty()) {
                            return;
                        }
                        
                        task = std::move(tasks_.front());
                        tasks_.pop();
                    }
                    
                    // 执行任务——在锁外执行，允许其他线程取任务
                    task();
                }
            });
        }
    }
    
    // 提交任务，返回 future 以获取结果
    template<typename F, typename... Args>
    auto submit(F&& f, Args&&... args) 
        -> std::future<std::invoke_result_t<F, Args...>> {
        
        using return_type = std::invoke_result_t<F, Args...>;
        
        // packaged_task 将函数和返回值打包
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
        std::future<return_type> result = task->get_future();
        
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            if (stop_) {
                throw std::runtime_error("线程池已停止，无法提交新任务");
            }
            // 将 packaged_task 包装成 void() 放入队列
            tasks_.emplace([task]() { (*task)(); });
        }
        
        condition_.notify_one();  // 只唤醒一个线程，避免惊群
        return result;
    }
    
    // 查询队列中的任务数
    size_t pending_count() const {
        std::unique_lock<std::mutex> lock(queue_mutex_);
        return tasks_.size();
    }
    
    // 查询工作线程数
    size_t worker_count() const {
        return workers_.size();
    }
    
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            stop_ = true;
        }
        condition_.notify_all();  // 唤醒所有线程让它们检查 stop_
        for (std::thread& worker : workers_) {
            if (worker.joinable()) worker.join();
        }
    }
    
private:
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    mutable std::mutex queue_mutex_;
    std::condition_variable condition_;
    std::atomic<bool> stop_;
};

8.3 使用示例与性能验证

int main() {
    ThreadPool pool(8);  // 8 个工作线程
    
    std::cout << "工作线程数: " << pool.worker_count() << "\n";
    
    // 提交 20 个计算任务
    std::vector<std::future<int>> results;
    for (int i = 0; i < 20; ++i) {
        results.push_back(
            pool.submit([i]() {
                // 模拟复杂计算
                std::this_thread::sleep_for(std::chrono::milliseconds(50));
                return i * i;
            })
        );
    }
    
    std::cout << "已提交 20 个任务，等待结果...\n";
    
    // 收集结果（按提交顺序）
    for (size_t i = 0; i < results.size(); ++i) {
        std::cout << "任务[" << i << "] 结果 = " << results[i].get() << "\n";
    }
    
    std::cout << "剩余任务: " << pool.pending_count() << "\n";
    // 析构时优雅关闭
    return 0;
}

8.4 线程池设计关键决策表

决策点	我们的选择	原因
任务队列类型	单一 FIFO 队列	简单可靠，满足大部分场景
通知策略	`notify_one()`	避免惊群效应，省 CPU
关闭策略	处理完剩余任务再退出	不丢失已提交的任务
返回类型	`std::future<T>`	标准、灵活，支持异常传播
任务类型擦除	`std::function<void()>`	统一接口，支持任何可调用对象
异常处理	`std::packaged_task` 自动传播	用户端 try-catch 即可捕获

九、常见问题 FAQ

Q1: std::thread 和 std::jthread 该用哪个？

A: 新项目统一用 std::jthread。它解决了 std::thread 的三个核心痛点：析构安全（自动 join，不会 std::terminate）、可中断（stop_token 协作取消）、移动友好。仅有的例外是你在维护 C++17 及以下的老项目，或者你的第三方库接口只接受 std::thread。

Q2: 为什么我的多线程程序比单线程还慢？

A: 这个问题有三种常见原因。第一是锁竞争过重——多个线程频繁争抢同一把锁，导致大量时间花在上下文切换上。用 perf 或 VTune 找到热点锁，考虑用读写锁、无锁结构或减少临界区。第二是伪共享（false sharing）——不同线程修改同一缓存行上的不同变量，导致缓存行在 CPU 核心间反复跳跃。用 alignas(64) 或 std::hardware_destructive_interference_size 隔离变量。第三是线程创建销毁开销——用线程池复用线程，避免频繁创建。

Q3: 如何检测数据竞争？

A: 两大神器：ThreadSanitizer (TSAN)——加 -fsanitize=thread 编译，运行时自动检测数据竞争，几乎零误报，性能开销约 5-15 倍；Helgrind (Valgrind)——更慢（50-100 倍），但能检测更多同步问题，包括锁顺序错误。日常开发用 TSAN（快速反馈），CI 中用 TSAN + Helgrind 双重保险。

# 使用 TSAN
g++ -fsanitize=thread -g -O1 my_program.cpp -o my_program -pthread
./my_program  # 检测到 race 时会打印详细报告：哪两个线程、哪两行代码

Q4: 什么时候该用无锁编程？

A: 三个条件同时满足时才考虑：1) 锁已经成为 profiling 里的最大瓶颈（实测数据，不是猜测）；2) 你的团队有精通无锁编程的成员，能正确使用 memory_order；3) 你有完善的测试覆盖（包括 TSAN 持续运行）和压力测试。如果三个条件缺一，优先用互斥锁或成熟的第三方无锁库。

Q5: C++20 协程现在能用于生产吗？

A: 可以，但需要注意。三大编译器（GCC 10+、Clang 14+、MSVC 16.8+）已提供成熟的协程支持。但标准库没有提供协程框架（如 task<T>），你需要使用第三方库（如 cppcoro、ASIO 的协程支持、folly::coro），或者自己实现 promise_type。推荐从 ASIO 的协程开始——它是最成熟的 C++ 异步生态。

Q6: std::async 和线程池哪个更好？

A: 简单场景用 std::async（一行代码搞定，无需管理线程池生命周期），复杂场景用线程池（需要控制并发度、避免线程频繁创建销毁）。注意：某些标准库实现中，std::async 会为每个任务创建新线程，在高并发下性能可能很差。如果不确定，先测量再决定。

Q7: 协程中可以使用 std::mutex 吗？

A: 技术上可以，但强烈不推荐。如果一个协程持有 mutex 后在某个 co_await 上挂起，mutex 会一直被锁住，成为严重的性能瓶颈。在协程中优先使用异步友好的同步原语——无锁结构、co_await 版本的锁，或确保锁的持有时间极短（不跨 co_await）。

Q8: C++26 Executors 什么时候能用？

A: P2300 提案已在 C++26 标准中正式通过。GCC 和 Clang 的初始实现正在进行中。目前你可以通过 stdexec（NVIDIA 开源）或 libunifex（Meta 开源）在生产环境中提前使用 Sender/Receiver 模型。这两个库是 P2300 的「孵化器」，API 与最终标准高度一致。

Q9: 线程池中任务有异常怎么办？

A: std::packaged_task 会自动捕获异常并存储，当你在 future::get() 时重新抛出。所以任务内部的异常不会导致线程崩溃——它会安全地传播给调用者。

十、总结与展望

10.1 本文要点回顾

我们从 C++ 并发编程的基石出发，走过了从 std::thread 到 C++26 Executors 的完整旅程。以下是本文覆盖的技术栈全景：

层级	核心技术	典型应用场景
基础层	`std::thread`, `std::jthread`, `thread_local`	所有并发程序的起点
同步层	`std::mutex`, `std::atomic`, `memory_order`, 条件变量	数据保护、线程间通信
异步层	`std::future`, `std::promise`, `std::async`, `std::packaged_task`	任务并行、异步 IO
协程层	`co_await`, `co_yield`, 生成器, Task 类型	复杂的异步流程控制
协调层	`std::latch`, `std::barrier`, `std::counting_semaphore`	多线程协作同步
执行层	Sender/Receiver, Scheduler (C++26)	统一的异步编程范式

10.2 推荐学习路径

入门（1 周）：掌握 std::thread / std::jthread + std::mutex + 手写一个简单线程池
进阶（2 周）：深入 std::atomic 和内存序，理解 happens-before 关系，学会用 TSAN
高级（1 月）：C++20 协程——理解 promise_type 和 awaitable 机制，阅读 ASIO 协程源码
前沿（持续关注）：跟踪 P2300 提案实现进展，阅读 stdexec 源码，关注 CppCon 演讲

10.3 延伸资源

📖 书籍：《C++ Concurrency in Action (2nd Edition)》——Anthony Williams 的并发圣经，覆盖 C++17
📹 演讲：CppCon 上 Eric Niebler 和 Kirk Shoop 关于 P2300 的系列演讲
🔧 库：folly::MPMCQueue（Facebook）、stdexec（NVIDIA）、libunifex（Meta）、moodycamel::ConcurrentQueue
📝 提案：P2300 - std::execution

🎯 最后的忠告：并发编程的魅力在于它能释放硬件的全部潜力，它的危险在于一次未定义行为就能让整个程序崩溃。保持敬畏，保持学习，保持测试。用 TSAN，用 CI，用 code review。你的用户会感谢你的。

本文由 MarkShareX AI 自动创作，分类：C/C++，方向：并发编程

C++ 并发编程