C++11 异步操作

异步操作简介

什么是异步操作，为何会有异步操作？
在C++中，不能直接从thread.join()得到结果，必须定义个变量，在线程执行时，对这个变量赋值，然后执行join()，过程相对繁琐。

Linux中有AIO（异步IO）做异步操作，C++中如何进行异步操作？
答：是有的，C++11提供了异步操作相关的类，主要有std::future, std::promise, std::package_task。
Linux AIO参见Linux 异步IO(AIO)

• std::future 作为异步结果传输通道，可以方便获取线程函数的返回值；
• std::promise 用来包装一个值，将数据和future绑定起来，方便线程赋值；
• std::package_task 用来包装一个可调用对象，将函数和future绑定起来，便于异步调用；

头文件：

std::future

获取线程返回值的类std::future，用来访问异步操作的结果。因为异步操作需要时间，future并不能马上获取其结果，只能在未来某个时候获取期待值，因此被称为future。
可以通过同步等待的方式，通过查询future的状态（future_status），来获取异步操作的结果。

future状态有3种：
1）Deferred 异步操作还没开始；
2）Ready 异步操作已经完成；
3）Timeout 异步操作超时；

获取future异步操作的结果，有4种方式：
1）get 等待异步结束并返回结果；
2）wait 只是等待异步操作结束，没有返回值；
3）wait_for 超时等待异步操作返回结果；
4）wait_until 等待达到指定的时间点，异步操作返回结果，常配合当前时间点 + 时间段来设置目标时间点；

其同步调用基本用法为：

// 同步查询future状态的例子
// 异步求和，同步查询
int work0(int a, int b) {
    this_thread::sleep_for(chrono::seconds(4)); // 通过休眠，模拟异步操作需要一定时间完成
    return a + b;
}

std::future_status status;
std::future<int> future;

future = async(work0, 12, 34); // 注意future经常搭配std::async使用，async参见下文
/* 等待、轮询future状态 */
do {
    status = future.wait_for(chrono::seconds(1)); // 定时获取future的状态
    if (status == future_status::deferred) { // 异步操作还没开始
        cout << "deferred" << endl;
    }
    else if (status == future_status::timeout) { // 异步操作超时
        cout << "timeout" << endl;
    }
    else if (status == future_status::ready) { // 异步操作已完成
        cout << "ready" << endl;
    }
} while (status != future_status::ready);

std::promise

协助线程赋值的类std::promise，将数据和future绑定。在线程函数中，为外面传入的promise对象赋值，线程函数执行完毕后，就可以通过promise的future获取该值。
注意：future是定义在promise内部的成员。

get_future函数

函数返回一个与promise共享状态相关联的future对象。返回的future对象可以访问由promise对象设置在共享状态上的值或者某个异常对象。
promise对象通常会在某个时间点准备好（设置一个值或者一个异常对象），然后在另一个线程中，用future对象的get获取值。

future<_Ty> get_future()

set_value函数

设置共享状态值，此后promise的共享状态标志变为ready

void set_value(const _Ty& _Val)

set_exception函数

为promise设置异常，此后promise的共享状态标志变为ready

void set_exception(_XSTD exception_ptr _Exc)

promise基本用法

promise<int> pr;
future<int> f = pr.get_future();
thread t([](promise<int> &p) { p.set_value_at_thread_exit(9); }, ref(pr)); // lambda是匿名函数，ref(pr)是该函数实参
t.detach();
try {
    auto r = f.get();
    cout << "get future from promise: " << r << endl;
}
catch (const exception& e) {
    cout << "exception: " << e.what() << endl;
}

可以将上面线程函数由lambda改成使用普通的线程函数：

thread t([](promise<int> &p) { p.set_value_at_thread_exit(9); }, ref(pr));
// <=>
thread t(work1, ref(pr));
void work1(promise<int> &pr) {
    this_thread::sleep_for(chrono::seconds(3));
#if 0
    try {
        throw runtime_error("Runtime error");
    }
    catch (...) {
        pr.set_exception(current_exception());
    }
#else
    pr.set_value(9);
#endif
}

