c++11的异步线程操作

2023-07-22 10:57:22

一、异步线程

无论是在哪种语言中，都会面临异步操作的问题。基本上异步操作的实现可以大致分为系统级和应用级（封装的库也算应用级）。系统级一般是通过中断或者线程实现，在应用层面上一般是通过线程来实现。异步操作的目的是为了提高响应的并发量和控制访问的安全性以及健壮性。说的再直白一些，就是把访问过程，处理过程和响应过程分离。
异步是相对于同步来说，同步相当于一问一答，必须实现，假如你去银行办理业务，你问柜台的小姐姐一句话，半天才回复你，估计你就怒了。但是如果你去带着一些木料去定作家具，就不愿意等在那儿，而愿意等弄好了，给你打个电话，你再去取。其实好多人学了很多年计算机编程，并没有明白一个道理，计算机编程其实就是现实世界的一种映射，或者准确一点儿说，是现实世界一部分的可以数字逻辑化的映射。
而实际情况中，线程的同步和异步，都是要针对具体的应用场景来说的。不过随着应用场景的越来越复杂，高并发和大数据量通信带的结果就是，异步编程越来越复杂，应用也越来越广泛。为了解决这个问题不同的语言和框架都提出了的解决方式，除最初的线程模拟，到封装，再到后来Go语言等提出的协程等，都朝着一个目标前进，那就是不断的减少异步编程的复杂度。

二、初级阶段

先看一个线程模拟的异步操作：

#include <iostream>
#include <thread>
#include <functional>

int main()
{
    int num = 0;


    std::thread th = std::thread([&](std::function<void(int d)> func) {
        while (num < 1000000)
        {
            num++;
        }

        func(num);
    }, ([](int d) { std::cout << d << std::endl; }));


    std::cout << "master going......" << std::endl;
    system("pause");
}

执行结果：

master going......
1000000
请按任意键继续. . .

程序很简单，就是直接操作一个创建一个线程，线程执行一个复杂的任务，完成后，把结果返回到调用线程。这就是异步的一个典型的应用，你把木材（数据）丢到工厂（线程）定制，完成后把家具（结果）返回给你。

三、std::async及相关

而在c++11以前，c++标准库是没有提供线程库的，线程编程只能使用和OS相关的API来创建和使用。学习成本和应用成本都是很高的，多线程编程即使到现在仍然是编程中相对来说较为复杂的一部分。从c++11开始，提供了std::thread，在此基础之上，为了简化异步编程，提供std::async这个模板类。相应的，为了进行异步多线程间的交互提供了std::future,std::promise,std::packaged_task这几个模板。
简单来说std::promise提供初步的多线程间的数据操作，std::future可以得到这些操作的结果，而在实际应用中不仅有简单的数值操作，有时候儿还需要对函数的运行进行处理（特别是函数运行结果），这时候就得需要std::packaged_task了。其实这两者可以从另外一个抽象层次上统一起来，就是这个操作的对象既可以是基础值也可以是一个函数，动态映射即可，但是，这不仅会消耗性能，还会增加应用的复杂度，这会不会是c++标准制定者有所考虑呢？不得而知。
而std::async又可以看作对上述三个的进一步的抽象，让他们应用起来更简单清晰。它的参数主要说明如下：
std::launch::deferred延迟调用，延迟到future对象调用get()或者wait()的时候才执行函数，否则不会执行。
std::launch::async：强制这个异步任务在新线程上执行，即系统要创建一个新线程来执行相关函数;
std::launch::async |std::launch::deferred “ |”符号代表二选一，都有可能。
不带参数；只有函数名；默认 std::launch::async |std::launch::deferred，看具体平台的默认设置。

更详细的说明可以看：
https://en.cppreference.com/w/cpp/thread/async

四、应用

看这几个的应用：

#include <iostream>
#include <thread>
#include <functional>

#include <iostream>
#include <vector>
#include <algorithm>
#include <numeric>
#include <future>
#include <string>
#include <mutex>
#include <chrono>

int func_async()
{
    std::cout << "run async" << std::endl;
                //延时为了模拟异步操作
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    return 100;
}

int TestAsync()
{
    //测试一下不同参数效果
    auto h = std::async(std::launch::async/*std::launch::deferred*/, func_async);
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    std::cout << "strat async get :" << h.get()<< std::endl;

    return 0;
}

void func_promise(std::promise<int>& p) {
    std::this_thread::sleep_for(std::chrono::milliseconds(3000));
    p.set_value(100);                         
}

int TestPromise()
{
    std::promise<int> p;
    std::thread t(func_promise, std::ref(p));

    std::future<int> f = p.get_future();
    std::cout << "get promis future is:" << f.get() << std::endl;
    t.join();

    return 0;
}
int func_task(int a, int b) {
    std::this_thread::sleep_for(std::chrono::seconds(3));

    return a<b?a:b;
}

