【RabbitMQ 项目】前置技术：含同步操作的线程池——C++11＜future＞使用

一.C++11 中的

该头文件中的接口是为了访问异步任务的结果，由于任务是异步的，获取结果是无法得知任务是否执行完毕，如果贸然获取，可能会得到一个无效结果，所以其中必须要有同步操作

二.future 对象

future 对象代表了一个异步任务的未来结果，调用它的 get 方法获取结果时，如果结果还没有就绪，线程就会阻塞，直到任务完成

获取 future 对象的三种方法：

• async
• promise::get_future
• packaged_task::get_future

三.async 全局函数

功能：设置一个任务，设置完成后立马返回一个 future 对象，代表这个任务未来的结果，之后函数的返回值被设置到结果中
通过设置标志位可选择任务的执行策略：

1. launch::async：立马创建一个线程，让新线程去异步地执行任务，
2. launch::deferred：推迟执行，直到用户调用 get 方法获取结果时才开始执行(有点鸡肋，最终任务我是线程自己执行的，那还不如自己调用函数执行)
3. launch::async | launch::deferred：根据当时的环境自动选择异步还是延迟

#include <iostream>
#include <future>
#include <chrono>
#include <thread>

int Add(int x, int y)
{
    std::this_thread::sleep_for(std::chrono::seconds(5));
    return x + y;
}
int main()
{
    std::future<int> fu =  std::async(std::launch::async, Add, 1, 2);
    int ret = fu.get();

    std::cout << "ret = " << ret << std::endl;
    return 0;
}

四.promise::get_future

使用 promise 对象的 get_future 方法获得一个 future 对象，它们共享同一个结果，把 promise 对象传递给异步工作的线程，主线程自己保留 future 对象，在异步任务中使用 promise 的 set_value 方法设置结果，主线程中的 future 对象也可以看到这个结果

#include <iostream>
#include <future>
#include <chrono>
#include <thread>

int Add(int x, int y)
{
    std::this_thread::sleep_for(std::chrono::seconds(5));
    return x + y;
}
int main()
{
    std::future<int> fu =  std::async(std::launch::async, Add, 1, 2);

    int ret = fu.get();

    std::cout << "ret = " << ret << std::endl;
    return 0;
}

五.packaged_task::get_future

使用 packaged_task 对象封装任务，用 get_future 方法获取一个 future 对象，代表所封装任务的未来结果。将 packaged_task 对象交给异步工作线程去执行，任务的返回结果会设置到 future 对象中

注意：packaged_task 对象和可调用对象类似，可以圆括号调用，但和可调用对象有所不同：

1. 不能拷贝，所以不能直接作为线程的入口函数
2. 传引用不可取，因为有生命周期问题
3. 所以想到在堆上开辟空间，传递指针，但是多线程之间传递指向堆空间的指针管理起来很麻烦，所以最好的办法是传递智能指针
4. 最终方案：线程执行一个 lambda 表达式，lambda 表达式值捕捉智能指针，在表达式中解引用调用

#include <iostream>
#include <future>
#include <chrono>
#include <thread>
#include <memory>
using namespace std;
int Add(int x, int y)
{
    std::this_thread::sleep_for(std::chrono::seconds(5));
    return x + y;
}
int main()
{

    auto ptaskPtr = std::make_shared<packaged_task<int(int, int)>>(Add);
    future<int> fu = ptaskPtr->get_future();
    thread t1([ptaskPtr]{
        (*ptaskPtr)(1, 2);
    });
    int ret = fu.get();
    std::cout << ret << std::endl;
    t1.join();
    
    return 0;
}

六.使用 packaged_task 和 future 实现可获取结果的线程池

ThreadPool.hpp

#pragma once
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <atomic>
#include <vector>
#include <queue>
#include <functional>
#include <iostream>
#include <future>
static const int defaultThreadNum = 5;
class ThreadPool
{
public:
    using task_t = std::function<void()>;

private:
    std::vector<std::thread> _workers;
    std::queue<task_t> _tasks;
    std::mutex _mtx;
    std::condition_variable _cond;
    std::atomic<bool> _isRunning;
    int _threadNum;

public:
    ThreadPool(int threadNum = defaultThreadNum)
        : _workers(),
          _tasks(),
          _mtx(),
          _cond(),
          _isRunning(false),
          _threadNum(threadNum)
    {
    }

    ~ThreadPool()
    {
        for (auto& t : _workers)
        {
            t.join();
        }
    }

    void start()
    {
        for (int i = 0; i < _threadNum; i++)
        {
            _workers.emplace_back(&ThreadPool::work, this);
        }
        _isRunning = true;
    }

    template <typename Fn, typename ...Args>
    auto push(Fn&& func, Args&& ... args) -> std::future<decltype(func(args...))>
    {
        using return_t = decltype(func(args...));
        if (!_isRunning)
        {
            std::cout << "Please start threadPool first" << std::endl;
            return std::future<return_t>();
        }
       
        //1.把这个任务的参数绑定，变成一个返回值位return_t,参数为void的可调用对象
        auto tmpFunc = std::bind(std::forward<Fn>(func), std::forward<Args>(args)...);

        //2.把这个bind后的可调用对象用packaged_task封装，用智能指针管理
        auto ptaskPtr = std::make_shared<std::packaged_task<return_t()>>(tmpFunc);
        
        //3.把智能指针封装到lambda表达式中解引用执行，并把lambda表达式丢进任务队列
        task_t task = [ptaskPtr](){(*ptaskPtr)();};

        std::unique_lock<std::mutex> lck(_mtx);
        _tasks.push(task);
        _cond.notify_one();
        return ptaskPtr->get_future();
    }

private:
    void work()
    {
        while (true)
        {
            task_t task;
            {
                std::unique_lock<std::mutex> lck(_mtx);
                // 判断条件是否成立
                while (_tasks.empty())
                {
                    _cond.wait(lck);
                }
                task = _tasks.front();
                _tasks.pop();
            }
            task();
        }
    }
};