C++20 多线程简介

1. 假设条件

• 新建项目 (Newproject)
• 使用 C++20 编译器和库

2. 为什么要使用多线程

选择并发模型（Concurrency Model）通常基于两个核心理由：

1. 可扩展性（Scalability）
   • 当我们希望程序能够在多核处理器上加速运行时，需要并行化任务。
2. 关注点分离（Separation of Concerns）
   • 当我们希望不同任务在逻辑上独立、在后台运行时，可以使用多线程。

3. 可扩展性与 Amdahl 定律

如果目标是 程序加速，则 Amdahl 定律告诉我们最大加速比：
$$S=\frac{1}{(1-p)+\frac{p}{n}}$$

• (S) = 最大加速倍数
• § = 可并行化的程序比例（0~1）
• (n) = 处理器数量

示例计算

1. (p = 0.9), (n = 1000):
   $$S\approx 9.91$$
2. (p = 0.9), (n = 100,000):
   $$S\approx 9.999$$
3. (p = 0.99), (n = 100,000):
   $$S\approx 99.9$$
   结论：即使处理器数量极大，如果程序的可并行比例 § 不高，加速效果也有限。

4. 并行算法（Parallel Algorithms）

C++20 标准库提供了 并行版本的算法：

std::vector<MyData> data = ...;
// 并行排序
std::sort(std::execution::par, data.begin(), data.end(), MyComparator{});

优化建议

• 尽量合并连续的并行算法调用：

std::transform(std::execution::par, ...);
std::reduce(std::execution::par, ...);

可改为：

std::transform_reduce(std::execution::par, ...);

5. 独立任务（Independent Tasks）

• 将工作拆分为 独立任务
• 使用线程池（thread pool）执行：

thread_pool tp;
void foo() {
    execute(tp, []{ do_work(); });
    execute(tp, []{ do_other_work(); });
}

注：标准库目前没有提供统一线程池，需要使用第三方或自定义实现。

6. 关注点分离（Separation of Concerns）

• 不追求极限性能，而是逻辑上把大任务放在 后台并行运行
• 每个长任务使用独立线程：

std::jthread gui{[]{ run_gui(); }};
std::jthread printing{[]{ do_printing(); }};

• std::jthread 是 C++20 新增的，支持 自动 join，更安全。

总结

目的	模型选择	实现方式
可扩展性	高性能并行	并行算法、独立任务拆分、线程池
关注点分离	逻辑独立、后台任务	独立线程、std::jthread

• Amdahl 定律是衡量多线程可扩展性的核心公式
• 标准库并行算法和 线程池/独立线程是主要实现手段
• C++20 对多线程支持更加安全（std::jthread、并行算法）

C++20 多线程：启动与管理线程

1. 协作式取消（Cooperative Cancellation）

• 动机：
  • GUI 应用中经常有 “Cancel” 按钮，用于停止长时间运行的操作。
  • 但是，即使没有 GUI，也可能需要停止后台任务。
  • 强制停止线程是不安全的，会导致资源泄漏或不一致状态。
• C++20 提供：
  • std::stop_source
  • std::stop_token
    这两个类用于实现协作式取消。

核心理念：取消操作是 协作式的，任务必须定期检查 stop_token，否则取消请求不会生效。

2. 协作取消的使用步骤

1. 创建 std::stop_source：

std::stop_source source;

2. 从 stop_source 获取 stop_token：

std::stop_token token = source.get_token();

3. 将 stop_token 传递给新线程或任务：

std::jthread t([token]{ stoppable_func(token); });

4. 请求停止操作：

source.request_stop();

5. 在线程中定期检查 stop_requested()：

void stoppable_func(std::stop_token st) {
    while (!st.stop_requested()) {
        do_stuff();
    }
}

• 注意：如果任务不检查 stop_requested()，取消请求不会生效。

3. 协作取消示例

// 可取消任务
void stoppable_func(std::stop_token st) {
    while (!st.stop_requested()) {
        do_stuff();
    }
}
// 触发停止请求
void stopper(std::stop_source source) {
    while (!done()) {
        do_something();
    }
    source.request_stop(); // 请求取消
}

4. 自定义取消机制（Custom Cancellation）

• 可以使用 std::stop_callback 实现自己的取消逻辑，例如取消异步 I/O 操作：

Data read_file(std::stop_token st, std::filesystem::path filename) {
    auto handle = open_file(filename);
    // 在 stop_token 被请求停止时执行取消操作
    std::stop_callback cb(st, [&]{ cancel_io(handle); });
    return read_data(handle); // 阻塞式读取
}

