C++ 多线程编程建议之 C++ 对多线程/并发的支持（下）

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒

前言：

本文承接前文 C++ 对多线程/并发的支持（上），翻译自 C++ 之父 Bjarne Stroustrup 的 C++ 之旅（A Tour of C++）一书的第 13 章 Concurrency。本文将继续介绍 C++ 并发中的 future/promise，packaged_task 以及 async() 的用法。

1、通信任务

标准库还在头文件 <future> 中提供了一些机制，能够让编程人员基于更高的抽象层次任务来开发，而不是直接使用低层的线程、锁：

future 和 promise：用于从任务（另一个线程）中返回一个值
packaged_task：帮助启动任务，封装了 future 和 promise，并且建立两者之间的关联
async() ：像调用一个函数那样启动一个任务。形式最简单，但也最强大！

1.1 future 和 promise

future 和 promise 可以在两个任务之间传值，而无需显式地使用锁，实现了高效地数据传输。其基本想法很简单：当一个任务向另一个任务传值时，把值放入 promise，通过特定的实现，使得值可以通过与之关联的 future 读出（一般谁启动了任务，谁从 future 中取结果）。

假如有一个 future<X> 叫 fx，我们可以通过 get() 获取类型 X 的值：


X v = fx.get(); // if necessary, wait for the value to get computed

如果值还没有计算出，则调用 get() 的线程阻塞，直到有值返回。如果值无法计算出，get()可能抛出异常。

promise 的主要目的是提供一个简单的“put”的操作（set_value 或 set_exception），和 future 的 get() 相呼应。

如果你有一个 promise，需要发送一个类型为 X 的结果到一个 future，你要么传递一个值，要么传递一个异常。举个例子：


void f(promise<X>& px) // 一个任务：把结果放入 px
{
    try {
        X res;
        // 计算 res 的值
        px.set_value(res);
    }
    catch(...) { // 如果无法计算 res 的值
        px.set_exception(current_exception()); // 传异常到 future 的线程
    }
}

current_exception() 即捕获到的异常。

要处理通过 future 传递的异常，get() 的调用者必须在什么地方捕获，例如：


void g(future<X>& fx) // 一个任务；从 fx 提取结果
{
    try {
        X v = fx.get(); // 如有必要，等待值计算完成
        // 使用 v
    }
    catch(...){ // 无法计算 v
        // 错误处理
    }
}

如果 g() 不需要自己处理错误，代码可以进一步简化：


void g(future<X>& fx) // 一个任务；从 fx 提取结果
{
    X v = fx.get(); // 如有必要，等待值计算完成
    // 使用 v
}

思考：future 和 promise 是怎么关联起来的？

1.2 packaged_task

如何把 future 放入一个需要结果的任务，并且把与之关联的、产生结果的 promise 放入线程？packaged_task 可以简化任务的设置，关联 future/promise。packaged_task 封装了把返回值或异常放入 promise 的操作，并且调用 packaged_task 的 get_future() 方法，可以得到一个与 promise 关联的 future。举个例子，我们可以设置两个任务，借助标准库的 accumulate() 分别累加 vector<double> 的前后部分：


double accum (double* beg, double* end, double init) // 计算以 init 为初值，[beg,end) 的和
{
    return accumulate(beg,end,init);
}

double comp2(vector<double>& v)
{
    using Task_type = double(double*,double*,double); // 任务的类型

    packaged_task<Task_type> pt0 {accum}; // 打包任务（即 accum）
    packaged_task<Task_type> pt1 {accum};

    future<double> f0 {pt0.get_future()}; // 取得 pt0 的 future
    future<double> f1 {pt1.get_future()}; // 取得 pt1 的 future

    double* first = &v[0];
    thread t1{move(pt0),first,first+v.size()/2,0};          // 为 pt0 启动线程
    thread t2{move(pt1),first+v.size()/2,first+v.size(),0}; // 为 pt1 启动线程

    return f0.get() + f1.get();
}

