编译期的烧脑我们先告个段落今天我们开始讲一个全新的话题——并发concurrency。为什么要使用并发编程在本世纪初之前大部分开发人员不常需要关心并发编程用到的时候也多半只是在单处理器上执行一些后台任务而已。只有少数为昂贵的工作站或服务器进行开发的程序员才会需要为并发性能而烦恼。原因无他程序员们享受着摩尔定律带来的免费性能提升而高速的 Intel 单 CPU 是性价比最高的系统架构可到了 2003 年左右大家骤然发现“免费午餐”已经结束了 [1]。主频的提升停滞了在 2001 年Intel 已经有了主频 2.0 GHz 的 CPU而 18 年后我现在正在使用的电脑主频也仍然只是 2.5 GHz虽然从单核变成了四核。服务器、台式机、笔记本、移动设备的处理器都转向了多核计算要求则从单线程变成了多线程甚至异构——不仅要使用 CPU还得使用 GPU。如果你不熟悉进程和线程的话我们就先来简单介绍一下它们的关系。我们编译完执行的 C 程序那在操作系统看来就是一个进程了。而每个进程里可以有一个或多个线程每个进程有自己的独立地址空间不与其他进程分享一个进程里可以有多个线程彼此共享同一个地址空间。堆内存、文件、套接字等资源都归进程管理同一个进程里的多个线程可以共享使用。每个进程占用的内存和其他资源会在进程退出或被杀死时返回给操作系统。并发应用开发可以用多进程或多线程的方式。多线程由于可以共享资源效率较高反之多进程默认不共享地址空间和资源开发较为麻烦在需要共享数据时效率也较低。但多进程安全性较好在某一个进程出问题时其他进程一般不受影响而在多线程的情况下一个线程执行了非法操作会导致整个进程退出。我们讲 C 里的并发主要讲的就是多线程。它对开发人员的挑战是全方位的。从纯逻辑的角度并发的思维模式就比单线程更为困难。在其之上我们还得加上编译器和处理器的重排问题原子操作和数据竞争互斥锁和死锁问题无锁算法条件变量信号量……即使对于专家并发编程都是困难的上面列举的也只是部分难点而已。对于并发的基本挑战Herb Sutter 在他的 Effective Concurrency 专栏给出了一个较为全面的概述 。要对 C 的并发编程有全面的了解则可以阅读曼宁出版的 C Concurrency in Action有中文版但翻译口碑不好[3]。而我们今天主要要介绍的则是并发编程的基本概念包括传统的多线程开发以及高层抽象 future姑且译为未来量的用法。基于 thread 的多线程开发我们先来看一个使用 thread 线程类的简单例子#include chrono #include iostream #include mutex #include thread using namespace std; mutex output_lock; void func(const char* name) { this_thread::sleep_for(100ms); lock_guardmutex guard{ output_lock}; cout I am thread name \n; } int main() { thread t1{func, A}; thread t2{func, B}; t1.join(); t2.join(); }这是某次执行的结果I am thread BI am thread A一个平台细节在 Linux 上编译线程相关的代码都需要加上 -pthread 命令行参数。Windows 和 macOS 上则不需要。代码是相当直截了当的执行了下列操作传递参数起两个线程两个线程分别休眠 100 毫秒使用互斥量mutex锁定 cout 然后输出一行信息主线程等待这两个线程退出后程序结束以下几个地方可能需要稍加留意一下thread 的构造函数的第一个参数是函数对象后面跟的是这个函数所需的参数。thread 要求在析构之前要么 join阻塞直到线程退出要么 detach放弃对线程的管理否则程序会异常退出。sleep_for 是 this_thread 名空间下的一个自由函数表示当前线程休眠指定的时间。如果没有 output_lock 的同步输出通常会交错到一起。建议你自己运行一下并尝试删除 lock_guard 和 join 的后果。thread 不能在析构时自动 join 有点不那么自然这可以算是一个缺陷吧。在 C20 的 jthread到来之前我们只能自己小小封装一下了。比如class scoped_thread { public: template typename... Arg scoped_thread(Arg... arg) : thread_( std::forwardArg(arg)...) {} scoped_thread( scoped_thread other) : thread_( std::move(other.thread_)) {} scoped_thread( const scoped_thread) delete; ~scoped_thread() { if (thread_.joinable()) { thread_.join(); } } private: thread thread_; };这个实现里有下面几点需要注意我们使用了可变模板和完美转发来构造 thread 对象。thread 不能拷贝但可以移动我们也类似地实现了移动构造函数。只有 joinable已经 join 的、已经 detach 的或者空的线程对象都不满足 joinable的 thread 才可以对其调用 join 成员函数否则会引发异常。使用这个 scoped_thread 类的话我们就可以把我们的 main 函数改写成int main() { scoped_thread t1{func, A}; scoped_thread t2{func, B}; }这虽然是个微不足道的小例子但我们已经可以发现执行顺序不可预期或者说不具有决定性。如果没有互斥量的帮助我们连完整地输出一整行信息都成问题。我们下面就来讨论一下互斥量。mutex互斥量的基本语义是一个互斥量只能被一个线程锁定用来保护某个代码块在同一时间只能被一个线程执行。在前面那个多线程的例子里我们就需要限制同时只有一个线程在使用 cout否则输出就会错乱。目前的 C 标准中事实上提供了不止一个互斥量类。我们先看最简单、也最常用的 mutex 类 [6]。mutex 只可默认构造不可拷贝或移动不可赋值主要提供的方法是lock锁定锁已经被其他线程获得时则阻塞执行try_lock尝试锁定获得锁返回 true在锁被其他线程获得时返回 falseunlock解除锁定只允许在已获得锁时调用你可能会想到如果一个线程已经锁定了某个互斥量再次锁定会发生什么对于 mutex回答是危险的未定义行为。你不应该这么做。如果有特殊需要可能在同一线程对同一个互斥量多次加锁就需要用到递归锁 recursive_mutex 了 [7]。除了允许同一线程可以无阻塞地多次加锁外也必须有对应数量的解锁操作recursive_mutex 的其他行为和 mutex 一致。除了 mutex 和 recursive_mutexC 标准库还提供了timed_mutex允许锁定超时的互斥量recursive_timed_mutex允许锁定超时的递归互斥量shared_mutex允许共享和独占两种获得方式的互斥量shared_timed_mutex允许共享和独占两种获得方式的、允许锁定超时的互斥量这些我们就不做讲解了需要的请自行查看参考资料 。另外 头文件中也定义了锁的 RAII 包装类如我们上面用过的 lock_guard。为了避免手动加锁、解锁的麻烦以及在有异常或出错返回时发生漏解锁我们一般应当使用 lock_guard而不是手工调用互斥量的 lock 和 unlock 方法。C 里另外还有 unique_lockC11和 scoped_lockC17提供了更多的功能你在有更复杂的需求时应该检查一下它们是否合用。执行任务返回数据如果我们要在某个线程执行一些后台任务然后取回结果我们该怎么做呢比较传统的做法是使用信号量或者条件变量。由于 C17 还不支持信号量我们要模拟传统的做法只能用条件变量了。由于我的重点并不是传统的做法条件变量我就不展开讲了而只是展示一下示例的代码。#include chrono #include condition_variable #include functional #include iostream #include mutex #include thread #include utility using namespace std; class scoped_thread { … // 定义同上略 }; void work(condition_variable cv, mutex cv_mut, bool result_ready, int result) { // 假装我们计算了很久 this_thread::sleep_for(2s); result 42; { unique_lock lock{cv_mut}; result_ready true; } cv.notify_one(); } int main() { condition_variable cv; mutex cv_mut; bool result_ready false; int result; scoped_thread th{ work, ref(cv), ref(cv_mut), ref(result_ready), ref(result)}; // 干一些其他事 cout I am waiting now\n; unique_lock lock{cv_mut}; cv.wait(lock, [] { return result_ready; }); cout Answer: result \n; }可以看到为了这个小小的“计算”我们居然需要定义 6 个变量线程、条件变量、互斥量、单一锁、结果状态变量和结果变量。我们也需要用 ref 模板来告诉 thread 的构造函数我们需要引用因为 thread 默认复制或移动所有的参数作为线程函数的参数。这种复杂性并非逻辑上的复杂性而只是实现导致的不是我们希望的写代码的方式。下面我们就看看更高层的抽象未来量 future可以如何为我们简化代码。future我们先把上面的代码直接翻译成使用 async它会返回一个 future#include chrono #include future #include iostream #include thread using namespace std; int work() { // 假装我们计算了很久 this_thread::sleep_for(2s); return 42; } int main() { auto fut async(launch::async, work); // 干一些其他事 cout I am waiting now\n; cout Answer: fut.get() \n; }完全同样的结果代码大大简化变量减到了只剩一个未来量还不赖吧我们稍稍分析一下work 函数现在不需要考虑条件变量之类的实现细节了专心干好自己的计算活、老老实实返回结果就可以了。调用 async 可以获得一个未来量launch::async 是运行策略告诉函数模板 async 应当在新线程里异步调用目标函数。在一些老版本的 GCC 里不指定运行策略默认不会起新线程。async 函数模板可以根据参数来推导出返回类型在我们的例子里返回类型是futureint。在未来量上调用 get 成员函数可以获得其结果。这个结果可以是返回值也可以是异常即如果 work 抛出了异常那 main 里在执行 fut.get() 时也会得到同样的异常需要有相应的异常处理代码程序才能正常工作。这里有两个要点从代码里看不出来我特别说明一下一个 future 上只能调用一次 get 函数第二次调用为未定义行为通常导致程序崩溃对于值类型的结果get 会以移动的方式来返回。这样一来自然一个 future 是不能直接在多个线程里用的。需要的话上面两个问题是可以解决的。你可以直接拿 future 来移动构造一个 shared_future或者调用 future 的 share 方法来生成一个 shared_future这样结果就可以在多个线程里使用。此外对于非 void、非引用的返回结果类型shared_future 的 get 永远给我们一个 const 引用这样的设计也就允许我们多次调用 get 了。promise我们上面用 async 函数生成了未来量但这不是唯一的方式。另外有一种常用的方式是 promise 我称之为“承诺量”。我们同样看一眼上面的例子用 promise 该怎么写#include chrono #include future #include iostream #include thread #include utility using namespace std; class scoped_thread { … // 定义同上略 }; void work(promiseint prom) { // 假装我们计算了很久 this_thread::sleep_for(2s); prom.set_value(42); } int main() { promiseint prom; auto fut prom.get_future(); scoped_thread th{work, move(prom)}; // 干一些其他事 cout I am waiting now\n; cout Answer: fut.get() \n; }promise 和 future 在这里成对出现可以看作是一个一次性管道有人需要兑现承诺往 promise 里放东西set_value有人就像收期货一样到时间去 future写到这里想到期货英文不就是 future 么是不是该翻译成期货量呢里拿get就行了。我们把 prom 移动给新线程这样老线程就完全不需要管理它的生命周期了。就这个例子而言使用 promise 没有 async 方便但可以看到这是一种非常灵活的方式你不需要在一个函数结束的时候才去设置 future 的值。仍然需要注意的是一组 promise 和 future 只能使用一次既不能重复设也不能重复取。promise 和 future 还有个有趣的用法是使用 void 类型模板参数。这种情况下两个线程之间不是传递参数而是进行同步当一个线程在一个 future 上等待时使用 get() 或 wait()另外一个线程可以通过调用 promise 上的 set_value() 让其结束等待、继续往下执行。有兴趣的话你可以自己试一下我就不给例子了。packaged_task我们最后要讲的一种 future 的用法是打包任务 packaged_task我们同样给出完成相同功能的示例让你方便对比一下#include chrono #include future #include iostream #include thread #include utility using namespace std; class scoped_thread { … // 定义同上略 }; int work() { // 假装我们计算了很久 this_thread::sleep_for(2s); return 42; } int main() { packaged_taskint() task{work}; auto fut task.get_future(); scoped_thread th{move(task)}; // 干一些其他事 this_thread::sleep_for(1s); cout I am waiting now\n; cout Answer: fut.get() \n; }打包任务里打包的是一个函数模板参数就是一个函数类型。跟 thread、future、promise 一样packaged_task 只能移动不能复制。它是个函数对象可以像正常函数一样被执行也可以传递给 thread 在新线程中执行。它的特别地方自然也是你可以从它得到一个未来量了。通过这个未来量你可以得到这个打包任务的返回值或者至少知道这个打包任务已经执行结束了。内容小结今天我们看了一下并发编程的原因、难点以及 C 里的进行多线程计算的基本类包括线程、互斥量、未来量等。这些对象的使用已经可以初步展现并发编程的困难但更麻烦的事情还在后头呢……