思路：stop_callback 会在 request_stop() 被调用时触发，你可以在回调中执行资源释放或取消操作。

总结

特性	用法	注意事项
协作取消	std::stop_source + std::stop_token	任务必须主动检查 stop_requested()
自定义取消	std::stop_callback	可在回调中实现 I/O 或其他操作的取消
线程安全	std::jthread 自动 join，避免资源泄漏	不要强制终止线程
核心理念：取消是协作式的，线程必须自己配合停止，否则无法安全中断。

#include <iostream>
#include <thread>
#include <chrono>
#include <stop_token>   // C++20 协作取消相关头文件
#include <filesystem>   // 文件系统头，用于模拟文件操作
// =============================
// 模拟一个长时间运行的任务
// =============================
void long_running_task(std::stop_token st) {
    int count = 0;
    // 循环直到收到停止请求
    while (!st.stop_requested()) {
        std::cout << "Task running: step " << count++ << std::endl;
        // 模拟任务工作量，每 500 毫秒一次
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }
    // 收到停止请求后的清理操作
    std::cout << "Task received stop request. Cleaning up..." << std::endl;
}
// =============================
// 模拟自定义取消操作（例如取消文件读取）
// =============================
void custom_io_task(std::stop_token st, const std::filesystem::path& filename) {
    auto handle = filename;  // 模拟打开文件句柄
    // std::stop_callback：当 stop_token 被请求停止时触发回调
    std::stop_callback cb(st, [&] {
        std::cout << "Cancelling I/O on file: " << handle << std::endl;
        // 这里可以放真正的取消 I/O 的代码
    });
    int i = 0;
    // 循环读取文件，每次读取前检查是否收到停止请求
    while (!st.stop_requested()) {
        std::cout << "Reading file: step " << i++ << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(300)); // 模拟读取延迟
    }
    std::cout << "Finished custom_io_task" << std::endl;
}
// =============================
// 主函数
// =============================
int main() {
    std::cout << "Starting cooperative cancellation example...\n";
    // 创建 stop_source 对象，用于控制取消
    std::stop_source source;
    // 从 stop_source 获取 stop_token
    // 线程或任务会使用这个 token 来检查是否需要停止
    std::stop_token token = source.get_token();
    // 启动线程，并传递 stop_token
    // std::jthread 会在作用域结束时自动 join，不需要手动 join
    std::jthread worker(long_running_task, token);
    std::jthread io_worker(custom_io_task, token, std::filesystem::path("example.txt"));
    // 主线程模拟一些工作，等待 3 秒
    std::this_thread::sleep_for(std::chrono::seconds(3));
    // 请求停止所有任务
    std::cout << "Requesting stop..." << std::endl;
    source.request_stop();  // 触发 stop_token 的停止请求
    // jthread 会在作用域结束时自动 join 所有线程
    std::cout << "All tasks requested to stop. Exiting main.\n";
    return 0;
}

https://godbolt.org/z/qf4rxExhx

注释说明

1. std::stop_source：用于发起取消请求的源对象。
2. std::stop_token：线程/任务持有它来检测取消请求。
3. std::stop_callback：在取消请求发出时执行自定义动作（例如取消 I/O）。
4. std::jthread：C++20 新线程类，自动 join，支持协作取消。
5. 循环体中通过 st.stop_requested() 定期检查是否需要停止，这是“协作取消”的核心。

管理线程

• 在 99% 的情况下，推荐使用 std::jthread 来管理线程。
• 当你需要获取线程执行结果时，可以考虑使用 std::async。
• std::thread 只有在没有其他选择时才使用，因为它没有自动取消和 join 的机制。

std::jthread 概览

• 创建一个 std::jthread 对象会立即启动一个新线程：

std::jthread t{my_func, arg1, arg2};

• 如果函数定义为：void my_func(std::stop_token st, int arg1, int arg2)
  → 新线程会自动传入一个 stop_token 作为第一个参数。
• 如果函数没有 stop_token 参数，则直接调用 my_func(arg1, arg2)。

std::jthread 的基本 API

• std::jthread()
  → 创建一个空对象，没有启动线程。
• std::jthread x{Callable, Args...}
  → 隐式创建一个 std::stop_source，启动线程执行 Callable(src.get_token(), Args...) 或 Callable(Args...)。
• 析构函数
  → 自动调用 src.request_stop() 并等待线程结束（自动 join）。
• x.get_id()
  → 获取线程 ID。
• x.join()
  → 等待线程结束（可手动调用，也可以依赖析构函数自动 join）。

线程可移动 & 容器存储

• std::jthread 是一个线程句柄，可以移动（movable）：
  → 所有权可以转移给其他对象。
  → 可以存储在容器中，例如：std::vector<std::jthread>。

