第 14 章 C++ 中的并发与多线程
1. 为什么需要多线程
在现代计算机中，通常会有多个 CPU 核心。在一些负载特别重的任务中，我们倾向于把这些任务分开到各个核心上进行执行，而不是把所有工作堆在一个核心上运行。这种情况下程序只吃到了一个核心，因此运行效率不高，甚至可能出现一核有难，多核围观的情况。绝大部分编程语言都支持多线程开发，C++ 也不例外。此外，多线程开发也是绝大多数大型程序（例如游戏引擎等）中的基础。在具备图形界面的应用程序中，我们也倾向于将界面线程与后台处理线程分离，以防止后台处理卡死图形界面的情况发生。
2. 开始使用多线程
要使用多线程来开发应用，我们需要包含 thread 头文件。在 thread 类中，我们需要以下参数：
- 一个函数，可以提供函数指针或 Lambda 表达式，作为线程的操作；
- 一个参数列表，用于有参函数情况下的参数传递，这个列表的长度是可变的。
我们可以使用以下代码开始开发我们的第一个多线程应用：
#include <chrono>
#include <iostream>
#include <thread>
using std::cout;
using std::endl;
using std::thread;
using std::chrono::milliseconds;
using std::this_thread::sleep_for;
int main() {
auto t1 = thread([&]() { cout << "Hello world!1" << endl; });
auto t2 = thread([&]() { cout << "Hello world!2" << endl; });
t1.join();
t2.join();
}我们会发现，代码执行结果似乎每次都不同。有时是第一个 Hello World 在上，有时是第二个在上，有时换行符会被挤掉，两行信息在同一行上显示了。这是为什么呢？这里我们需要改变一种思维方式来看待这段代码。在我们以前编写的所有程序中，所有语句都是串行运行的，也就是说，如果我们把 thread 相关的内容去掉，Hello World!2 的输出时机是永远不可能早于 Hello World!1 的，因为它们的顺序在编译时就已经确定了。但是在并发程序设计中，这两个线程是同时运行的。C++ 标准库并不保证同时只能有一个线程调用 ostream::operator<<，也就是说，在某一个瞬间，可能会出现两个线程同时写 cout 缓冲区的情况发生。这种问题被称为竞态条件（或竞争条件
#include <chrono>
#include <iostream>
#include <thread>
using std::cout;
using std::endl;
using std::thread;
using std::chrono::milliseconds;
using std::this_thread::sleep_for;
int main() {
int i = 0;
auto t1 = thread([&]() {
for (int j = 0; j < 10; j++) {
i++;
cout << i << endl;
}
});
auto t2 = thread([&]() {
for (int j = 0; j < 10; j++) {
i++;
cout << i << endl;
}
});
t1.join();
t2.join();
}运行这个程序，我们会发现程序的输出十分混乱。这里的共享变量除了 cout 缓冲区以外，还有变量 i。这会导致很多问题，包括不可重复读（在另一线程写入前读取变量，导致写入前后读取结果不一致
3. 锁机制——多线程安全的核心
1. 使用互斥锁
在多线程开发中，锁机制能够解决变量竞争问题。要使用锁，我们需要包含 mutex 头文件，它为我们提供了多线程的互斥锁支持。我们可以修改上面的程序为一个线程安全的版本：
#include <chrono>
#include <iostream>
#include <mutex>
#include <thread>
using std::cout;
using std::endl;
using std::mutex;
using std::thread;
using std::chrono::milliseconds;
using std::this_thread::sleep_for;
int main() {
mutex m;
int i = 0;
auto t1 = thread([&]() {
for (int j = 0; j < 10; j++) {
m.lock();
i++;
cout << i << endl;
m.unlock();
}
});
auto t2 = thread([&]() {
for (int j = 0; j < 10; j++) {
m.lock();
i++;
cout << i << endl;
m.unlock();
}
});
t1.join();
t2.join();
}执行这个程序，我们会发现输出终于有序化起来了。在互斥锁中，锁变量会禁止任意其他线程写入该线程中使用的变量，其他线程会等待到锁变量释放它们。这就完美提供了我们上面需要拥有的机制。
这里可能有人会问：主播主播，如果我对变量上锁忘记解锁了会发生什么呢？我们考虑下面的代码：
#include <chrono>
#include <iostream>
#include <mutex>
#include <thread>
using std::cout;
using std::endl;
using std::mutex;
using std::thread;
using std::chrono::milliseconds;
using std::this_thread::sleep_for;
int main() {
mutex m;
int i = 0;
auto t1 = thread([&]() {
for (int j = 0; j < 10; j++) {
m.lock();
i++;
cout << i << endl;
}
});
auto t2 = thread([&]() {
for (int j = 0; j < 10; j++) {
m.lock();
i++;
cout << i << endl;
m.unlock();
}
});
t1.join();
t2.join();
}这份代码相比上面的线程安全版本，去掉了 t1 的解锁过程。我们再次执行程序，会发现程序输出第一个值后就不再输出，同时程序永远不会停止。这种现象被称为死锁，指的是线程永久等待加锁变量，但这些变量永远不会解锁的情况。
死锁不止会发生在忘记解锁的情况下。当程序处理异常时，由于已经跳出当前函数执行，解锁操作可能也不会被执行，进而发生死锁。所以我们在工程实践中极其不推荐使用这种方式对变量手动加锁与解锁。
提示
在多线程开发中，还有一种情况会诱发死锁——当线程 A 锁定了线程 B 需要使用的变量，同时线程 B 锁定了线程 A 需要使用的变量时会发生。这时线程 A 会等待线程 B 释放变量，线程 B 会等待线程 A 释放变量，进而导致无限等待，例如经典的 “先交钱” 和 “先交货” 问题。E. G. Coffman Jr. 提出，死锁的产生通常需要满足以下四个条件：
- 互斥：这两个线程持有的变量不能被其他线程持有；
- 持有并等待：这两个线程在持有变量的基础上等待其他变量释放；
- 非抢占：这两个线程的变量无法被其他线程释放；
- 循环等待：两个线程循环等待彼此释放变量。
死锁的四个条件缺一不可。在实践中，最有效的避免方法是破坏 “循环等待
2. lock_guard 类
为了防止这种情况，C++11 开始为我们提供了 lock_guard 类，它在 lock_guard 变量声明时使用指定的互斥锁对线程内变量进行加锁，在线程执行结束或处理异常需要回收数据时自动解锁这些变量，它也是我们之前提到的 RAII 类。我们使用以下代码对程序进行改进，同时为 t1 引入一个异常进行测试：
#include <chrono>
#include <exception>
#include <iostream>
#include <mutex>
#include <thread>
using std::cout;
using std::endl;
using std::lock_guard;
using std::mutex;
using std::runtime_error;
using std::thread;
using std::chrono::milliseconds;
using std::this_thread::sleep_for;
int main() {
mutex m;
int i = 0;
auto t1 = thread([&]() {
try {
for (int j = 0; j < 10; j++) {
lock_guard<mutex> lock(m);
i++;
cout << i << endl;
if (i >= 5) {
throw runtime_error("我还有事 我先走了");
}
}
} catch (runtime_error &e) {
cout << e.what() << endl;
}
});
auto t2 = thread([&]() {
for (int j = 0; j < 10; j++) {
lock_guard<mutex> lock(m);
i++;
cout << i << endl;
}
});
t1.join();
t2.join();
}可以发现，即便抛出异常，lock_guard 仍然能保证变量能正常被解锁。
3. unique_lock 类
虽然 lock_guard 能帮我们解决很多问题，但它有一个致命缺陷——它会对整个线程加锁。在上面的情况下，由于我们的所有操作都在操作变量，需要持续加锁，所以我们没有意识到这个问题。但是在实际开发中，部分读取数据与处理的场合是不需要对变量加锁的，或这些锁会锁住处理完毕的无关变量。更糟糕的是，如果这些线程引入了一些无关的耗时操作，会导致所有需要使用这些锁内变量的程序被卡在门外（像极了那些在厕所里上完厕所不走玩手机的人
std::unique_lock<std::mutex> lock(cout_mutex);
std::cout << "第一段输出" << std::endl;
lock.unlock(); // 手动提前解锁（解锁线程）
// 这里执行其他耗时操作（不持有锁）
lock.lock(); // 再次手动加锁（加锁线程）
std::cout << "第二段输出" << std::endl;与 lock_guard 一样，unique_lock 类也是 RAII 类，也就是说它也能够确保线程在抛出异常或结束执行时自动解锁变量。
4. osyncstream 类
从 C++20 开始，C++ 提供了 osyncstream 类用于 cout 以及 ostream 输出的场合。我们可以将上面的代码改成下面这样：
#include <iostream>
#include <syncstream>
#include <thread>
using std::cout;
using std::endl;
using std::osyncstream;
using std::thread;
int main() {
auto t1 =
thread([&]() { osyncstream(cout) << "Hello world!1" << endl; });
auto t2 =
thread([&]() { osyncstream(cout) << "Hello world!2" << endl; });
t1.join();
t2.join();
}osyncstream 会在输出到 ostream 对象时自动为缓冲区加锁。
DLC：其他的锁机制
除了 mutex 互斥锁以外，在操作系统开发与嵌入式软件开发中，还有一种锁形式叫信号量。它通过制定一个临界区并指定初始资源数量，通过 P 操作获取信号量，通过 V 操作释放它。在 C++20 中，新增了 counting_semaphore 类，它提供了跨平台的信号量支持。在 C++20 以前的版本，则需要使用各操作系统提供的信号量实现。信号量能够实现在多个资源下的简易线程调度，它在资源量不足时自动暂停所有需要该信号量的线程，在信号量被释放时使用 FCFS（先来先服务，First come first serve
4. 原子变量
在 C++ 中，我们可能会遇到这样一种情况：变量在所有写入中都要加锁，但频繁使用 unique_lock 过于麻烦的情况。这时需要加锁的对象就不再是线程内的特定操作，而是跨越多个线程的变量。C++ 为我们提供了原子变量来解决这个问题。
要使用原子变量，我们需要包含 atomic 头文件，它为我们提供了原子变量支持。一个好原子变量的自我修养是在自己被写入时自动锁定自己，在写入结束时自动解锁。例如：
#include <atomic>
#include <chrono>
#include <iostream>
#include <mutex>
#include <thread>
using std::atomic;
using std::cout;
using std::endl;
using std::runtime_error;
using std::thread;
using std::chrono::milliseconds;
using std::this_thread::sleep_for;
int main() {
atomic<int> i{0};
auto t1 = thread([&]() {
for (int j = 0; j < 10; j++) {
++i;
}
});
auto t2 = thread([&]() {
for (int j = 0; j < 10; j++) {
++i;
}
});
t1.join();
t2.join();
cout << i << endl;
}这个程序输出的 i 的值始终是 20。
关于原子变量的命名
原子变量之所以被称作原子变量，是因为它的工作机制就像原子一样——一般情况下我们无法对原子进行再次拆分（那个扛着粒子对撞机的，说你呢，出去啊
原子变量类是一个模板类，这也就意味着它不只支持所有的标准类，还支持我们的自定义类型。
5. 条件变量与多线程同步
在一些情况下，我们可能希望指定的线程等待另一个线程的结果。这里的问题在于——除了信号量机制以外，根本没有这样一个机制来实现指定的功能。而一些问题又过于简单，使用信号量很不划算。为了应对这个问题，C++ 提供了条件变量来满足需要特定线程协作运行的场合，最经典的是生产者 - 消费者问题。在这个问题中，有一个有限队列存储数据，消费者需要等待生产者释放数据，生产者在队列满时需要等待消费者消费全部数据后继续生产。这种问题就十分适合使用条件变量解决。
首先，使用条件变量需要我们包含 condition_variable、mutex 和 thread 头文件。这样我们就可以用如下语句声明一个条件变量：
condition_variable conditon_var;在生产者 - 消费者问题中，我们还需要一个给条件变量的互斥锁，以及一个准备完毕的全局标志位。之后，我们就可以使用条件变量了。
由于条件变量需要在线程运行中途解锁，所以我们必须使用 unique_lock 传递给条件变量。条件变量的等待函数接受两个参数：
- 要使用的锁；
- 一个函数，满足我们前面说到的谓词约束。
我们使用 sleep_for 来模拟耗时操作，代码示例如下：
#include <chrono>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
std::mutex mtx;
std::condition_variable cv;
std::queue<int> data_queue;
const int MAX_SIZE = 5;
void producer() {
for (int i = 0; i < 10; ++i) {
std::this_thread::sleep_for(std::chrono::milliseconds(200));
std::unique_lock<std::mutex> lock(mtx);
// 如果队列已满，等待消费者消费
cv.wait(lock, [] { return data_queue.size() < MAX_SIZE; });
data_queue.push(i);
std::cout << "Produced: " << i << std::endl;
lock.unlock();
cv.notify_one(); // 通知消费者
}
}
void consumer() {
for (int i = 0; i < 10; ++i) {
std::unique_lock<std::mutex> lock(mtx);
// 如果队列为空，等待生产者生产
cv.wait(lock, [] { return !data_queue.empty(); });
int val = data_queue.front();
data_queue.pop();
std::cout << "Consumed: " << val << std::endl;
lock.unlock();
cv.notify_one(); // 通知生产者
}
}
int main() {
std::thread t1(producer);
std::thread t2(consumer);
t1.join();
t2.join();
return 0;
}条件变量使用 wait 来等待指定条件。当指定的条件变量调用 notify_one 或 notify_all 函数以及谓词函数条件满足时，wait 操作自动结束，线程继续执行；当指定的数据准备就绪后，可以使用 notify_one 通知一个线程继续执行或使用 notify_all 通知所有线程继续执行。当条件变量有多个线程调用 wait 时，使用 notify_one 无法指定要唤醒的线程，系统会随机选取一个线程进行唤醒。
此外，由于 wait 可能在没有 notify 唤醒时也继续线程执行，推荐在使用条件变量时始终包含谓词约束来进行二次检查。
由于条件变量也会导致死锁问题，应该在线程内保持通知和等待的顺序清晰。一个线程通常不应该对自己正在等待的条件进行 notify，除非我们在使用屏障模式或进行分阶段唤醒线程。