int TestPackaged()
{
    std::packaged_task<int(int, int)> task(func_task);
    std::future<int> f = task.get_future();

    std::thread t(std::move(task), 20, 10);
    std::cout <<"get task future is:"<< f.get() << std::endl;          

    t.join();

    return 0;
}

int main()
{
    TestAsync();
    TestPromise();
    TestPackaged();
    return 0;
}

是不是在上面提到的几种场景，其实你明白了整体的来龙去脉，学习才会更有效率。当然，这些需要不断的积累，是一个量变到质变的过程，特别对于c++这门语言，还是需要下功夫才能学好的。
下面再看官网提供的例子：

#include <iostream>
#include <vector>
#include <algorithm>
#include <numeric>
#include <future>
#include <string>
#include <mutex>

std::mutex m;
struct X {
    void foo(int i, const std::string& str) {
        std::lock_guard<std::mutex> lk(m);
        std::cout << str << ' ' << i << '\n';
    }
    void bar(const std::string& str) {
        std::lock_guard<std::mutex> lk(m);
        std::cout << str << '\n';
    }
    int operator()(int i) {
        std::lock_guard<std::mutex> lk(m);
        std::cout << i << '\n';
        return i + 10;
    }
};

template <typename RandomIt>
int parallel_sum(RandomIt beg, RandomIt end)
{
    auto len = end - beg;
    if (len < 1000)
        return std::accumulate(beg, end, 0);

    RandomIt mid = beg + len/2;
    auto handle = std::async(std::launch::async,
                             parallel_sum<RandomIt>, mid, end);
    int sum = parallel_sum(beg, mid);
    return sum + handle.get();
}

int main()
{
    std::vector<int> v(10000, 1);
    std::cout << "The sum is " << parallel_sum(v.begin(), v.end()) << '\n';

    X x;
    // Calls (&x)->foo(42, "Hello") with default policy:
    // may print "Hello 42" concurrently or defer execution
    auto a1 = std::async(&X::foo, &x, 42, "Hello");
    // Calls x.bar("world!") with deferred policy
    // prints "world!" when a2.get() or a2.wait() is called
    auto a2 = std::async(std::launch::deferred, &X::bar, x, "world!");
    // Calls X()(43); with async policy
    // prints "43" concurrently
    auto a3 = std::async(std::launch::async, X(), 43);
    a2.wait();                     // prints "world!"
    std::cout << a3.get() << '\n'; // prints "53"
} // if a1 is not done at this point, destructor of a1 prints "Hello 42" here

这里面通过二分方式进行了和的计算。

五、注意点

std::async和c++的形象有点匹配，里面有几个注意的地方，不小心就掉了进去了。
第一个是如果出现线程运行非常慢可能是如下原因：
The implementation may extend the behavior of the first overload of std::async by enabling additional (implementation-defined) bits in the default launch policy.
Examples of implementation-defined launch policies are the sync policy (execute immediately, within the async call) and the task policy (similar to async, but thread-locals are not cleared)
If the std::future obtained from std::async is not moved from or bound to a reference, the destructor of the std::future will block at the end of the full expression until the asynchronous operation completes, essentially making code such as the following synchronous:

std::async(std::launch::async, []{ f(); }); // temporary's dtor waits for f()
std::async(std::launch::async, []{ g(); }); // does not start until f() completes
(note that the destructors of std::futures obtained by means other than a call to std::async never block)
上述方式应用时，应将返回值move 或者bound到一个对象引用，否则std::future析构函数将阻塞async表达式直到异步操作的完成。从而导致任务的串行化操作。（VS上表现是正常的，但是在GCC上表现就如上面所述）换句话说，想通过不处理std::async返回值来忽略运行的异步处理，反而有可能导致串行化的结果，即：
1、由std::async时创建
2、还没有处理完成
3、最后一个引用由当前future持有.
在正常的情况下，会通过get和wait等来处理结果对象，就不会遇到上述的问题了。
第二个是如果出现异常怎么处理：
Throws std::system_error with error condition std::errc::resource_unavailable_try_again if the launch policy equals std::launch::async and the implementation is unable to start a new thread (if the policy is async|deferred or has additional bits set, it will fall back to deferred or the implementation-defined policies in this case), or std::bad_alloc if memory for the internal data structures could not be allocated.
但是由于c++的异常信息没有栈信息，所以在大的工程上，估计头就要大了。
另外还有一个需要注意的是平台的不同，导致默认参数的不同，比如在Win平台默认是std::launch::async而在Gcc上是std::launch::deferred，这个得注意这些小的细节。

六、总结

异步编程在所谓的“高级”编程中是一块相当重要的部分，很多编程者，其中不乏老鸟，仍然在这方面应用中掉到各种坑中。所以明白异步的原理和实现非常重要，异步的数据操作更因为上层不断的封装，导致整个对整个数据流的把握产生混乱，这就需要不断的在实践中印证相关的知识，由简入繁。明白怎么回事儿，不代表能解决所有问题，还是要保持不断学习的心态。

码农公寓