可调用对象与参数

• 可调用对象（Callable）和参数会被复制到新线程的本地存储：
  → 避免悬空引用或竞态条件。
  → 如果需要传引用，请使用 std::ref 或 lambda。

析构函数语义

void thread_func(std::stop_token st, std::string arg1, int arg2){
    while(!st.stop_requested()){
        do_stuff(arg1,arg2);
    }
}
void foo(std::string s){
    std::jthread t(thread_func, s, 42);
    do_stuff();
} // 析构函数会请求停止并 join

• 析构函数会自动：
  1. 调用 stop_source.request_stop()
  2. 等待线程结束

协作取消与 std::jthread

• std::jthread 与 std::stop_token 集成，支持 协作取消。
• 创建线程时，std::jthread 会隐式创建一个 std::stop_source。
• 从 source.get_token() 获取的 stop_token 可以作为线程函数的第一个可选参数传入。
• 线程必须自行定期检查 stop_token，否则无法停止线程。

std::jthread 的取消 API

假设有线程对象：

std::jthread x{some_callable};

• x.get_stop_source()
  → 获取线程的 stop_source
• x.get_stop_token()
  → 获取线程的 stop_token
• x.request_stop()
  → 等价于 x.get_stop_source().request_stop()
  总结
• std::jthread 自动管理线程生命周期，自动 join，支持协作取消。
• stop_token 是协作取消的核心，需要线程函数主动检查。
• 避免手动管理 std::thread，大部分场景下使用 std::jthread 更安全。

C++20 并发同步机制（Synchronization Facilities） 的讲义内容。

一、为什么需要同步（Synchronization Facilities）

1⃣ 多线程共享状态的必要性

“Most multithreaded programs need to share state between threads.”
在多线程程序中，不同线程往往需要共享一些数据，比如：

• 共享任务队列
• 共享统计结果
• 共享标志变量（如 stop_flag）
  因此就存在多个线程同时读写同一内存位置的情况。

二、Data Race（数据竞争）

“Unsynchronized access to a memory location from more than one thread, where at least one thread is writing.”
数据竞争 指的是：

• 两个或更多线程同时访问同一个内存位置；
• 其中至少一个线程执行写操作；
• 并且没有使用同步机制（比如锁、原子操作等）。
  结果：未定义行为 (undefined behavior)
  这意味着程序可能崩溃、输出错误、甚至看似“偶尔”出错。
  所以我们需要同步机制来防止数据竞争。

三、C++ 提供的同步设施

名称	用途
std::latch	一次性同步多个线程（单次计数器）
std::barrier	可多次复用的同步屏障（多阶段循环同步）
std::future	线程间任务结果传递
std::mutex	互斥锁（保护共享资源）
std::semaphore	信号量（控制并发数量）
std::atomic	原子操作（轻量同步）

四、std::latch（一次性倒计时同步器）

概念：

std::latch 是一种一次性计数器。
当计数减到 0 时，它会唤醒所有等待的线程，之后就不能重置了。

工作流程：

1. 创建 latch（初始计数非零）
2. 若干线程执行任务，每次完成后 count_down()
3. 其他线程调用 wait() 等待
4. 当计数为 0 时，所有等待线程被唤醒

latch 示例

#include <latch>
#include <thread>
#include <vector>
#include <iostream>
#include <optional>
struct MyData {
    int value;
};
MyData make_data(int i) { return {i * 10}; }
void process_data(const std::vector<std::optional<MyData>>& data) {
    for (auto& d : data)
        if (d) std::cout << "处理结果：" << d->value << "\n";
}
void foo() {
    unsigned const thread_count = 4;
    std::latch done(thread_count);
    std::vector<std::optional<MyData>> data(thread_count);
    std::vector<std::jthread> threads;
    for (unsigned i = 0; i < thread_count; ++i) {
        threads.emplace_back([&, i] {
            data[i] = make_data(i);
            done.count_down();     // 任务完成，减少计数
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        });
    }
    done.wait();  // 等待所有线程都 count_down()
    process_data(data);
}
int main() {
    foo();
}

执行逻辑：

• 每个线程完成自己的工作后调用 count_down()。
• 主线程调用 wait()，直到所有线程都结束。

五、在测试中使用 latch 同步线程

在多线程单元测试中，latch 很有用：

1. 设置测试数据
2. 创建一个 std::latch
3. 启动多个线程，每个线程一开始执行：

   test_latch.arrive_and_wait();
   

4. 等所有线程都到达后，同时开始执行测试逻辑
   这样可避免某些线程“抢跑”，让测试同步启动。

