本文介绍C++中异步编程相关的基础操作类，以及借鉴promise and future思想解决回调地狱介绍。
std::thread and std::jthread
std::thread为C++11引入，一个简单的例子如下：
class Worker final {
public:
void Execute()
{
std::cout << __FUNCTION__ << std::endl;
}
};
int main()
{
Worker w;
auto thread = std::thread(&Worker::Execute, &w);
thread.join();
return 0;
}这里如果少调用了thread.join()，类析构了但线程仍在运行，导致代码异常终止；
添加封装Worker：
class Worker final {
public:
Worker()
{
m_thread = std::thread(&Worker::execute, this);
}
~Worker()
{
m_thread.join();
}
private:
void execute()
{
std::cout << __FUNCTION__ << std::endl;
}
private:
std::thread m_thread;
};这里应用了 RAII ，封装了Worker类在析构时自动调用join函数等待线程运行结束。
而std::jthread为C++20引入，是对std::thread的扩展：
/// A thread that can be requested to stop and automatically joined.
~jthread()
{
if (joinable())
{
request_stop();
join();
}
}Worker w;
auto thread = std::jthread(&Worker::Execute, &w); // 无需再单独调用join同时std::jthread增加了能够主动停止线程执行的新特性，修改上述例子为：
class Worker final {
public:
void operator()(std::stop_token st)
{
while (!st.stop_requested()) {
sleep(1);
std::cout << __FUNCTION__ << std::endl;
}
}
};
int main()
{
Worker w;
auto thread = std::jthread(w);
sleep(3);
std::cout << "request stop" << std::endl;
thread.request_stop();
return 0;
}进一步，我们打开std::jthread的源码可以看到其实现原理，stop_token会作为入参送给可调用对象，通过stop_source获取到stop_token对象，两者共享stop_token::_Stop_state_ref状态；
class jthread
{
public:
template<typename _Callable, typename... _Args,
typename = enable_if_t<!is_same_v<remove_cvref_t<_Callable>,
jthread>>>
explicit
jthread(_Callable&& __f, _Args&&... __args)
: _M_thread{_S_create(_M_stop_source, std::forward<_Callable>(__f),
std::forward<_Args>(__args)...)}
{ }
// ...
[[nodiscard]] stop_token
get_stop_token() const noexcept
{
return _M_stop_source.get_token();
}
bool request_stop() noexcept
{
return _M_stop_source.request_stop(); // 通知线程停止
}
private:
template<typename _Callable, typename... _Args>
static thread
_S_create(stop_source& __ssrc, _Callable&& __f, _Args&&... __args)
{
if constexpr(is_invocable_v<decay_t<_Callable>, stop_token,
decay_t<_Args>...>)
return thread{std::forward<_Callable>(__f), __ssrc.get_token(),
std::forward<_Args>(__args)...}; // 将stop_token作为入参送入到可调用对象
else
{
static_assert(is_invocable_v<decay_t<_Callable>,
decay_t<_Args>...>,
"std::thread arguments must be invocable after"
" conversion to rvalues");
return thread{std::forward<_Callable>(__f),
std::forward<_Args>(__args)...};
}
}
stop_source _M_stop_source;
// 通过stop_source获取到stop_token对象，两者共享stop_token::_Stop_state_ref状态
thread _M_thread;
}；std::async
但软件线程是有限的资源，当试图创建的线程数量大于系统能够提供的最大数量，则会抛std::system_error异常，因此使用std::thread就需要考虑这种异常处理；
使用std::async则将线程管理的责任转交给标准库的实现者，使用std::async时系统不保证会创建一个新的软件线程，基本用法如下：
int main()
{
std::cout << "threadId:" << std::this_thread::get_id() << std::endl;
std::future<int> f1 = std::async(std::launch::async, [](){
std::cout << "threadId:" << std::this_thread::get_id() << std::endl;
return 1;
}); // 创建新的线程
std::cout << f1.get() << std::endl;
std::future<int> f2 = std::async(std::launch::deferred, [](){
std::cout << "threadId:" << std::this_thread::get_id() << std::endl;
return 2;
});
std::cout << "wait:" << std::endl;
f2.wait(); // 在主线程调用回调，没有创建新的线程
std::cout << f2.get() << std::endl;
std::future<int> f3 = std::async(std::launch::async, [](){
sleep(5);
return 3;
});
std::future_status status;
do {
status = f3.wait_for(std::chrono::seconds(1));
if (status == std::future_status::timeout) {
std::cout << "timeout" << std::endl;
} else if (status == std::future_status::ready) {
std::cout << "ready" << std::endl;
}
} while (status != std::future_status::ready);
return 0;
}std::async有两种启动策略：
- std::launch::async:在调用async时就开始创建线程
- std::launch::deferred：延迟运行，函数只有在调用了get或者wait时才会运行；
默认启动策略是std::launch::async|std::launch::deferred，运行函数以同步或者异步的方式运行，这里有线程管理组件承担起负载均衡的责任【3】
注意：std::future的get()为移动语义，不能进行多次调用，否则会抛异常
改成std::shared_future则是拷贝语义，可以进行多次调用
std::promise and std::future
std::future在上一节中已经使用，而std::promise将数据与std::future绑定，为获取线程函数中的某个值提供便利：
std::promise<int> p;
{
auto thread = std::jthread([&p](){
p.set_value(10);
});
}
auto fu = p.get_future();
std::cout << fu.get() << std::endl; // 获取promise在线程中set的值std::package_task
std::package_task用于封装可调用对象，并且可获取future以便异步获取可调用对象的返回值：
auto func = [](){
return 10;
};
std::packaged_task<int()> task(func);
auto fu = task.get_future();
{
auto thread = std::jthread(std::ref(task));
}
std::cout << fu.get() << std::endl;callback hell
借鉴future and promise思想，可以用于解决“回调地狱的问题”，什么是“回调地狱”？
template<typename Callback>
void Async(Callback&& call) // 异步处理
{
// ...
}
// callX为可调用对象需要进行异步处理，并且可调用对象依赖上一个可调用的输出
// outputA = callA(input) // outputB = callB(outputA) // outputC = callC(outputB) // outputD = callD(outputC)
// 传统的实现方式可以为
Async([](input){
outputA = callA(input);
Async([outputA](){
outputB = callB(outputA);
Async([outputB](){
outputC = callC(outputB);
Async([outputC](){
outputD = callD(outputC);
...
});
});
});
})上述代码中异步回调函数会出现不断的嵌套， 形成一个横向的金字塔，对于代码阅读者带来不必要的负担，不利用维护与扩展。
Facebook开源的folly future库、ananas都提供了 promise/future技术的解决方案，可以实现类似如下的效果：
future
.then(&loop, [](){ return ouputA; })
.then(&loop, [](ouputA){ return ouputB; })
.then(&loop, [](ouputB){ return ouputC; })
.then(&loop, [](ouputC){ return ouputD; })详细实现源码下次再展开分析~
参考资料
【1】https://github.com/loveyacper/ananas
【2】https://en.cppreference.com/w/cpp/thread/async
【3】Effective Modern C++