packaged_task 模板以任务的类型（Task_type，double(double*,double*,double) 的别名）作为其模板参数，以任务（accum）作为其构造函数的参数。move() 操作是必要的，因为 packaged_task 不可拷贝（只能移动）。packaged_task 不可拷贝是因为它是一个资源处理程序（resource handler），拥有 promise 的所有权，并且（间接地）负责与之关联的任务可能拥有的资源。

请注意，这里的代码没有显式地使用锁：我们能够专注于要完成的任务，而不是来管理它们通信的机制。这两个任务在不同的线程中执行，具有了潜在的并发性。

1.3 async()

我在本章所追求的思路，最简单，但也非常强大：把任务看成是一个恰巧可能和其他任务同时运行的函数。这并不是 C++ 标准库所支持的唯一模型，但它能很好地满足各类广泛的需求。其他更微妙、棘手的模型，如依赖于共享内存的编程风格也可以根据实际需要使用。

要启动潜在异步执行的任务，我们可以用 async()：


double comp4(vector<double>& v) // 如果 v 足够大，派生多个任务
{
    if(v.size()<10000) // 犯得着用并发吗？
        return accum(v.begin(),v.end(),0);
    
    auto v0 = &v[0];
    auto sz = v.size();
    
    auto f0 = async(accum,v0,v0+sz/4,0.0);
    auto f1 = async(accum,v0+sz/4,v0+sz/2,0.0);
    auto f2 = async(accum,v0+sz/2,v0+sz*3/4,0.0);
    auto f3 = async(accum,v0+sz*3/4,v0+sz,0.0);
    
    return f0.get()+f1.get()+f2.get()+f3.get(); // 收集 4 部分的结果，求和
}

大体上，async() 把“调用部分”和“获取结果部分“分离开来，并且将两者和实际执行的任务分离。使用 async() 你不需要考虑线程、锁；你只要从任务（潜在地、异步地计算结果）的角度去考虑就可以了。async() 也有明显的限制：使用了共享资源、需要上锁的任务无法使用 async() ，你甚至不知道会用到多少线程，这完全是由 async() 决定的，它会根据调用时系统可用资源的情况，决定使用多少线程。例如，async() 在决定使用几个线程前，会检查有多少核心（处理器）空闲。

示例代码中的猜测计算开销和启动线程的相对开销（v.size()<10000）只是一个很原始、粗略的性能估计。这里不适合展开讨论怎么去管理线程，但这个估计仅仅是一个简单（可能很烂）的猜测。

请注意，async()不仅仅是专门用于并行计算、提高性能的机制。例如，它也能用于派生任务，从用户获取输入，让“主程序”忙其他事情。

2、建议

使用并发改善响应性和吞吐量
尽可能在最高级别的抽象上工作（比如优先考虑 async、packaged_task 而不是 thread、mutex）
考虑使用进程作为线程的替代方案
标准库的并发支持是类型安全的
内存模型把多数程序员从考虑机器架构的工作中解放出来
内存模型使得内存的表现和我们的预期基本一致
原子操作为无锁编程提供了可能性
把无锁编程留给专家
有时顺序操作比起并发更简单、更快
避免数据竞争（不受控地同时访问可变数据）
std::thread 是类型安全的系统线程接口
用 join() 等待一个线程结束
尽量避免显式共享数据
用 unique_lock 管理 mutexes
用 lock() 一次性获取多个锁
用 condition_variable 管理线程之间的通信
从（可以并行执行的）任务的角度思考，而非线程
不要低估“简单性”的价值
选择 packaged_task 和 future，而不是直接使用 thread 和 mutex
用 promise 返回结果，从 future 获取结果
用 packaged_task 处理任务抛出的异常或返回值
用 packaged_task 和 future 来表示对外部服务的请求，以及等待其回复
用 async() 启动简单的任务