六、std::barrier（可重复使用的屏障）

概念：

std::barrier 是一种可重置的同步屏障，通常用于循环并行任务。
每一轮（phase）：

• 所有线程都调用 arrive_and_wait()
• 当计数归零：
  • 调用“完成函数” (completion function)
  • 自动重置计数，进入下一轮同步

API 概览

函数	作用
std::barrier(count, completion)	创建 barrier，带有计数与完成函数
x.arrive()	减少计数但不等待
x.wait(token)	等待完成
x.arrive_and_wait()	同时到达并等待
x.arrive_and_drop()	永久减少计数（线程退出时用）

barrier 示例

#include <barrier>
#include <iostream>
#include <thread>
#include <vector>
#include <chrono>
void finish_task() {
    std::cout << "所有线程到达屏障，执行完成函数。\n";
}
int main() {
    const unsigned num_threads = 3;
    std::barrier b(num_threads, finish_task);
    auto worker = [&](int id) {
        for (int round = 0; round < 3; ++round) {
            std::cout << "[线程 " << id << "] 执行第 " << round << " 轮任务\n";
            std::this_thread::sleep_for(std::chrono::milliseconds(200 * id));
            b.arrive_and_wait();  // 等待其他线程
        }
    };
    std::vector<std::jthread> threads;
    for (unsigned i = 0; i < num_threads; ++i)
        threads.emplace_back(worker, i);
    return 0;
}

输出示例：

[线程 0] 执行第 0 轮任务
[线程 1] 执行第 0 轮任务
[线程 2] 执行第 0 轮任务
所有线程到达屏障，执行完成函数。
[线程 0] 执行第 1 轮任务
...

说明：

• 每一轮循环结束时，所有线程都必须到达屏障；
• 当最后一个线程到达时，finish_task() 被执行；
• 然后 barrier 重置，进入下一轮。

七、小结

同步机制	用途	是否可重用
std::latch	一次性同步多个线程	否
std::barrier	多轮循环同步	是
std::mutex	互斥访问资源	是
std::future	异步任务结果同步	是
std::atomic	轻量级同步原子操作	是

C++20 并发与同步的高级总结章节，内容涵盖了所有主要同步原语：

Futures、Promises、Async、Mutex、Condition Variables、Semaphores、Atomics。
我来帮你系统地梳理和讲解这一整块内容，附带解释 + 小例子。

C++20 并发同步机制总结与例子

一、Futures（期物）

概念

std::future 提供了 跨线程单次数据传递 的机制。
有三种创建方式：

方式	含义
std::async	自动创建线程并返回 future
std::promise	手动创建 future 并从外部设置值
std::packaged_task	将函数封装为异步任务
所有这些都会生成一个 std::future<T> 用于获取结果。

示例 1：std::async

#include <future>
#include <iostream>
int compute(int x) {
    return x * x;
}
int main() {
    // 启动异步任务
    auto f = std::async(std::launch::async, compute, 10);
    // 等待结果
    std::cout << "结果: " << f.get() << "\n";  // 输出 100
}

std::async 会自动启动一个线程来运行 compute
f.get() 会阻塞直到任务结束

示例 2：std::promise + std::future

#include <future>
#include <iostream>
#include <thread>
int main() {
    std::promise<int> prom;
    std::future<int> fut = prom.get_future();
    std::thread producer([&prom] {
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        prom.set_value(42);
    });
    std::cout << "等待结果...\n";
    std::cout << "结果: " << fut.get() << "\n";
    producer.join();
}

std::promise 创建了一个“数据通道”
std::future 用于从另一个线程获取值

示例 3：异常传递

#include <future>
#include <iostream>
#include <thread>
#include <stdexcept>
int main() {
    std::promise<int> prom;
    auto fut = prom.get_future();
    std::thread t([&prom] {
        try {
            throw std::runtime_error("错误：计算失败");
        } catch (...) {
            prom.set_exception(std::current_exception());
        }
    });
    try {
        std::cout << fut.get() << "\n";
    } catch (const std::exception& e) {
        std::cout << "捕获异常: " << e.what() << "\n";
    }
    t.join();
}

示例 4：共享结果 — std::shared_future

#include <future>
#include <iostream>
#include <thread>
int main() {
    std::promise<int> prom;
    std::shared_future<int> fut = prom.get_future().share();
    std::jthread t1([fut] { std::cout << "T1: " << fut.get() << "\n"; });
    std::jthread t2([fut] { std::cout << "T2: " << fut.get() << "\n"; });
    prom.set_value(100);
}

std::future 只能 get() 一次
std::shared_future 可以多次读取同一结果

