# 同步操作 "同步操作"是指在计算机科学和信息技术中的一种操作方式，其中不同的任务或操作按顺序执行，一个操作完成后才能开始下一个操作。在多线程编程中，各个任务通常需要通过**同步设施**进行相互**协调和等待**，以确保数据的**一致性**和**正确性**。 本章的主要内容有： * 条件变量 * `std::future` 等待异步任务 * 在规定时间内等待 * Qt 实现异步任务的示例 * 其它 C++20 同步设施：信号量、闩与屏障 本章将讨论如何使用条件变量等待事件，介绍 future 等标准库设施用作同步操作，使用Qt+CMake 构建一个项目展示多线程的必要性，介绍 C++20 引入的新的同步设施。 ## 等待事件或条件 假设你正在一辆夜间运行的地铁上，那么你要如何在正确的站点下车呢？ 1. 一直不休息，每一站都能知道，这样就不会错过你要下车的站点，但是这会很疲惫。 2. 可以看一下时间，估算一下地铁到达目的地的时间，然后设置一个稍早的闹钟，就休息。这个方法听起来还行，但是你可能被过早的叫醒，甚至估算错误导致坐过站，又或者闹钟没电了睡过站。 3. 事实上最简单的方式是，到站的时候有人或者其它东西能将你叫醒（比如手机的地图，到达设置的位置就提醒）。 这和线程有什么关系呢？其实第一种方法就是在说”[忙等待](https://zh.wikipedia.org/wiki/%E5%BF%99%E7%A2%8C%E7%AD%89%E5%BE%85)（busy waiting）”也称“**自旋**“。 ```cpp bool flag = false; std::mutex m; void wait_for_flag(){ std::unique_lock lk{ m }; while (!flag){ lk.unlock(); // 1 解锁互斥量 lk.lock(); // 2 上锁互斥量 } } ``` 第二种方法就是加个延时，这种实现进步了很多，减少浪费的执行时间，但很难确定正确的休眠时间。这会影响到程序的行为，在需要快速响应的程序中就意味着丢帧或错过了一个时间片。循环中，休眠②前函数对互斥量解锁①，再休眠结束后再对互斥量上锁，让另外的线程有机会获取锁并设置标识（因为修改函数和等待函数共用一个互斥量）。 ```cpp void wait_for_flag(){ std::unique_lock lk{ m }; while (!flag){ lk.unlock(); // 1 解锁互斥量 std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 2 休眠 lk.lock(); // 3 上锁互斥量 } } ``` 第三种方式（也是最好的）实际上就是使用条件变量了。通过另一线程触发等待事件的机制是最基本的唤醒方式，这种机制就称为“条件变量”。 C++ 标准库对条件变量有两套实现：[`std::condition_variable`](https://zh.cppreference.com/w/cpp/thread/condition_variable) 和 [`std::condition_variable_any`](https://zh.cppreference.com/w/cpp/thread/condition_variable_any)，这两个实现都包含在 [``](https://zh.cppreference.com/w/cpp/header/condition_variable) 头文件中。 `condition_variable_any` 类是 `std::condition_variable` 的泛化。相对于只在 `std::unique_lock` 上工作的 `std::condition_variable`，`condition_variable_any` 能在任何满足[*可基本锁定(BasicLockable)*](https://zh.cppreference.com/w/cpp/named_req/BasicLockable)要求的锁上工作，所以增加了 `_any` 后缀。显而易见，这种区分必然是 `any` 版**更加通用但是却有更多的性能开销**。所以通常**首选** `std::condition_variable`。有特殊需求，才会考虑 `std::condition_variable_any`。 ```cpp std::mutex mtx; std::condition_variable cv; bool arrived = false; void wait_for_arrival() { std::unique_lock lck(mtx); cv.wait(lck, []{ return arrived; }); // 等待 arrived 变为 true std::cout << "到达目的地，可以下车了！" << std::endl; } void simulate_arrival() { std::this_thread::sleep_for(std::chrono::seconds(5)); // 模拟地铁到站，假设5秒后到达目的地 { std::lock_guard lck(mtx); arrived = true; // 设置条件变量为 true，表示到达目的地 } cv.notify_one(); // 通知等待的线程 } ``` > [运行](https://godbolt.org/z/KhhP9T45s)测试。更换为 `std::condition_variable_any` 效果[相同](https://godbolt.org/z/PY35843eE)。 * `std::mutex mtx`: 创建了一个互斥量，用于保护共享数据的访问，确保在多线程环境下的数据同步。 * `std::condition_variable cv`: 创建了一个条件变量，用于线程间的同步，当条件不满足时，线程可以等待，直到条件满足时被唤醒。 * `bool arrived = false`: 设置了一个标志位，表示是否到达目的地。 在 `wait_for_arrival` 函数中： 1. `std::unique_lock lck(mtx)`: 使用互斥量创建了一个独占锁。 2. `cv.wait(lck, []{ return arrived; })`: 阻塞当前线程，释放（unlock）锁，直到条件被满足。 3. 一旦条件满足，即 `arrived` 变为 true，并且条件变量 `cv` 被**唤醒**（包括**虚假唤醒**），那么当前线程会重新获取锁（lock），并执行后续的操作。 在 `simulate_arrival` 函数中： 1. `std::this_thread::sleep_for(std::chrono::seconds(5))`: 模拟地铁到站，暂停当前线程 5 秒。 2. 设置 `arrived` 为 true，表示到达目的地。 3. `cv.notify_one()`: 唤醒一个等待条件变量的线程。 这样，当 `simulate_arrival` 函数执行后，`arrived` 被设置为 true，并且通过 `cv.notify_one()` 唤醒了等待在条件变量上的线程，从而使得 `wait_for_arrival` 函数中的等待结束，可以执行后续的操作，即输出提示信息。 *** 条件变量的 [`wait`](https://zh.cppreference.com/w/cpp/thread/condition_variable/wait) 成员函数有两个版本，以上代码使用的就是第二个版本，传入了一个[*谓词*](https://zh.cppreference.com/w/cpp/named_req/Predicate)。 ```cpp void wait(std::unique_lock& lock); // 1 template void wait(std::unique_lock& lock, Predicate pred); // 2 ``` ②等价于： ```cpp while (!pred()) wait(lock); ``` 第二个版本只是对第一个版本的**包装**，等待并判断谓词，会调用第一个版本的重载。这可以避免“[虚假唤醒（spurious wakeup）](https://en.wikipedia.org/wiki/Spurious_wakeup)”。 > 条件变量虚假唤醒是指在使用条件变量进行线程同步时，有时候线程可能会在没有收到通知的情况下被唤醒。问题取决于程序和系统的具体实现。解决方法很简单，在循环中等待并判断条件可一并解决。使用 C++ 标准库则没有这个烦恼了。 我们也可以简单看一下 MSVC STL 的[源码实现](https://github.com/microsoft/STL/blob/22a8826/stl/inc/mutex#L555-L565)： ```cpp void wait(unique_lock& _Lck) noexcept { _Cnd_wait(_Mycnd(), _Lck.mutex()->_Mymtx()); } template void wait(unique_lock& _Lck, _Predicate _Pred) { while (!_Pred()) { wait(_Lck); } } ``` ## 线程安全的队列 在本节中，我们介将绍一个更为复杂的示例，以巩固我们对条件变量的学习。为了实现一个线程安全的队列，我们需要考虑以下两个关键点： 1. 当执行 `push` 操作时，需要确保没有其他线程正在执行 `push` 或 `pop` 操作；同样，在执行 `pop` 操作时，也需要确保没有其他线程正在执行 `push` 或 `pop` 操作。 2. 当队列为**空**时，不应该执行 `pop` 操作。因此，我们需要使用条件变量来传递一个谓词，以确保在执行 `pop` 操作时队列不为空。 基于以上思考，我们设计了一个名为 `threadsafe_queue` 的模板类，如下： ```cpp template class threadsafe_queue { mutable std::mutex m; // 互斥量，用于保护队列操作的独占访问 std::condition_variable data_cond; // 条件变量，用于在队列为空时等待 std::queue data_queue; // 实际存储数据的队列 public: threadsafe_queue() {} void push(T new_value) { { std::lock_guard lk { m }; data_queue.push(new_value); } data_cond.notify_one(); } // 从队列中弹出元素（阻塞直到队列不为空） void pop(T& value) { std::unique_lock lk{ m }; data_cond.wait(lk, [this] {return !data_queue.empty(); }); value = data_queue.front(); data_queue.pop(); } // 从队列中弹出元素（阻塞直到队列不为空），并返回一个指向弹出元素的 shared_ptr std::shared_ptr pop() { std::unique_lock lk{ m }; data_cond.wait(lk, [this] {return !data_queue.empty(); }); std::shared_ptr res { std::make_shared(data_queue.front()) }; data_queue.pop(); return res; } bool empty()const { std::lock_guard lk (m); return data_queue.empty(); } }; ``` 请无视我们省略的构造、赋值、交换、`try_xx` 等操作。以上示例已经足够。 光写好了肯定不够，我们还得测试运行，我们可以写一个经典的：”*生产者消费者模型*“，也就是一个线程 `push` ”**生产**“，一个线程 `pop` ”**消费**“。 ```cpp void producer(threadsafe_queue& q) { for (int i = 0; i < 5; ++i) { q.push(i); } } void consumer(threadsafe_queue& q) { for (int i = 0; i < 5; ++i) { int value{}; q.pop(value); } } ``` 两个线程分别运行 `producer` 与 `consumer`，为了观测运行我们可以为 `push` 与 `pop` 中增加打印语句： ```cpp std::cout << "push:" << new_value << std::endl; std::cout << "pop:" << value << std::endl; ``` **可能**的[**运行**](https://godbolt.org/z/Mh4bYGsTP)结果是： ```txt push:0 pop:0 push:1 pop:1 push:2 push:3 push:4 pop:2 pop:3 pop:4 ``` 这很正常，到底哪个线程会抢到 CPU 时间片持续运行，是系统调度决定的，我们只需要保证一开始提到的两点就行了： > `push` 与 `pop` 都只能单独执行；当队列为**空**时，不执行 `pop` 操作。 我们可以给一个简单的示意图帮助你理解这段运行结果： ```diff 初始状态：队列为空 +---+---+---+---+---+ Producer 线程插入元素 0： +---+---+---+---+---+ | 0 | | | | | Consumer 线程弹出元素 0： +---+---+---+---+---+ | | | | | | Producer 线程插入元素 1： +---+---+---+---+---+ | 1 | | | | | Consumer 线程弹出元素 1： +---+---+---+---+---+ | | | | | | Producer 线程插入元素 2： +---+---+---+---+---+ | | 2 | | | | Producer 线程插入元素 3： +---+---+---+---+---+ | | 2 | 3 | | | Producer 线程插入元素 4： +---+---+---+---+---+ | | 2 | 3 | 4 | | Consumer 线程弹出元素 2： +---+---+---+---+---+ | | | 3 | 4 | | Consumer 线程弹出元素 3： +---+---+---+---+---+ | | | | 4 | | Consumer 线程弹出元素 4： +---+---+---+---+---+ | | | | | | 队列为空，所有元素已被弹出 ``` 到此，也就可以了。 ## 使用条件变量实现后台提示音播放 一个常见的场景是：当你的软件完成了主要功能后，领导可能突然要求添加一些竞争对手产品的功能。比如领导看到了人家的设备跑起来总是有一些播报，说明当前的情况，执行的过程，或者报错了也会有提示音说明。于是就想让我们的程序也增加“**语音提示**”的功能。此时，你需要考虑如何在程序运行到不同状态时添加适当的语音播报，并且**确保这些提示音的播放不会影响其他功能的正常运行**。 为了不影响程序的流畅执行，提示音的播放显然不能占据业务线程的资源。我们需要额外启动一个线程来专门处理这个任务。 但是，大多数的提示音播放都是短暂且简单。如果每次播放提示音时都新建一个线程，且不说创建线程也需要大量时间，可能影响业务正常的执行任务的流程，就光是其频繁创建线程的开销也是不能接受的。 *** 因此，更合理的方案是：**在程序启动时，就启动一个专门用于播放提示音的线程。当没有需要播放的提示时，该线程会一直处于等待状态；一旦有提示音需要播放，线程就被唤醒，完成播放任务**。 具体来说，我们可以通过条件变量来实现这一逻辑，核心是监控一个音频队列。我们可以封装一个类型，包含以下功能： * 一个成员函数在对象构造时就启动，使用条件变量监控队列是否为空，互斥量确保共享资源的同步。如果队列中有任务，就取出并播放提示音；如果队列为空，则线程保持阻塞状态，等待新的任务到来。 * 提供一个外部函数，以供在需要播放提示音的时候调用它，向队列添加新的元素，该函数需要通过互斥量来保护数据一致性，并在成功添加任务后唤醒条件变量，通知播放线程执行任务。 > 这种设计通过合理利用**条件变量**和**互斥量**，不仅有效减少了 CPU 的无效开销，还能够确保主线程的顺畅运行。它不仅适用于提示音的播放，还能扩展用于其他类似的后台任务场景。 我们引入 [SFML](https://github.com/SFML/SFML) 三方库进行声音播放，然后再自己进行上层封装。 ```cpp class AudioPlayer { public: AudioPlayer() : stop{ false }, player_thread{ &AudioPlayer::playMusic, this } {} ~AudioPlayer() { // 等待队列中所有音乐播放完毕 while (!audio_queue.empty()) { std::this_thread::sleep_for(50ms); } stop = true; cond.notify_all(); if (player_thread.joinable()) { player_thread.join(); } } void addAudioPath(const std::string& path) { std::lock_guard lock{ mtx }; // 互斥量确保了同一时间不会有其它地方在操作共享资源（队列） audio_queue.push(path); // 为队列添加元素 表示有新的提示音需要播放 cond.notify_one(); // 通知线程新的音频 } private: void playMusic() { while (!stop) { std::string path; { std::unique_lock lock{ mtx }; cond.wait(lock, [this] { return !audio_queue.empty() || stop; }); if (audio_queue.empty()) return; // 防止在对象为空时析构出错 path = audio_queue.front(); // 从队列中取出元素 audio_queue.pop(); // 取出后就删除元素，表示此元素已被使用 } if (!music.openFromFile(path)) { std::cerr << "无法加载音频文件: " << path << std::endl; continue; // 继续播放下一个音频 } music.play(); // 等待音频播放完毕 while (music.getStatus() == sf::SoundSource::Playing) { sf::sleep(sf::seconds(0.1f)); // sleep 避免忙等占用 CPU } } } std::atomic stop; // 控制线程的停止与退出， std::thread player_thread; // 后台执行音频任务的专用线程 std::mutex mtx; // 保护共享资源 std::condition_variable cond; // 控制线程等待和唤醒，当有新任务时通知音频线程 std::queue audio_queue; // 音频任务队列，存储待播放的音频文件路径 sf::Music music; // SFML 音频播放器，用于加载和播放音频文件 }; ``` 该代码实现了一个简单的**后台音频播放类型**，通过**条件变量**和**互斥量**确保播放线程 `playMusic` 只在只在**有音频任务需要播放时工作**（当外部通过调用 `addAudioPath()` 向队列添加播放任务时）。在没有任务时，线程保持等待状态，避免占用 CPU 资源影响主程序的运行。 > #### 注意 > > 其实这段代码还存在着一个初始化顺序导致的问题，见 [**#27**](https://github.com/Mq-b/ModernCpp-ConcurrentProgramming-Tutorial/issues/27) 此外，关于提示音的播报，为了避免每次都手动添加路径，我们可以创建一个音频资源数组，便于使用： ```cpp static constexpr std::array soundResources{ "./sound/01初始化失败.ogg", "./sound/02初始化成功.ogg", "./sound/03试剂不足，请添加.ogg", "./sound/04试剂已失效，请更新.ogg", "./sound/05清洗液不足，请添加.ogg", "./sound/06废液桶即将装满，请及时清空.ogg", "./sound/07废料箱即将装满，请及时清空.ogg", "./sound/08激发液A液不足，请添加.ogg", "./sound/09激发液B液不足，请添加.ogg", "./sound/10反应杯不足，请添加.ogg", "./sound/11检测全部完成.ogg" }; ``` 为了提高代码的可读性，我们还可以使用一个枚举类型来表示音频资源的索引： ```cpp enum SoundIndex { InitializationFailed, InitializationSuccessful, ReagentInsufficient, ReagentExpired, CleaningAgentInsufficient, WasteBinAlmostFull, WasteContainerAlmostFull, LiquidAInsufficient, LiquidBInsufficient, ReactionCupInsufficient, DetectionCompleted, SoundCount // 总音频数量，用于计数 }; ``` > \[!Note] > > ~~需要注意的是 SFML不支持 `.mp3` 格式的音频文件，大家可以使用 ffmpeg 或者其它软件~~[~~网站~~](https://www.freeconvert.com/audio-converter)~~将音频转换为支持的格式。~~ > > SFML 自 [2.6](https://github.com/SFML/SFML/releases/tag/2.6.0) 版本开始通过 [**minimp3**](https://github.com/lieff/minimp3) 支持 `.mp3` 格式的音频文件。不过新版本的 SFML 也要求更高版本的工具链。 > > 如果只是为了播放 `.mp3` 的音乐，也可以直接使用 `minimp3` 。 如果是测试使用，不知道去哪生成这些语音播报，我们推荐 [`tts-vue`](https://github.com/LokerL/tts-vue)。 > 我们的代码也可以在 Linux 中运行，并且整体仅需 C++11 标准（除了 `soundResources` 数组）。 SFML 依赖于 [**FLAC**](https://xiph.org/flac/) 和 [**OpenAL**](https://www.openal.org/) 这两个库。官网上[下载](https://www.sfml-dev.org/download/sfml/2.5.1/)的 windows 版本的 SFML 已包含这些依赖，但在 Linux 上需要用户自行下载并安装它们。如： > > ```shell > sudo apt-get install libflac-dev > sudo apt-get install libopenal-dev > ``` > \[!Tip] 这种设计思路非常常见。例如，[`USBMonitor-cpp`](https://github.com/Mq-b/USBMonitor-cpp) 是一个跨平台的 C++ 库，用于监测 U 盘插拔状态变化。感兴趣的话可以参考该项目，进一步学习和实践。 ## 使用 `future` 举个例子：我们在车站等车，你可能会做一些别的事情打发时间，比如学习[现代 C++ 模板教程](https://github.com/Mq-b/Modern-Cpp-templates-tutorial)、观看 [mq白](https://space.bilibili.com/1292761396) 的视频教程、玩手机等。不过，你始终在等待一件事情：***车到站***。 C++ 标准库将这种事件称为 [future](https://zh.cppreference.com/w/cpp/thread#.E6.9C.AA.E6.9D.A5.E4.BD.93)。它用于处理线程中需要等待某个事件的情况，线程知道预期结果。等待的同时也可以执行其它的任务。 C++ 标准库有两种 future，都声明在 [``](https://zh.cppreference.com/w/cpp/header/future) 头文件中：独占的 [`std::future`](https://zh.cppreference.com/w/cpp/thread/future) 、共享的 [`std::shared_future`](https://zh.cppreference.com/w/cpp/thread/shared_future)。它们的区别与 `std::unique_ptr` 和 `std::shared_ptr` 类似。`std::future` 只能与**单个**指定事件关联，而 `std::shared_future` 能关联**多个**事件。它们都是模板，它们的模板类型参数，就是其关联的事件（函数）的返回类型。当多个线程需要访问一个独立 future 对象时， 必须使用互斥量或类似同步机制进行保护。而多个线程访问同一共享状态，若每个线程都是通过其自身的 `shared_future` 对象副本进行访问，则是安全的。 最简单有效的使用是，我们先前讲的 `std::thread` 在线程中执行任务是没有返回值的，这个问题就能使用 future 解决。 ### 创建异步任务获取返回值 假设需要执行一个耗时任务并获取其返回值，但是并不急切的需要它。那么就可以启动新线程计算，然而 `std::thread` 没提供直接从线程获取返回值的机制。所以我们可以使用 [`std::async`](https://zh.cppreference.com/w/cpp/thread/async) 函数模板。 使用 `std::async` 启动一个异步任务，它会返回一个 `std::future` 对象，这个对象和任务关联，将持有最终计算出来的结果。当需要任务执行完的结果的时候，只需要调用 [`get()`](https://zh.cppreference.com/w/cpp/thread/future/get) 成员函数，就会阻塞直到 `future` 为就绪为止（即任务执行完毕），返回执行结果。[`valid()`](https://zh.cppreference.com/w/cpp/thread/future/valid) 成员函数检查 future 当前是否关联共享状态，即是否当前关联任务。还未关联，或者任务已经执行完（调用了 get()、set()），都会返回 **`false`**。 ```cpp #include #include #include // 引入 future 头文件 int task(int n) { std::cout << "异步任务 ID: " << std::this_thread::get_id() << '\n'; return n * n; } int main() { std::future future = std::async(task, 10); std::cout << "main: " << std::this_thread::get_id() << '\n'; std::cout << std::boolalpha << future.valid() << '\n'; // true std::cout << future.get() << '\n'; std::cout << std::boolalpha << future.valid() << '\n'; // false } ``` > [运行](https://godbolt.org/z/6xq7o9bTa)测试。 与 `std::thread` 一样，`std::async` 支持任意[可调用(Callable)](https://zh.cppreference.com/w/cpp/named_req/Callable)对象，以及传递调用参数。包括支持使用 `std::ref` ，以及**支持只能移动的类型**。我们下面详细聊一下 `std::async` 参数传递的事。 ```cpp struct X{ int operator()(int n)const{ return n * n; } }; struct Y{ int f(int n)const{ return n * n; } }; void f(int& p) { std::cout << &p << '\n'; } int main(){ Y y; int n = 0; auto t1 = std::async(X{}, 10); auto t2 = std::async(&Y::f,&y,10); auto t3 = std::async([] {}); auto t4 = std::async(f, std::ref(n)); std::cout << &n << '\n'; } ``` > [运行](https://godbolt.org/z/fEvs3M3vv)测试。 如你所见，它支持所有[可调用(Callable)](https://zh.cppreference.com/w/cpp/named_req/Callable)对象，并且也是默认按值复制，必须使用 `std::ref` 才能传递引用。并且它和 `std::thread` 一样，内部会将保有的参数副本转换为**右值表达式进行传递**，这是为了那些**只支持移动的类型**，左值引用没办法引用右值表达式，所以如果不使用 `std::ref`，这里 `void f(int&)` 就会导致编译错误，如果是 `void f(const int&)` 则可以通过编译，不过引用的不是我们传递的局部对象。 ```cpp void f(const int& p) {} void f2(int& p ){} int n = 0; std::async(f, n); // OK! 可以通过编译，不过引用的并非是局部的n std::async(f2, n); // Error! 无法通过编译 ``` 我们来展示使用 `std::move` ，也就是移动传递参数并接受返回值： ```cpp struct move_only{ move_only() { std::puts("默认构造"); } move_only(move_only&&)noexcept { std::puts("移动构造"); } move_only& operator=(move_only&&) noexcept { std::puts("移动赋值"); return *this; } move_only(const move_only&) = delete; }; move_only task(move_only x){ std::cout << "异步任务 ID: " << std::this_thread::get_id() << '\n'; return x; } int main(){ move_only x; std::future future = std::async(task, std::move(x)); std::this_thread::sleep_for(std::chrono::seconds(1)); std::cout << "main\n"; move_only result = future.get(); // 等待异步任务执行完毕 } ``` > [运行](https://godbolt.org/z/5cfvdEWWh)测试。 如你所见，它**支持只移动类型**，我们将参数使用 `std::move` 传递，接收参数的时候直接调用 `get` 函数即可。 *** 接下来我们聊 `std::async` 的[执行策略](https://zh.cppreference.com/w/cpp/thread/launch)，我们前面一直没有使用，其实就是在传递可调用对象与参数之前传递枚举值罢了： 1. `std::launch::async` 在不同**线程上**执行异步任务。 2. `std::launch::deferred` 惰性求值，**不创建线程**，等待 `future` 对象调用 `wait` 或 `get` 成员函数的时候执行任务。 而我们先前一直没有写明这个参数，是因为 `std::async` 函数模板有两个**重载**，不给出执行策略就是以：`std::launch::async | std::launch::deferred` 调用另一个重载版本（这一点中在[源码](https://github.com/microsoft/STL/blob/f54203f/stl/inc/future#L1425-L1430)中很明显），此策略表示由实现选择到底是否创建线程执行异步任务。典型情况是，如果系统资源充足，并且异步任务的执行不会导致性能问题，那么系统可能会选择在新线程中执行任务。但是，如果系统资源有限，或者延迟执行可以提高性能或节省资源，那么系统可能会选择延迟执行。 > 如果你阅读 [`libstdc++`](https://github.com/gcc-mirror/gcc/blob/bcb9dad/libstdc%2B%2B-v3/include/std/future#L1818-L1819) 的代码，会发现的确如此。 > > 然而值得注意的是，在 MSVC STL 的实现中，`launch::async | launch::deferred` 与 `launch::async` 执行策略毫无区别，[**源码**](https://github.com/microsoft/STL/blob/f54203f/stl/inc/future#L1400-L1410)如下： > > ```cpp > template > _Associated_state::type>* _Get_associated_state(launch _Psync, _Fty&& _Fnarg) { > // construct associated asynchronous state object for the launch type > switch (_Psync) { // select launch type > case launch::deferred: > return new _Deferred_async_state<_Ret>(_STD forward<_Fty>(_Fnarg)); > case launch::async: // TRANSITION, fixed in vMajorNext, should create a new thread here > default: > return new _Task_async_state<_Ret>(_STD forward<_Fty>(_Fnarg)); > } > } > ``` > > 且 `_Task_async_state` 会通过 `::Concurrency::create_task` 从线程池中获取线程并执行任务返回包装对象。 > > 简而言之，使用 `std::async`，只要不是 `launch::deferred` 策略，那么 MSVC STL 实现中都是必然在线程中执行任务。因为是线程池，所以执行新任务是否创建新线程，任务执行完毕线程是否立即销毁，***不确定***。 我们来展示一下： ```cpp void f(){ std::cout << std::this_thread::get_id() << '\n'; } int main(){ std::cout << std::this_thread::get_id() << '\n'; auto f1 = std::async(std::launch::deferred, f); f1.wait(); // 在 wait() 或 get() 调用时执行，不创建线程 auto f2 = std::async(std::launch::async,f); // 创建线程执行异步任务 auto f3 = std::async(std::launch::deferred | std::launch::async, f); // 实现选择的执行方式 } ``` > [运行](https://godbolt.org/z/abr96xqvM)测试。 *** 其实到此基本就差不多了，我们再介绍两个常见问题即可： 1. 如果从 `std::async` 获得的 [`std::future`](https://zh.cppreference.com/w/cpp/thread/future) 没有被移动或绑定到引用，那么在完整表达式结尾， `std::future` 的\*\*[析构函数](https://zh.cppreference.com/w/cpp/thread/future/%7Efuture)将阻塞，直到到异步任务完成\*\*。因为临时对象的生存期就在这一行，而对象生存期结束就会调用调用析构函数。 ```cpp std::async(std::launch::async, []{ f(); }); // 临时量的析构函数等待 f() std::async(std::launch::async, []{ g(); }); // f() 完成前不开始 ``` 如你所见，这并不能创建异步任务，它会阻塞，然后逐个执行。 2. 被移动的 `std::future` 没有所有权，失去共享状态，不能调用 `get`、`wait` 成员函数。 ```cpp auto t = std::async([] {}); std::future future{ std::move(t) }; t.wait(); // Error! 抛出异常 ``` 如同没有线程资源所有权的 `std::thread` 对象调用 `join()` 一样错误，这是移动语义的基本语义逻辑。 ### `future` 与 `std::packaged_task` 类模板 [`std::packaged_task`](https://zh.cppreference.com/w/cpp/thread/packaged_task) 包装任何[*可调用(Callable)*](https://zh.cppreference.com/w/cpp/named_req/Callable)目标（函数、lambda 表达式、bind 表达式或其它函数对象），使得能**异步**调用它。其返回值或所抛异常被存储于能通过 [std::future](https://zh.cppreference.com/w/cpp/thread/future) 对象访问的共享状态中。 通常它会和 `std::future` 一起使用，不过也可以单独使用，我们一步一步来： ```cpp std::packaged_task task([](int a, int b){ return std::pow(a, b); }); task(10, 2); // 执行传递的 lambda，但无法获取返回值 ``` 它有 [`operator()`](https://zh.cppreference.com/w/cpp/thread/packaged_task/operator\(\)) 的重载，它会执行我们传递的[*可调用(Callable)*](https://zh.cppreference.com/w/cpp/named_req/Callable)对象，不过这个重载的返回类型是 `void` **没办法获取返回值**。 如果想要异步的获取返回值，我们需要在调用 `operator()` 之前，让它和 future 关联，然后使用 `future.get()`，也就是： ```cpp std::packaged_task task([](int a, int b){ return std::pow(a, b); }); std::futurefuture = task.get_future(); task(10, 2); // 此处执行任务 std::cout << future.get() << '\n'; // 不阻塞，此处获取返回值 ``` > [运行](https://godbolt.org/z/799Khvadc)测试。 先关联任务，再执行任务，当我们想要获取任务的返回值的时候，就 `future.get()` 即可。值得注意的是，任务并不会在线程中执行，想要在线程中执行异步任务，然后再获取返回值，我们可以这么做： ```cpp std::packaged_task task([](int a, int b){ return std::pow(a, b); }); std::future future = task.get_future(); std::thread t{ std::move(task),10,2 }; // 任务在线程中执行 // todo.. 幻想还有许多耗时的代码 t.join(); std::cout << future.get() << '\n'; // 并不阻塞，获取任务返回值罢了 ``` > [运行](https://godbolt.org/z/85r9db49z)测试。 因为 `task` 本身是重载了 `operator()` 的，是可调用对象，自然可以传递给 `std::thread` 执行，以及传递调用参数。唯一需要注意的是我们使用了 `std::move` ，这是因为 `std::packaged_task` 只能移动，不能复制。 *** 简而言之，其实 `std::packaged_task` 也就是一个“包装”类而已，它本身并没什么特殊的，老老实实执行我们传递的任务，且方便我们获取返回值罢了，明确这一点，那么一切都不成问题。 `std::packaged_task` 也可以在线程中传递，在需要的时候获取返回值，而非像上面那样将它自己作为可调用对象： ```cpp template requires std::invocable&, Args...> void async_task(std::packaged_task& task, Args&&...args) { // todo.. task(std::forward(args)...); } int main() { std::packaged_task task([](int a,int b){ return a + b; }); int value = 50; std::future future = task.get_future(); // 创建一个线程来执行异步任务 std::thread t{ [&] {async_task(task, value, value); } }; std::cout << future.get() << '\n'; t.join(); } ``` > [运行](https://godbolt.org/z/qdK8GMaGE)测试。 我们套了一个 lambda，这是因为函数模板不是函数，它并非具体类型，没办法直接被那样传递使用，只能包一层了。这只是一个简单的示例，展示可以使用 `std::packaged_task` 作函数形参，然后我们来传递任务进行异步调用等操作。 我们再将第二章实现的并行 `sum` 改成 `std::package_task` + `std::future` 的形式： ```cpp template auto sum(ForwardIt first, ForwardIt last) { using value_type = std::iter_value_t; std::size_t num_threads = std::thread::hardware_concurrency(); std::ptrdiff_t distance = std::distance(first, last); if (distance > 1024000) { // 计算每个线程处理的元素数量 std::size_t chunk_size = distance / num_threads; std::size_t remainder = distance % num_threads; // 存储每个线程要执行的任务 std::vector> tasks; // 和每一个任务进行关联的 future 用于获取返回值 std::vector> futures(num_threads); // 存储关联线程的线程对象 std::vector threads; // 制作任务、与 future 关联、启动线程执行 auto start = first; for (std::size_t i = 0; i < num_threads; ++i) { auto end = std::next(start, chunk_size + (i < remainder ? 1 : 0)); tasks.emplace_back(std::packaged_task{[start, end, i] { return std::accumulate(start, end, value_type{}); }}); start = end; // 开始迭代器不断向前 futures[i] = tasks[i].get_future(); // 任务与 std::future 关联 threads.emplace_back(std::move(tasks[i])); } // 等待所有线程执行完毕 for (auto& thread : threads) thread.join(); // 汇总线程的计算结果 value_type total_sum {}; for (std::size_t i = 0; i < num_threads; ++i) { total_sum += futures[i].get(); } return total_sum; } value_type total_sum = std::accumulate(first, last, value_type{}); return total_sum; } ``` > [运行](https://godbolt.org/z/79fe1Gvcq)测试。 相比于之前，其实不同无非是定义了 `std::vector> tasks` 与 `std::vector> futures` ，然后在循环中制造任务插入容器，关联 future，再放到线程中执行。最后汇总的时候写一个循环，`futures[i].get()` 获取任务的返回值加起来即可。 到此，也就可以了。 ### 使用 `std::promise` 类模板 [`std::promise`](https://zh.cppreference.com/w/cpp/thread/promise) 用于存储一个值或一个异常，之后通过 `std::promise` 对象所创建的 [std::future](https://zh.cppreference.com/w/cpp/thread/future) 对象异步获得。 ```cpp // 计算函数，接受一个整数并返回它的平方 void calculate_square(std::promise promiseObj, int num) { // 模拟一些计算 std::this_thread::sleep_for(std::chrono::seconds(1)); // 计算平方并设置值到 promise 中 promiseObj.set_value(num * num); } // 创建一个 promise 对象，用于存储计算结果 std::promise promise; // 从 promise 获取 future 对象进行关联 std::future future = promise.get_future(); // 启动一个线程进行计算 int num = 5; std::thread t(calculate_square, std::move(promise), num); // 阻塞，直到结果可用 int result = future.get(); std::cout << num << " 的平方是：" << result << std::endl; t.join(); ``` > [运行](https://godbolt.org/z/qrnq16Mxh)测试。 我们在新线程中通过调用 [`set_value()`](https://zh.cppreference.com/w/cpp/thread/promise/set_value) 函数设置 `promise` 的值，并在主线程中通过与其关联的 future 对象的 `get()` 成员函数获取这个值，如果`promise`的值还没有被设置，那么将阻塞当前线程，直到被设置为止。同样的 `std::promise` **只能移动**，不可复制，所以我们使用了 `std::move` 进行传递。 *** 除了 `set_value()` 函数外，`std::promise` 还有一个 [`set_exception()`](https://zh.cppreference.com/w/cpp/thread/promise/set_exception) 成员函数，它接受一个 [`std::exception_ptr`](https://zh.cppreference.com/w/cpp/error/exception_ptr) 类型的参数，这个参数通常通过 [`std::current_exception()`](https://zh.cppreference.com/w/cpp/error/current_exception) 获取，用于指示当前线程中抛出的异常。然后，`std::future` 对象通过 `get()` 函数获取这个异常，如果 `promise` 所在的函数有异常被抛出，则 `std::future` 对象会重新抛出这个异常，从而允许主线程捕获并处理它。 ```cpp void throw_function(std::promise prom) { try { throw std::runtime_error("一个异常"); } catch (...) { prom.set_exception(std::current_exception()); } } int main() { std::promise prom; std::future fut = prom.get_future(); std::thread t(throw_function, std::move(prom)); try { std::cout << "等待线程执行，抛出异常并设置\n"; fut.get(); } catch (std::exception& e) { std::cerr << "来自线程的异常: " << e.what() << '\n'; } t.join(); } ``` [**运行结果**](https://godbolt.org/z/MdsaT8rdo)： ```txt 等待线程执行，抛出异常并设置 来自线程的异常: 一个异常 ``` 你可能对这段代码还有一些疑问：我们写的是 `promise` ，但是却没有使用 `set_value` 设置值，你可能会想着再写一行 `prom.set_value(0)`？ 共享状态的 promise 已经存储值或者异常，再次调用 `set_value`（`set_exception`） 会抛出 [std::future\_error](https://zh.cppreference.com/w/cpp/thread/future_error) 异常，将错误码设置为 [`promise_already_satisfied`](https://zh.cppreference.com/w/cpp/thread/future_errc)。这是因为 `std::promise` 对象只能是存储值或者异常其中一种，而**无法共存**。 简而言之，`set_value` 与 `set_exception` 二选一，如果先前调用了 `set_value` ，就不可再次调用 `set_exception`，反之亦然（不然就会抛出异常），示例如下： ```cpp void throw_function(std::promise prom) { prom.set_value(100); try { throw std::runtime_error("一个异常"); } catch (...) { try{ // 共享状态的 promise 已存储值，调用 set_exception 产生异常 prom.set_exception(std::current_exception()); }catch (std::exception& e){ std::cerr << "来自 set_exception 的异常: " << e.what() << '\n'; } } } int main() { std::promise prom; std::future fut = prom.get_future(); std::thread t(throw_function, std::move(prom)); std::cout << "等待线程执行，抛出异常并设置\n"; std::cout << "值：" << fut.get() << '\n'; // 100 t.join(); } ``` [**运行结果**](https://godbolt.org/z/hb9arEsoE)： ```txt 等待线程执行，抛出异常并设置 值：100 来自 set_exception 的异常: promise already satisfied ``` ### future 的状态变化 需要注意的是，**future 是一次性的**，所以你需要注意移动。并且，调用 `get` 函数后，future 对象也会**失去共享状态**。 * **移动语义**：这一点很好理解并且常见，因为**移动操作标志着所有权的转移**，意味着 `future` 不再拥有共享状态（如之前所提到）。`get` 和 `wait` 函数要求 `future` 对象拥有共享状态，否则会抛出异常。 * **共享状态失效**：调用 `get` 成员函数时，`future` 对象必须拥有共享状态，但调用完成后，它就会**失去共享状态**，不能再次调用 `get`。这是我们在本节需要特别讨论的内容。 ```cpp std::futurefuture = std::async([] {}); std::cout << std::boolalpha << future.valid() << '\n'; // true future.get(); std::cout << std::boolalpha << future.valid() << '\n'; // false try { future.get(); // 抛出 future_errc::no_state 异常 } catch (std::exception& e) { std::cerr << e.what() << '\n'; } ``` > [运行](https://godbolt.org/z/hvfMGnMbj)测试。 这个问题在许多文档中没有明确说明，但通过阅读源码（[MSVC STL](https://github.com/microsoft/STL/blob/f54203f/stl/inc/future)），可以很清楚地理解： ```cpp // std::future void get() { // block until ready then return or throw the stored exception future _Local{_STD move(*this)}; _Local._Get_value(); } // std::future _Ty get() { // block until ready then return the stored result or throw the stored exception future _Local{_STD move(*this)}; return _STD move(_Local._Get_value()); } // std::future _Ty& get() { // block until ready then return the stored result or throw the stored exception future _Local{_STD move(*this)}; return *_Local._Get_value(); } ``` 如上所示，我们展示了 `std::future` 的所有特化中 `get` 成员函数的实现。注意到了吗？尽管我们可能不了解移动构造函数的具体实现，但根据通用的语义，可以看出 `future _Local{_STD move(*this)};` 将当前对象的共享状态转移给了这个局部对象，而局部对象在函数结束时析构。这意味着当前对象失去共享状态，并且状态被完全销毁。 另外一提，`std::future` 这个特化，它 `return std::move` 是为了**支持只能移动的类型**能够使用 `get` 返回值，参见前文的 `move_only` 类型。 如果需要进行多次 `get` 调用，可以考虑使用下文提到的 `std::shared_future`。 ### 多个线程的等待 `std::shared_future` 之前的例子中我们一直使用 `std::future`，但 `std::future` 有一个局限：**future 是一次性的**，它的结果只能被一个线程获取。`get()` 成员函数只能调用一次，当结果被某个线程获取后，`std::future` 就无法再用于其他线程。 ```cpp int task(){ // todo.. return 10; } void thread_functio(std::future& fut){ // todo.. int result = fut.get(); std::cout << result << '\n'; // todo.. } int main(){ auto future = std::async(task); // 启动耗时的异步任务 // 可能有多个线程都需要此任务的返回值，于是我们将与其关联的 future 对象的引入传入 std::thread t{ thread_functio,std::ref(future) }; std::thread t2{ thread_functio,std::ref(future) }; t.join(); t2.join(); } ``` > 可能有多个线程都需要耗时的异步任务的返回值，于是我们将与其关联的 future 对象的引入传给线程对象，让它能在需要的时候获取。 > > 但是这存在个问题，future 是一次性的，只能被调用一次 `get()` 成员函数，所以以上代码存在问题。 此时就需要使用 `std::shared_future` 来替代 `std::future` 了。`std::future` 与 `std::shared_future` 的区别就如同 `std::unique_ptr`、`std::shared_ptr` 一样。 `std::future` 是只能移动的，其所有权可以在不同的对象中互相传递，但只有一个对象可以获得特定的同步结果。而 `std::shared_future` 是可复制的，多个对象可以指代同一个共享状态。 在多个线程中对**同一个 `std::shared_future` 对象进行操作时（如果没有进行同步保护）存在条件竞争。而从多个线程访问同一共享状态，若每个线程都是通过其自身的 `shared_future` 对象副本**进行访问，则是安全的。 ```cpp std::string fetch_data() { std::this_thread::sleep_for(std::chrono::seconds(1)); // 模拟耗时操作 return "从网络获取的数据！"; } int main() { std::future future_data = std::async(std::launch::async, fetch_data); // // 转移共享状态，原来的 future 被清空 valid() == false std::shared_future shared_future_data = future_data.share(); // 第一个线程等待结果并访问数据 std::thread thread1([&shared_future_data] { std::cout << "线程1：等待数据中..." << std::endl; shared_future_data.wait(); std::cout << "线程1：收到数据：" << shared_future_data.get() << std::endl; }); // 第二个线程等待结果并访问数据 std::thread thread2([&shared_future_data] { std::cout << "线程2：等待数据中..." << std::endl; shared_future_data.wait(); std::cout << "线程2：收到数据：" << shared_future_data.get() << std::endl; }); thread1.join(); thread2.join(); } ``` 这段代码存在数据竞争，就如同我们先前所说：“***在多个线程中对**同一个 **`std::shared_future` 对象进行操作时（如果没有进行同步保护）存在条件竞争***”，它并没有提供线程安全的方式。而我们的 lambda 是按引用传递，也就是“**同一个**”进行操作了。可以改为： ```cpp std::string fetch_data() { std::this_thread::sleep_for(std::chrono::seconds(1)); // 模拟耗时操作 return "从网络获取的数据！"; } int main() { std::future future_data = std::async(std::launch::async, fetch_data); std::shared_future shared_future_data = future_data.share(); std::thread thread1([shared_future_data] { std::cout << "线程1：等待数据中..." << std::endl; shared_future_data.wait(); std::cout << "线程1：收到数据：" << shared_future_data.get() << std::endl; }); std::thread thread2([shared_future_data] { std::cout << "线程2：等待数据中..." << std::endl; shared_future_data.wait(); std::cout << "线程2：收到数据：" << shared_future_data.get() << std::endl; }); thread1.join(); thread2.join(); } ``` 这样访问的就都是 `std::shared_future` 的副本了，我们的 lambda 按复制捕获 `std::shared_future` 对象，每个线程都有一个 shared\_future 的副本，这样不会有任何问题。这一点和 `std::shared_ptr` 类似。 `std::promise` 也同，它的 `get_future()` 成员函数一样可以用来构造 `std::shared_future`，虽然它的返回类型是 `std::future`，不过不影响，这是因为 `std::shared_future` 有一个 `std::future&&` 参数的[构造函数](https://zh.cppreference.com/w/cpp/thread/shared_future/shared_future)，转移 `std::future` 的所有权。 ```cpp std::promise p; std::shared_future sf{ p.get_future() }; // 隐式转移所有权 ``` 就不需要再强调了。 ## 限时等待 阻塞调用会将线程挂起一段（不确定的）时间，直到对应的事件发生。通常情况下，这样的方式很好，但是在一些情况下，需要限定线程等待的时间，因为无限期地等待事件发生可能会导致性能下降或资源浪费。一个常见的例子是在很多网络库中的 `connect` 函数，这个函数调用是阻塞的，但是也是限时的，一定时间内没有连接到服务器就不会继续阻塞了，会进行其它处理，比如抛出异常。 介绍两种指定超时的方式，一种是“**时间段**”，另一种是“**时间点**”，其实就是先前讲的 [`std::this::thread::sleep_for`](https://zh.cppreference.com/w/cpp/thread/sleep_for) 与 [`std::this_thread::sleep_until`](https://zh.cppreference.com/w/cpp/thread/sleep_until) 的区别。前者是需要指定等待一段时间（比如 10 毫秒）。而后者是指定等待到一个具体的时间点（比如到 2024-05-07T12:01:10.123）。多数函数都对两种超时方式进行处理。**处理持续时间的函数以 `_for` 作为后缀，处理绝对时间的函数以 `_until` 作为后缀**。 条件变量 `std::condition_variable` 的等待函数，也有两个超时的版本 [`wait_for`](https://zh.cppreference.com/w/cpp/thread/condition_variable/wait_for) 和 [`wait_until`](https://zh.cppreference.com/w/cpp/thread/condition_variable/wait_until) 。它们和我们先前讲的 `wait` 成员函数一样有两个重载，可以选择是否传递一个[*谓词*](https://zh.cppreference.com/w/cpp/named_req/Predicate)。它们相比于 `wait` 多了一个解除阻塞的可能，即：**超过指定的时长或抵达指定的时间点**。 在讲述它的使用细节之前，我们还是要来先聊一下 C++ 中的[**时间库**](https://zh.cppreference.com/w/cpp/chrono#std::chrono_.E5.BA.93)（chrono），指定时间的方式，它较为麻烦。我们分：***时钟**（clock）*、***时间段**（duration）*、\***时间点**（time point）\*三个阶段稍微介绍一下。 ### 时钟 在 C++ 标准库中，时钟被视为时间信息的来源。C++ 定义了很多种时间类型，每种时钟类型都提供了四种不同的信息： * 当前时间 * 时间类型 * 时钟节拍 * 稳定时钟 当前时间可以通过静态成员函数 `now` 获取，例如，[`std::chrono::system_clock::now()`](https://zh.cppreference.com/w/cpp/chrono/system_clock/now) 会返回系统的当前时间。特定的时间点则可以通过 [`time_point`](https://zh.cppreference.com/w/cpp/chrono/time_point) 来指定。`system_clock::now()` 的返回类型就是 `time_point`。 时钟节拍被指定为 1/x（x 在不同硬件上有不同的值）秒，这是由时间周期所决定。假设一个时钟一秒有 25 个节拍，因此一个周期为 `std::ratio<1,25>` 。当一个时钟的时钟节拍每 2.5 秒一次，周期就可以表示为 `std::ratio<5,2>`。 类模板 [**`std::chrono::duration`**](https://zh.cppreference.com/w/cpp/chrono/duration) 表示时间间隔。 ```cpp template> class duration; ``` > [`std::ratio`](https://zh.cppreference.com/w/cpp/numeric/ratio/ratio) 是一个分数类模板，它有两个非类型模板参数，也就是分子与分母，分母有默认实参 1，所以 `std::ratio<1>` 等价于 `std::ratio<1,1>`。 如你所见，它默认的时钟节拍是 1，这是一个很重要的类，标准库通过它定义了很多的时间类型，比如 **`std::chrono::minutes`** 是分钟类型，那么它的 `Period` 就是 `std::ratio<60>` ，因为一分钟等于 60 秒。 ```cpp using minutes = duration>; ``` 稳定时钟（Steady Clock）是指提供稳定、持续递增的时间流逝信息的时钟。它的特点是不受系统时间调整或变化的影响，即使在系统休眠或时钟调整的情况下，它也能保持稳定。在 C++ 标准库中，[`std::chrono::steady_clock`](https://zh.cppreference.com/w/cpp/chrono/steady_clock) 就是一个稳定时钟。它通常用于测量时间间隔和性能计时等需要高精度和稳定性的场景。可以通过 `is_steady` 静态常量判断当前时钟是否是稳定时钟。 稳定时钟的主要优点在于，它可以提供相对于起始时间的稳定的递增时间，因此适用于需要保持时间顺序和不受系统时间变化影响的应用场景。相比之下，像 [`std::chrono::system_clock`](https://zh.cppreference.com/w/cpp/chrono/system_clock) 这样的系统时钟可能会受到系统时间调整或变化的影响，因此在某些情况下可能不适合对时间间隔进行精确测量。 不管使用哪种时钟获取时间，C++ 都提供了函数，可以将时间点转换为 [**time\_t**](https://zh.cppreference.com/w/cpp/chrono/c/time_t) 类型的值： ```cpp auto now = std::chrono::system_clock::now(); time_t now_time = std::chrono::system_clock::to_time_t(now); std::cout << "Current time:\t" << std::put_time(std::localtime(&now_time), "%H:%M:%S\n"); auto now2 = std::chrono::steady_clock::now(); now_time = std::chrono::system_clock::to_time_t(now); std::cout << "Current time:\t" << std::put_time(std::localtime(&now_time), "%H:%M:%S\n"); ``` C++ 的时间库极其繁杂，主要在于类型之多，以及实现之复杂。根据我们的描述，了解基本构成、概念、使用，即可。 ### 时间段 时间部分最简单的就是时间段，主要的内容就是我们上面讲的类模板 `std::chrono::duration` ，它用于对时间段进行处理。 它的第一个参数是类型表示，第二个参数就是先前提到的“节拍”，需要传递一个 `std::ratio` 类型，也就是一个时钟所用的秒数。 标准库在 `std::chrono` 命名空间内为时间段提供了一系列的类型，它们都是通过 `std::chrono::duration` 定义的[别名](https://github.com/microsoft/STL/blob/daeb0a6/stl/inc/__msvc_chrono.hpp#L508-L519)： ```cpp using nanoseconds = duration; using microseconds = duration; using milliseconds = duration; using seconds = duration; using minutes = duration>; using hours = duration>; // CXX20 using days = duration, hours::period>>; using weeks = duration, days::period>>; using years = duration, days::period>>; using months = duration>>; ``` 如果没有指明 `duration` 的第二个非类型模板参数，那么代表默认 `std::ratio<1>`，比如 `seconds` 也就是一秒。 如上，是 MSVC STL 定义的，看似有一些没有使用 `ratio` 作为第二个参数，其实也还是别名罢了，[见](https://github.com/microsoft/STL/blob/daeb0a6/stl/inc/ratio#L262-L277)： ```cpp using milli = ratio<1, 1000>; // 千分之一秒，也就是一毫秒了 ``` 并且为了方便使用，在 C++14 标准库增加了时间字面量，存在于 `std::chrono_literals` 命名空间中，让我们得以简单的使用： ```cpp using namespace std::chrono_literals; auto one_nanosecond = 1ns; auto one_microsecond = 1us; auto one_millisecond = 1ms; auto one_second = 1s; auto one_minute = 1min; auto one_hour = 1h; ``` 当不要求截断值的情况下（时转换为秒时没问题的，但反过来不行）时间段有隐式转换，显式转换可以由 [`std::chrono::duration_cast<>`](https://zh.cppreference.com/w/cpp/chrono/duration/duration_cast) 来完成。 ```cpp std::chrono::milliseconds ms{ 3999 }; std::chrono::seconds s = std::chrono::duration_cast(ms); std::cout << s.count() << '\n'; ``` 这里的结果是**截断**的，而不会进行所谓的四舍五入，`3999` 毫秒，也就是 `3.999` 秒最终的值是 `3`。 > 很多时候这并不是我们想要的，比如我们想要的其实是输出 `3.999` 秒，而不是 `3` 秒 或者 `3999` 毫秒。 > > seconds 是 `duration` 这意味着它无法接受浮点数，我们直接改成 `duration` 即可： > > ```cpp > std::chrono::duration s = std::chrono::duration_cast>(ms); > ``` > > 当然了，这样写很冗余，并且这种形式的转换是可以直接隐式的，也就是其实我们可以直接： > > ```cpp > std::chrono::duration s = ms; > ``` > > 无需使用 `duration_cast`，可以直接隐式转换。 > > 另外我们用的 `duration` 都是省略了 `ratio` 的，其实默认类型就是 `ratio<1>`，代表一秒。参见源码[声明](https://github.com/microsoft/STL/blob/0be5257/stl/inc/__msvc_chrono.hpp#L80-L81)： > > ```cpp > _EXPORT_STD template > > class duration; > ``` 时间库支持四则运算，可以对两个时间段进行加减乘除。时间段对象可以通过 [`count()`](https://zh.cppreference.com/w/cpp/chrono/duration/count) 成员函数获得计次数。例如 `std::chrono::milliseconds{123}.count()` 的结果就是 123。 基于时间段的等待都是由 `std::chrono::duration<>` 来完成。例如：等待一个 future 对象在 35 毫秒内变为就绪状态： ```cpp std::future future = std::async([] {return 6; }); if (future.wait_for(35ms) == std::future_status::ready) std::cout << future.get() << '\n'; ``` [`wait_for`](https://zh.cppreference.com/w/cpp/thread/future/wait_for)： 等*待结果，如果在指定的超时间隔后仍然无法得到结果，则返回*。它的返回类型是一个枚举类 [`std::future_status`](https://zh.cppreference.com/w/cpp/thread/future_status) ，三个枚举项分别表示三种 future 状态。 | `deferred` | 共享状态持有的函数正在延迟运行，结果将仅在明确请求时计算 | | ---------- | ---------------------------- | | `ready` | **共享状态就绪** | | `timeout` | **共享状态在经过指定的等待时间内仍未就绪** | `timeout` 超时，也很好理解，那我们就提一下 `deferred` ： ```cpp auto future = std::async(std::launch::deferred, []{}); if (future.wait_for(35ms) == std::future_status::deferred) std::cout << "future_status::deferred " << "正在延迟执行\n"; future.wait(); // 在 wait() 或 get() 调用时执行，不创建线程 ``` ### 时间点 时间点可用 [`std::chrono::time_point<>`](https://zh.cppreference.com/w/cpp/chrono/time_point) 来表示，第一个模板参数用来指定使用的时钟，第二个模板参数用来表示时间单位（`std::chrono::duration<>`）。时间点顾名思义就是时间中的一个点，在 C++ 中用于表达当前时间，先前提到的静态成员函数 `now()` 获取当前时间，它们的返回类型都是 `std::chrono::time_point`。 ```cpp template< class Clock, class Duration = typename Clock::duration > class time_point; ``` 如你所见，它的第二个模板参数是**时间段**，就是时间的间隔，其实也就可以理解为表示时间点的**精度**，默认是根据第一个参数时钟得到的，所以假设有类型： ```cpp std::chrono::time_point ``` 那它等价于： ```cpp std::chrono::time_point ``` 也就是说第二个参数的实际类型是： ```cpp std::chrono::duration> // // 100 nanoseconds ``` 也就是说 `std::chrono::time_point` 的精度是 100 纳秒。 更多的问题参见[源码](https://github.com/microsoft/STL/blob/f54203f/stl/inc/__msvc_chrono.hpp#L644-L647)都很直观。 > 注意，这里的精度并非是实际的时间精度。时间和硬件系统等关系极大，以 windows 为例： > > Windows 内核中的时间间隔计时器默认每隔 **15.6** 毫秒触发一次中断。因此，如果你使用基于系统时钟的计时方法，默认情况下精度约为 15.6 毫秒。不可能达到纳秒级别。 > > 由于这个系统时钟的限制，那些基于系统时钟的 API（例如 `Sleep()`、`WaitForSingleObject()` 等）的最小睡眠时间默认就是 15.6 毫秒左右。 > > 如： > > ```cpp > std::this_thread::sleep_for(std::chrono::milliseconds(1)); > ``` > > 不过我们也可以使用系统 API 调整系统时钟的精度，需要链接 windows 多媒体库 **`winmm.lib`** ，然后使用 API： > > ```cpp > timeBeginPeriod(1); // 设置时钟精度为 1 毫秒 > // todo.. > timeEndPeriod(1); // 恢复默认精度 > ``` *** 同样的，时间点也支持加减以及比较操作。 ```cpp std::chrono::steady_clock::now() + std::chrono::nanoseconds(500); // 500 纳秒之后的时间 ``` 可以减去一个时间点，结果是两个时间点的时间差。这对于代码块的计时是很有用的，如： ```cpp auto start = std::chrono::steady_clock::now(); std::this_thread::sleep_for(std::chrono::seconds(1)); auto end = std::chrono::steady_clock::now(); auto result = std::chrono::duration_cast(end - start); std::cout << result.count() << '\n'; ``` > [运行](https://godbolt.org/z/ExdnzKoYj)测试。 我们进行了一个显式的转换，最终输出的是以毫秒作为单位，有可能不会是 1000，没有这么精确。 *** 等待条件变量满足条件——带超时功能 ```cpp using namespace std::chrono_literals; std::condition_variable cv; bool done{}; std::mutex m; bool wait_loop() { const auto timeout = std::chrono::steady_clock::now() + 500ms; std::unique_lock lk{ m }; while (!done) { if (cv.wait_until(lk, timeout) == std::cv_status::timeout) { std::cout << "超时 500ms\n"; return false; } } return true; } ``` > [运行](https://godbolt.org/z/s6vTThqK1)测试。 `_until` 也就是等待到一个时间点，我们设置的是等待到当前时间往后 500 毫秒。如果超过了这个时间还没有被唤醒，那就打印超时，并退出循环，函数返回 `false`。 到此，时间点的知识也就足够了。 ## 异步任务执行 在开发带有 UI 的程序时，主线程用于处理 UI 更新和用户交互，如果在主线程中执行耗时任务会导致界面卡顿。因此，需要使用异步任务来减轻主线程的压力。以下是一个使用 Qt 实现异步任务的示例，展示了如何在不阻塞 UI 线程的情况下执行耗时任务，并更新进度条。 ### 背景介绍 在 Qt 中，GUI 控件通常只能在创建它们的线程中进行操作，因为它们是线程不安全的。我们可以使用 `QMetaObject::invokeMethod` 来跨线程调用主线程上的控件方法，从而在其他线程中安全地更新 UI 控件。以下代码示例展示了如何通过 `QMetaObject::invokeMethod` 确保 UI 控件的更新操作在主线程中执行。 ```cpp void task(){ future = std::async(std::launch::async, [=] { QMetaObject::invokeMethod(this, [this] { button->setEnabled(false); progressBar->setRange(0, 1000); button->setText("正在执行..."); }); for (int i = 0; i < 1000; ++i) { std::this_thread::sleep_for(10ms); QMetaObject::invokeMethod(this, [this, i] { progressBar->setValue(i); }); } QMetaObject::invokeMethod(this, [this] { button->setText("start"); button->setEnabled(true); }); }); } ``` 上面的代码创建了一个异步任务，并指明了执行策略。任务在线程中执行，不会阻塞 UI 线程。如果不这样做，界面将会卡顿（可以尝试将函数的第一行与最后一行注释掉以验证这一点）。 ![进度条](/files/pJAhAs0Tt2tOLLrcwkhP) 在启动进度条后，能够正常点击“**测试**”按钮并触发弹窗，说明 UI 没有被阻塞。相反，如果不使用线程，界面将会卡住，无法点击“**测试**”按钮或移动窗口。 ### 项目说明 项目使用 Visual Studio + CMake，可以直接安装 Qt 插件后打开此项目。项目结构简单，所有界面与设置均通过代码控制，无需进行其他 UI 操作。只需关注 `async_progress_bar.h`、`async_progress_bar.cpp` 和 `main.cpp` 这三个文件，它们位于仓库的 **`code`** 文件夹中。 ### 完整代码实现 ```cpp class async_progress_bar : public QMainWindow{ Q_OBJECT public: async_progress_bar(QWidget *parent = nullptr); ~async_progress_bar(); void task(){ future = std::async(std::launch::async, [=] { QMetaObject::invokeMethod(this, [this] { // 这里显示的线程 ID 就是主线程，代表这些任务就是在主线程，即 UI 线程执行 QMessageBox::information(nullptr, "线程ID", std::to_string(_Thrd_id()).c_str()); button->setEnabled(false); progress_bar->setRange(0, 1000); button->setText("正在执行..."); }); for (int i = 0; i <= 1000; ++i) { std::this_thread::sleep_for(10ms); QMetaObject::invokeMethod(this, [this, i] { progress_bar->setValue(i); }); } QMetaObject::invokeMethod(this, [this] { button->setText("start"); button->setEnabled(true); }); // 不在 invokeMethod 中获取线程 ID，这里显示的是子线程的ID auto s = std::to_string(_Thrd_id()); QMetaObject::invokeMethod(this, [=] { QMessageBox::information(nullptr, "线程ID", s.c_str()); }); }); } private: QString progress_bar_style = "QProgressBar {" " border: 2px solid grey;" " border-radius: 5px;" " background-color: lightgrey;" " text-align: center;" // 文本居中 " color: #000000;" // 文本颜色 "}" "QProgressBar::chunk {" " background-color: #7FFF00;" " width: 10px;" // 设置每个进度块的宽度 " font: bold 14px;" // 设置进度条文本字体 "}"; QString button_style = "QPushButton {" " text-align: center;" // 文本居中 "}"; QProgressBar* progress_bar{}; QPushButton* button{}; QPushButton* button2{}; Ui::async_progress_barClass ui{}; std::futurefuture; }; // 创建控件 设置布局、样式 连接信号 async_progress_bar::async_progress_bar(QWidget *parent) : QMainWindow{ parent }, progress_bar{ new QProgressBar(this) }, button{ new QPushButton("start",this) },button2{ new QPushButton("测试",this) } { ui.setupUi(this); progress_bar->setStyleSheet(progress_bar_style); progress_bar->setRange(0, 1000); button->setMinimumSize(100, 50); button->setMaximumWidth(100); button->setStyleSheet(button_style); button->setSizePolicy(QSizePolicy::Minimum, QSizePolicy::Fixed); button2->setMinimumSize(100, 50); button2->setMaximumWidth(100); button2->setStyleSheet(button_style); button2->setSizePolicy(QSizePolicy::Minimum, QSizePolicy::Fixed); QVBoxLayout* layout = new QVBoxLayout; layout->addWidget(progress_bar); layout->addWidget(button, 0, Qt::AlignHCenter); layout->addWidget(button2, 0, Qt::AlignHCenter); // 设置窗口布局为垂直布局管理器 centralWidget()->setLayout(layout); connect(button, &QPushButton::clicked, this, &async_progress_bar::task); connect(button2, &QPushButton::clicked, []{ QMessageBox::information(nullptr, "测试", "没有卡界面！"); }); } ``` ### 注意事项 * `QMetaObject::invokeMethod` 的 lambda 是在主线程运行的，通过显示的线程 ID 可以验证这一点。 * 使用 `std::async` 的 `std::launch::async` 参数强制异步执行任务，以确保任务在新线程中运行。 ### 跨平台兼容性 C++11 的 `std::this_thread::get_id()` 返回的内部类 [`std::thread::id`](https://zh.cppreference.com/w/cpp/thread/thread/id) 没办法直接转换为 `unsigned int`，我们就直接使用了 win32 的 API *`_Thrd_id()`* 了。如果您是 Linux 之类的环境，使用 [POSIX](https://pubs.opengroup.org/onlinepubs/9699919799/) 接口 [*`pthread_self()`*](https://pubs.opengroup.org/onlinepubs/009696699/functions/pthread_self.html)。 ### 实践建议 这个例子其实很好的展示了多线程异步的作用，因为有 UI，所以很直观，毕竟如果你不用线程，那么不就卡界面了，用了就没事。 建议下载并运行此项目，通过实际操作理解代码效果。同时，可以尝试修改代码，观察不同情况下 UI 的响应情况，以加深对异步任务处理的理解。 ## C++20 信号量 C++20 引入了**信号量**，对于那些熟悉操作系统或其它并发支持库的开发者来说，这个同步设施的概念应该不会感到陌生。[信号量](https://zh.wikipedia.org/wiki/%E4%BF%A1%E5%8F%B7%E9%87%8F)源自操作系统，是一个古老而广泛应用的同步设施，在各种编程语言中都有自己的抽象实现。然而，C++ 标准库对其的支持却来得很晚，在 C++20 中才得以引入。 信号量是一个非常**轻量简单**的同步设施，它维护一个计数，这个计数不能小于 `0`。信号量提供两种基本操作：**释放**（增加计数）和**等待**（减少计数）。如果当前信号量的计数值为 `0`，那么执行“***等待***”操作的线程将会**一直阻塞**，直到计数大于 `0`，也就是其它线程执行了“***释放***”操作。 C++ 提供了两个信号量类型：`std::counting_semaphore` 与 `std::binary_semaphore`，定义在 [``](https://zh.cppreference.com/w/cpp/header/semaphore) 中。 `binary_semaphore` 只是 `counting_semaphore` 的一个特化别名： ```cpp using binary_semaphore = counting_semaphore<1>; ``` 好了，我们举一个简单的例子来使用一下： ```cpp // 全局二元信号量对象 // 设置对象初始计数为 0 std::binary_semaphore smph_signal_main_to_thread{ 0 }; std::binary_semaphore smph_signal_thread_to_main{ 0 }; void thread_proc() { smph_signal_main_to_thread.acquire(); std::cout << "[线程] 获得信号" << std::endl; std::this_thread::sleep_for(3s); std::cout << "[线程] 发送信号\n"; smph_signal_thread_to_main.release(); } int main() { std::jthread thr_worker{ thread_proc }; std::cout << "[主] 发送信号\n"; smph_signal_main_to_thread.release(); smph_signal_thread_to_main.acquire(); std::cout << "[主] 获得信号\n"; } ``` [**运行结果**](https://godbolt.org/z/c55MeGYrT)： ```txt [主] 发送信号 [线程] 获得信号 [线程] 发送信号 [主] 获得信号 ``` [`acquire`](https://zh.cppreference.com/w/cpp/thread/counting_semaphore/acquire) 函数就是我们先前说的“*等待*”（**原子**地减少计数），[`release`](https://zh.cppreference.com/w/cpp/thread/counting_semaphore/release) 函数就是"释放"（**原子**地增加计数）。 *** 信号量常用于*发信/提醒*而非互斥，通过初始化该信号量为 0 从而阻塞尝试 acquire() 的接收者，直至提醒者通过调用 release(n) “发信”。在此方面可把信号量当作**条件变量的替代品**，**通常它有更好的性能**。 假设我们有一个 Web 服务器，它只能处理有限数量的并发请求。为了防止服务器过载，我们可以**使用信号量来限制并发请求的数量**。 ```cpp // 定义一个信号量，最大并发数为 3 std::counting_semaphore<3> semaphore{ 3 }; void handle_request(int request_id) { // 请求到达，尝试获取信号量 std::cout << "进入 handle_request 尝试获取信号量\n"; semaphore.acquire(); std::cout << "成功获取信号量\n"; // 此处延时三秒可以方便测试，会看到先输出 3 个“成功获取信号量”，因为只有三个线程能成功调用 acquire，剩余的会被阻塞 std::this_thread::sleep_for(3s); // 模拟处理时间 std::random_device rd; std::mt19937 gen{ rd() }; std::uniform_int_distribution<> dis(1, 5); int processing_time = dis(gen); std::this_thread::sleep_for(std::chrono::seconds(processing_time)); std::cout << std::format("请求 {} 已被处理\n", request_id); semaphore.release(); } int main() { // 模拟 10 个并发请求 std::vector threads; for (int i = 0; i < 10; ++i) { threads.emplace_back(handle_request, i); } } ``` > [运行](https://godbolt.org/z/xs9asvqre)测试。 这段代码很简单，以至于我们可以在这里来再说一条概念： * `counting_semaphore` 是一个轻量同步原语，能控制对共享资源的访问。不同于 [std::mutex](https://zh.cppreference.com/w/cpp/thread/mutex)，`counting_semaphore` **允许同一资源进行多个并发的访问**，***至少**允许 `LeastMaxValue` 个同时访问者*。 * `binary_semaphore` 是 `std::counting_semaphore` 的特化的别名，其 `LeastMaxValue` 为 1。 `LeastMaxValue` 是我们设置的非类型模板参数，意思是信号量维护的**计数最大值**。我们这段代码设置的是 `3`，也就是允许 3 个同时访问者。 > 虽然说是说有 LeastMaxValue 可能不是最大，但是我们通常不用在意这个事情，[MSVC STL 的实现](https://github.com/microsoft/STL/blob/697653d/stl/inc/semaphore#L63-L65)中 max 函数就是直接返回 `LeastMaxValue`，将它视为信号量维护的计数最大值即可。 牢记信号量的基本的概念不变，计数的值不能小于 `0`，如果当前信号量的计数值为 `0`，那么执行“***等待***”（acquire）操作的线程将会**一直阻塞**。明白这点，那么就都不存在问题。 通过这种方式，可以有效控制 Web 服务器处理并发请求的数量，防止服务器过载。 ## C++20 闩与屏障 闩 (latch) 与屏障 (barrier) 是线程协调机制，允许任何数量的线程阻塞**直至期待数量的线程到达**。闩不能重复使用，而屏障则可以。 * **`std::latch`：单次使用的线程屏障** * **`std::barrier`：可复用的线程屏障** 它们定义在标头 **``** 与 **``**。 与信号量类似，屏障也是一种古老而广泛应用的同步机制。许多系统 API 提供了对屏障机制的支持，例如 POSIX 和 Win32。此外，[OpenMP](https://learn.microsoft.com/zh-cn/cpp/parallel/openmp/2-directives?view=msvc-170#263-barrier-directive) 也提供了屏障机制来支持多线程编程。 ### `std::latch` “*闩*” ，中文语境一般说“*门闩*” 是指门背后用来关门的棍子。不过不用在意，在 C++ 中的意思就是先前说的：***单次使用的线程屏障***。 `latch` 类维护着一个 [`std::ptrdiff_t`](https://zh.cppreference.com/w/cpp/types/ptrdiff_t) 类型的计数，且只能**减少**计数，**无法增加计数**。在创建对象的时候初始化计数器的值。线程可以阻塞，直到 latch 对象的计数减少到零。由于无法增加计数，这使得 `latch` 成为一种**单次使用的屏障**。 ```cpp std::latch work_start{ 3 }; void work(){ std::cout << "等待其它线程执行\n"; work_start.wait(); // 等待计数为 0 std::cout << "任务开始执行\n"; } int main(){ std::jthread thread{ work }; std::this_thread::sleep_for(3s); std::cout << "休眠结束\n"; work_start.count_down(); // 默认值是 1 减少计数 1 work_start.count_down(2); // 传递参数 2 减少计数 2 } ``` [**运行结果**](https://godbolt.org/z/sEhMeraYs)： ```txt 等待其它线程执行 休眠结束 任务开始执行 ``` 在这个例子中，通过调用 `wait` 函数阻塞子线程，直到主线程调用 `count_down` 函数原子地将计数减至 `0`，从而解除阻塞。这个例子清楚地展示了 `latch` 的使用，其逻辑比信号量更简单。 *** 由于 `latch` 的计数不可增加，它的使用通常非常简单，可以用来划分任务执行的工作区间。例如： ```cpp std::latch latch{ 10 }; void f(int id) { //todo.. 脑补任务 std::this_thread::sleep_for(1s); std::cout << std::format("线程 {} 执行完任务，开始等待其它线程执行到此处\n", id); latch.arrive_and_wait(); std::cout << std::format("线程 {} 彻底退出函数\n", id); } int main() { std::vector threads; for (int i = 0; i < 10; ++i) { threads.emplace_back(f,i); } } ``` > [运行](https://godbolt.org/z/KKjdWWKdq)测试。 [`arrive_and_wait`](https://zh.cppreference.com/w/cpp/thread/latch/arrive_and_wait) 函数等价于：`count_down(n); wait();`。也就是减少计数 + 等待。这意味着 必须等待所有线程执行到 `latch.arrive_and_wait();` 将 latch 的计数减少至 `0` 才能继续往下执行。这个示例非常直观地展示了如何使用 `latch` 来划分任务执行的工作区间。 由于 `latch` 的功能受限，通常用于简单直接的需求，不少情况很多同步设施都能完成你的需求，在这个时候请考虑**使用尽可能功能最少的那一个**。 * 使用功能尽可能少的设施有助于开发者阅读代码理解含义。如果使用的是一个功能丰富的设施，可能就无法直接猜测其意图。 ### `std::barrier` 上节我们学习了 `std::latch` ，本节内容也不会对你构成难度。 ```cpp template< class CompletionFunction = /* 未指定 */ > class barrier; ``` **CompletionFunction** - 函数对象类型。 *** [`std::barrier`](https://zh.cppreference.com/w/cpp/thread/barrier) 和 `std::latch` 最大的不同是，前者可以在阶段完成之后将计数重置为构造时传递的值，而后者只能减少计数。我们用一个非常简单直观的示例为你展示： ```cpp std::barrier barrier{ 10, [n = 1]()mutable noexcept {std::cout << "\t第" << n++ << "轮结束\n"; } }; void f(int start, int end){ for (int i = start; i <= end; ++i) { std::osyncstream{ std::cout } << i << ' '; barrier.arrive_and_wait(); // 减少计数并等待 解除阻塞时就重置计数并调用函数对象 std::this_thread::sleep_for(300ms); } } int main(){ std::vector threads; for (int i = 0; i < 10; ++i) { threads.emplace_back(f, i * 10 + 1, (i + 1) * 10); } } ``` **可能的**[**运行结果**](https://godbolt.org/z/9Tnsz537e)： ```txt 1 21 11 31 41 51 61 71 81 91 第1轮结束 12 2 22 32 42 52 62 72 92 82 第2轮结束 13 63 73 33 23 53 83 93 43 3 第3轮结束 14 44 24 34 94 74 64 4 84 54 第4轮结束 5 95 15 45 75 25 55 65 35 85 第5轮结束 6 46 16 26 56 96 86 66 76 36 第6轮结束 47 17 57 97 87 67 77 7 27 37 第7轮结束 38 8 28 78 68 88 98 58 18 48 第8轮结束 9 39 29 69 89 99 59 19 79 49 第9轮结束 30 40 70 10 90 50 60 20 80 100 第10轮结束 ``` 注意输出的规律，第一轮每个数字最后一位都是 `1`，第二轮每个数字最后一位都是 `2`……以此类推，因为我们分配给每个线程的输出任务就是如此，然后利用了屏障一轮一轮地打印。 `arrive_and_wait` 等价于 `wait(arrive());`。原子地将期待**计数减少 1**，然后在当前阶段的同步点**阻塞**直至运行当前阶段的阶段完成步骤。 `arrive_and_wait()` 会在期待计数减少至 `0` 时调用我们构造 barrier 对象时传入的 lambda 表达式，并**解除**所有在阶段同步点上阻塞的线程。之后**重置期待计数为构造中指定的值**。屏障的一个阶段就完成了。 * 并发调用`barrier` 除了析构函数外的成员函数不会引起数据竞争。 另外你可能注意到我们使用了 [`std::osyncstream`](https://zh.cppreference.com/w/cpp/io/basic_osyncstream) ，它是 C++20 引入的，此处是确保输出流在多线程环境中同步，**避免除数据竞争，而且将不以任何方式穿插或截断**。 > 虽然 `std::cout` 的 `operator<<` 调用是线程安全的，不会被打断，但多个 `operator<<` 的调用在多线程环境中可能会**交错**，导致输出结果混乱，使用 `std::osyncstream` 就可以解决这个问题。开发者可以尝试去除 `std::osyncstream` 直接使用 `std::cout`，效果会非常明显。 *** 使用 `arrive` 或 `arrive_and_wait` 减少的都是**当前屏障计数**，我们称作“*期待计数*”。不管如何减少计数，当完成一个*阶段*，就重置期待计数为**构造中指定的值**了。 标准库还提供一个函数 [`arrive_and_drop`](https://zh.cppreference.com/w/cpp/thread/barrier/arrive_and_drop) 可以改变重置的计数值：*它将所有后继阶段的初始期待计数减少一，当前阶段的期待计数也减少一*。 不用感到难以理解，我们来解释一下这个概念： ```cpp std::barrier barrier{ 4 }; // 初始化计数为 4 完成阶段重置计数也是 4 barrier.arrive_and_wait(); // 当前计数减 1，不影响之后重置计数 4 barrier.arrive_and_drop(); // 当前计数与重置之后的计数均减 1 完成阶段会重置计数为 3 ``` `arrive_and_drop` 可以用来控制在需要的时候，让一些线程退出同步，如： ```cpp std::atomic_int active_threads{ 4 }; std::barrier barrier{ 4, [n = 1]() mutable noexcept { std::cout << "\t第" << n++ << "轮结束，活跃线程数: " << active_threads << '\n'; } }; void f(int thread_id) { for (int i = 1; i <= 5; ++i) { std::osyncstream{ std::cout } << "线程 " << thread_id << " 输出: " << i << '\n'; if (i == 3 && thread_id == 2) { // 假设线程 ID 为 2 的线程在完成第三轮同步后退出 std::osyncstream{ std::cout } << "线程 " << thread_id << " 完成并退出\n"; --active_threads; // 减少活跃线程数 barrier.arrive_and_drop(); // 减少当前计数 1，并减少重置计数 1 return; } barrier.arrive_and_wait(); // 减少计数并等待，解除阻塞时重置计数并调用函数对象 } } int main() { std::vector threads; for (int i = 1; i <= 4; ++i) { threads.emplace_back(f, i); } } ``` > [运行](https://godbolt.org/z/T7WYeYGd1)测试。 初始线程有 4 个，线程 2 在执行了**三轮**同步便直接退出了，调用 `arrive_and_drop` 函数，下一个阶段的计数会重置为 `3`，也就是执行完第三轮同步后只有**三个活跃线程**继续执行。查看输出结果，非常的直观。 这样，`arrive_and_drop` 的作用就非常明显了，使用也十分的简单。 *** 最后请注意，我们的 lambda 表达式必须声明为 `noexcept` ，因为 `std::barrier` 要求其**函数对象类型必须是不抛出异常的**。即要求 `std::is_nothrow_invocable_v<_Completion_function&>` 为 **true**，见 [MSVC STL](https://github.com/microsoft/STL/blob/9ad382e/stl/inc/barrier#L75-L76)。 ```cpp std::barrier barrier{ 1,[] {} }; ``` 按照标准规定，这行代码会产生一个**编译错误**。因为传入的函数对象它不是 `noexcept` 的。不过，在 gcc 与 clang（即 libstdc++ 和 libc++）均可以通过编译，这是因为它们没有进行相应的检测，**存在缺陷**，为了代码的可维护性开发者应遵守标准规定，确保传入的函数对象是 `noexcept` 的。 ## 总结 在并发编程中，同步操作对于并发编程至关重要。如果没有同步，线程基本上就是独立的，因其任务之间的相关性，才可作为一个整体执行（比如第二章的并行求和）。本章讨论了多种用于同步操作的工具，包括条件变量、future、promise、package\_task、信号量。同时，详细介绍了 C++ 时间库的知识，以使用并发支持库中的“限时等待”。还使用 CMake + Qt 构建了一个带有 UI 界面的示例，展示异步**多线程的必要性**。最后介绍了 C++20 引入的两种新的并发设施，信号量、闩与屏障。 在讨论了 C++ 中的高级工具之后，现在让我们来看看底层工具：C++ 内存模型与原子操作。 --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://mq-bai.gitbook.io/modern-cpp-concurrent-programming/md/04-tong-bu-cao-zuo.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.