C++11多线程编程-std::async

C++11中提供了异步线程接口std::async，std::async是异步编程的高级封装，相对于直接使用std::thread，std::async的优势在于:

1、std::async会自动创建线程去调用线程函数，相对于低层次的std::thread，使用起来非常方便；

2、std::async返回std::future对象，通过返回的std::future对象我们可以非常方便的获取到线程函数的返回结果；

3、std::async提供了线程的创建策略，可以指定同步或者异步的方式去创建线程；

1、函数原型

C++ 11中提供如下函数原型：

template< class Function, class... Args>
std::future<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>>
    async( Function&& f, Args&&... args );

template< class Function, class... Args >
std::future<std::result_of_t<std::decay_t<Function>(std::decay_t<Args>...)>>
    async( std::launch policy, Function&& f, Args&&... args );

其中， 参数f 接收一个可调用对象(仿函数、lambda表达式、类成员函数、普通函数……)，用于异步或是同步执行。

参数policy用于指定同步执行或者异步执行可调用对象，它的可选值有三种：

1）std::launch::async：异步执行可调用对象；

2）std::launch::deferred：同步执行可调用对象；

3）std::launch::async | std::launch::deferred 可以异步或是同步，取决于具体实现。

函数返回值：

函数返回值是std::future对象，我们可以执行get、wait、wait_for、wait_until函数获取或者等待执行结果。

调用std::future对象的get函数时，如果执行的是异步执行策略，如果异步执行没有结束，get函数调用会阻塞当前当前调用线程；如果执行的是同步执行策略，只有当调用get函数时才真正执行。

调用std::future对象的wait*函数时，可能返回三种状态：

1）std::future_status::deferred：可调用对象尚未开始执行；

2）std::future_status::ready：可调用对象执行完毕；

3）std::future_status::timeout：可调用对象执行超时；

2、头文件

#include <future>

3、同步或异步读取文件内容

我们模拟异步从数据库中读取数据和同步方式从文件中读取数据，从其中可以看到std::async的使用方法。

#include <iostream>
#include <string>
#include <chrono>
#include <thread>
#include <future>
 
using namespace std::chrono;
 
std::string fetchDataFromDB(std::string recvData) {
	std::cout << "fetchDataFromDB start" << std::this_thread::get_id() << std::endl;
	std::this_thread::sleep_for(seconds(5));
	return "DB_" + recvData;
}
 
std::string fetchDataFromFile(std::string recvData) {
	std::cout << "fetchDataFromFile start" << std::this_thread::get_id() << std::endl;
	std::this_thread::sleep_for(seconds(3));
	return "File_" + recvData;
}
 
int main() {
	std::cout << "main start" << std::this_thread::get_id() << std::endl;
 
	//获取开始时间
	system_clock::time_point start = system_clock::now();
 
	std::future<std::string> resultFromDB = std::async(std::launch::async, fetchDataFromDB, "Data");
 
	//从文件获取数据
	std::future<std::string> fileData = std::async(std::launch::deferred, fetchDataFromFile, "Data");
 
	//调用get函数fetchDataFromFile才开始执行
	std::string FileData = fileData.get();
	//如果fetchDataFromDB执行没有完成，get会一直阻塞当前线程
	std::string dbData = resultFromDB.get();
	
	//获取结束时间
	auto end = system_clock::now();
 
	auto diff = duration_cast<std::chrono::seconds>(end - start).count();
	std::cout << "Total Time taken= " << diff << "Seconds" << std::endl;
 
	//组装数据
	std::string data = dbData + " :: " + FileData;
 
	//输出组装的数据
	std::cout << "Data = " << data << std::endl;
 
	return 0;
}

代码输出：

main start140677737994048
fetchDataFromFile start140677737994048
fetchDataFromDB start140677720131328
Total Time taken= 5Seconds
Data = DB_Data :: File_Data

4、设置异步数据读取超时机制

有时我们不能无限制的等待异步任务执行，可以设置超时等待时间(timeout)，当超时时间到达时，可以选择放弃等待异步任务。

如果代码中，我们设置了1s的超时设置，用于检查异步线程是否执行完毕。