promise有set_value和set_value_at_thread_exit，都可以用来对promise赋值，区别在于后者要求赋值的promise对象不能被销毁（如不能使用右值传值，然后在内部用std::move构造临时promise对象），而前者没有这个要求。
参见关于std::promise的set_value_at_thread_exit | CSDN

std::package_task

可调用对象的包装类std::package_task，包装了一个可调用的目标类（如function, lambda expression, bind expression, another function object），将其与future绑定，以便异步调用。
promise和package_task类似，promise保存了一个共享状态的值，package_task保存的是一个可调用对象（如函数）。

package_task基本用法

packaged_task<int()> task([]() {return 7; }); // 包装可调用对象（lambda expression）
thread t2(ref(task));
future<int> f1 = task.get_future();
auto r1 = f1.get();
t2.detach();

cout << r1 << endl;

std::promise, std::package_task, std::future三者关系

future提供异步操作结果访问机制，跟线程一个级别，属于低层次对象。
promise和package_task是在future的高一层，内部封装了future以便访问异步操作结果。promise内部包装的是一个值，package_task内部包装的是一个可调用对象。当需要线程中的某个值时，使用promise；当需要获得一个异步操作的返回值时，使用package_task。
简单来说，package_task会自动将异步操作的结果保存到promise对象中，而promise需要在线程函数中手动设置promise值。

另外，可以用future保存异步操作结果，但future的copy函数是禁用的，只能使用move函数。而shared_future是可以copy的。因此，当要把future保存到容器中时，请使用shared_future。

package_task和shared_future基本用法

int func(int x) {
    return x + 2;
}

packaged_task<int(int)> task(func);
future<int> ft = task.get_future();

thread(std::move(task), 2).detach(); // task作为线程函数
int res = ft.get();
cout << res << endl;  // 打印4

// future不可copy，无法放入容器。要放入容器，使用shared_future
vector<shared_future<int>> vec;
auto f = std::async(launch::async, [](int a, int b) { return a + b; }, 2, 3).share(); // 注意这里的share()将future转化为shared_future
vec.push_back(f);
cout << "The shared_future result is " << vec[0].get() << endl; // 打印5

std::async

线程异步操作函数std::async，比promise, package_task, thread更高一层，可用来创建异步task，其返回结果保存到future对象中。当需要取得异步结果时，调用future.get()即可。如果不关注结果，只是简单等任务完成，可以使用future.wait()

async原型：

async(std::launch::async | std::launch::deferred, f, args,...);

• 第一个参数是线程的创建策略，默认是std::launch::async。
  1）std::launch::async 调用async时就开始创建线程；
  2）std::launch::deferred 延迟加载方式创建线程。调用async时不创建线程，直到调用了future的get或者wait，才创建线程。

• 第二个参数是线程函数

• 第三个参数是传入线程函数的参数

async使用示例

// 使用launch::async策略创建线程，并阻塞get其值
future<int> f1 = async(launch::async, []() {return 8; });
cout << f1.get() << endl; // 打印8

// 使用launch::async策略创建线程，并轮询future状态
future<int> f2 = async(launch::async, []() { 
    this_thread::sleep_for(chrono::seconds(3));
    return 8; });
cout << "waiting..." << endl;

future_status status;
do {
    status = f2.wait_for(chrono::seconds(1));
    if (status == future_status::deferred) { // 异步操作尚未完成
        cout << "deferred" << endl;
    }
    else if (status == future_status::timeout) { // 异步操作超时
        cout << "timeout" << endl;
    }
    else if (status == future_status::ready) {
        cout << "ready!" << endl;
    }
} while (status != future_status::ready);
cout << "result is " << f2.get() << endl;

// 使用launch::deferred策略创建线程，并轮询future状态
future<int> f3 = async(launch::deferred, []() { return 3; });
future_status status3;
do {
    status3 = f3.wait_for(chrono::seconds(1));
    if (status3 == future_status::deferred) {
        cout << "deferred" << endl;
    }
} while (status3 != future_status::ready);

参考

[1]祁宇. 深入应用C++11 : 代码优化与工程级应用 : In-Depth C++11 : code optimization and engineering level application[M]. 机械工业出版社, 2015.