二、Mutex（互斥锁）

概念

Mutex（互斥量）用于保证在任意时刻只有一个线程可以访问某段代码（临界区）。

示例：使用 std::scoped_lock

#include <mutex>
#include <iostream>
#include <thread>
int data = 0;
std::mutex data_mutex;
void add_to_data(int delta) {
    std::scoped_lock lock(data_mutex); // 自动加锁、析构时解锁
    data += delta;
}
int main() {
    std::jthread t1(add_to_data, 5);
    std::jthread t2(add_to_data, 7);
}

死锁示例与解决

错误版本（死锁）：

std::scoped_lock lock_from(from.m);
std::scoped_lock lock_to(to.m);

正确版本（避免死锁）：

std::scoped_lock locks(from.m, to.m);

std::scoped_lock 会同时锁住多个互斥量，自动避免死锁。

💤 三、Condition Variable（条件变量）

概念

用于在数据未准备好时阻塞等待，并在数据到来时被通知。
避免了“忙等（busy wait）”浪费 CPU。

示例：

#include <condition_variable>
#include <mutex>
#include <optional>
#include <thread>
#include <iostream>
std::mutex m;
std::condition_variable cond;
std::optional<int> data;
void producer() {
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    {
        std::scoped_lock lock(m);
        data = 42;
    }
    cond.notify_one();  // 通知等待者
}
void consumer() {
    std::unique_lock lock(m);
    cond.wait(lock, [] { return data.has_value(); });
    std::cout << "收到数据: " << *data << "\n";
}
int main() {
    std::jthread t1(consumer);
    std::jthread t2(producer);
}

wait(lock, predicate)：等待直到条件为真
notify_one()：唤醒一个等待线程

忙等的坏处

while (true) {
    std::scoped_lock lock(m);
    if (data.has_value()) break;
}

• 持续占用 CPU
• 浪费能耗
• 响应慢
  ➡ 应该用条件变量来等待。

示例：支持取消等待（C++20）

#include <condition_variable>
#include <mutex>
#include <optional>
#include <thread>
#include <iostream>
std::mutex m;
std::condition_variable_any cond;
std::optional<int> data;
void consumer(std::stop_token st) {
    std::unique_lock lock(m);
    if (!cond.wait(lock, st, [] { return data.has_value(); }))
        return; // 被取消
    std::cout << "收到: " << *data << "\n";
}
int main() {
    std::jthread t(consumer);
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    t.request_stop();  // 取消等待
}

四、Semaphores（信号量）

概念

信号量表示“可用资源的数量”。
当计数为 0 时，新线程必须等待。
C++20 引入：

std::counting_semaphore<max_count>
std::binary_semaphore  // 等价于 counting_semaphore<1>

示例：

#include <semaphore>
#include <thread>
#include <iostream>
std::counting_semaphore<5> slots(5); // 最多允许 5 个并发线程
void task(int id) {
    slots.acquire();  // 占用一个 slot
    std::cout << "线程 " << id << " 进入临界区\n";
    std::this_thread::sleep_for(std::chrono::milliseconds(500));
    slots.release();  // 释放 slot
}
int main() {
    std::vector<std::jthread> threads;
    for (int i = 0; i < 10; ++i)
        threads.emplace_back(task, i);
}

同时最多 5 个线程进入 do_stuff()
其余线程等待 slots.release()

五、Atomics（原子操作）

概念

原子操作是最底层的同步机制。
类型写作 std::atomic<T>。
特点：

• 对 T 的读写是不可分割的（atomic）
• 可用于计数、标志、轻量同步
• 不一定“无锁”，但大多数平台上的整数是 lock-free 的

示例：

#include <atomic>
#include <thread>
#include <iostream>
std::atomic<int> counter = 0;
void worker() {
    for (int i = 0; i < 1000; ++i)
        counter.fetch_add(1, std::memory_order_relaxed);
}
int main() {
    std::jthread t1(worker);
    std::jthread t2(worker);
    t1.join(); t2.join();
    std::cout << "最终计数: " << counter << "\n"; // 输出 2000
}

六、总结表格

场景	推荐机制	说明
启动任务并等待结果	std::async / std::future	自动线程 + 结果传递
线程间传递数据	std::promise + std::future	手动控制值/异常
简单线程管理	std::jthread	自动停止 + join
临界区保护	std::mutex + std::scoped_lock	避免同时访问
数据等待	std::condition_variable	等待通知
限制并发数量	std::counting_semaphore	信号量
高性能计数/标志	std::atomic	轻量同步
多线程同步点	std::latch / std::barrier	单次或循环同步