#include <iostream>
#include <string>
#include <chrono>
#include <thread>
#include <future>
 
using namespace std::chrono;
 
std::string fetchDataFromDB(std::string recvData) {
 
	std::cout << "fetchDataFromDB start" << std::this_thread::get_id() << std::endl;
	std::this_thread::sleep_for(seconds(5));
	return "DB_" + recvData;
}
 
 
int main() {
 
	std::cout << "main start" << std::this_thread::get_id() << std::endl;
 
	//获取开始时间
	system_clock::time_point start = system_clock::now();
 
	std::future<std::string> resultFromDB = std::async(std::launch::async, fetchDataFromDB, "Data");
 
	std::future_status status;
	std::string dbData;
	do
	{
		status = resultFromDB.wait_for(std::chrono::seconds(1));
 
		switch (status)
		{
		case std::future_status::ready:
			std::cout << "Ready..." << std::endl;
			//获取结果
			dbData = resultFromDB.get();
			std::cout << dbData << std::endl;
			break;
		case std::future_status::timeout:
			std::cout << "timeout..." << std::endl;
			break;
		case std::future_status::deferred:
			std::cout << "deferred..." << std::endl;
			break;
		default:
			break;
		}
 
	} while (status != std::future_status::ready);
 
	
	//获取结束时间
	auto end = system_clock::now();
 
	auto diff = duration_cast<std::chrono::seconds>(end - start).count();
	std::cout << "Total Time taken= " << diff << "Seconds" << std::endl;
 
	return 0;
}

程序输出：

main start140406593357632
fetchDataFromDB start140406575482624
timeout...
timeout...
timeout...
timeout...
Ready...
DB_Data
Total Time taken= 5Seconds

5、使用std::async实现多线程并发

既然std::async可以实现异步调用，我们很容易就可以借用它实现多线程并发。

#include <iostream>
#include <vector>
#include <algorithm>
#include <numeric>
#include <future>
#include <string>
#include <mutex>
#include <thread>
 
template <typename RandomIt>
int parallel_sum(RandomIt beg, RandomIt end)
{
    std::cout << "thread id:" << std::this_thread::get_id() << std::endl;
    
    auto len = end - beg;
    if (len < 1000)
        return std::accumulate(beg, end, 0);
 
    RandomIt mid = beg + len/2;
    auto handle_me = std::async(std::launch::async,
                             parallel_sum<RandomIt>, mid, end);
    auto handle_bm = std::async(std::launch::async,
                             parallel_sum<RandomIt>, beg, mid);
    // int sum = parallel_sum(beg, mid);
    return handle_bm.get() + handle_me.get();
}
 
int main()
{
    std::vector<int> v(10000, 1);
    std::cout << "The sum is " << parallel_sum(v.begin(), v.end()) << std::endl;
}

程序输出如下：

The sum is thread id:140594794530624
thread id:140594776655616
thread id:140594768262912
thread id:140594759870208
thread id:140594672297728
thread id:140594680690432
thread id:140594663905024
thread id:140594655512320
thread id:140594647119616
thread id:140594638726912
thread id:140594269644544
thread id:140594630334208
thread id:140594278037248
thread id:140594252859136
thread id:140594261251840
thread id:140594252859136
thread id:140594236073728
thread id:140594252859136
thread id:140594261251840
thread id:140594630334208
thread id:140594244466432
thread id:140594252859136
thread id:140594227681024
thread id:140594261251840
thread id:140593875384064
thread id:140593850205952
thread id:140593858598656
thread id:140593866991360
thread id:140594647119616
thread id:140594269644544
thread id:140594672297728
10000

6、其它注意事项

在使用时需要注意，std::future对象的析构需要等待std::async执行完毕，也就是说，如下面的代码并不能实现并发。原因在于std::async的返回的std::future对象无人接收，是个临时变量，临时变量的析构会阻塞，直至std::async异步任务执行完成。

std::async(std::launch::async, []{ f(); }); // temporary's dtor waits for f()
std::async(std::launch::async, []{ g(); }); // does not start until f() completes

参考材料

https://en.cppreference.com/w/cpp/thread/async

https://blog.csdn.net/u012372584/article/details/97108417