异步操作实现线程池

future

在C++11标准库中，提供了一个future的模板类，它表示的是一个异步操作的结果，当在多线程编程中使用异步任务的时候，使用这个类可以帮助在需要的时候获取到对应的数据处理结果，而future类本质上的一个重要特性是可以阻塞当前线程，直到异步操作完成，而确保在获取结果的时候不会出现结果未完成的情况出现

应⽤场景

• 异步任务：当我们需要在后台执⾏⼀些耗时操作时，如⽹络请求或计算密集型任务等，std::future
  可以⽤来表⽰这些异步任务的结果。通过将任务与主线程分离，我们可以实现任务的并⾏处理，从
  ⽽提⾼程序的执⾏效率

• 并发控制：在多线程编程中，我们可能需要等待某些任务完成后才能继续执⾏其他操作。通过使⽤
  std::future，我们可以实现线程之间的同步，确保任务完成后再获取结果并继续执⾏后续操作

• 结果获取：std::future提供了⼀种安全的⽅式来获取异步任务的结果。我们可以使⽤
  std::future::get()函数来获取任务的结果，此函数会阻塞当前线程，直到异步操作完成。这样，在
  调⽤get()函数时，我们可以确保已经获取到了所需的结果

下面来看官方文档对于这个类的讲述：

大体意思是，future会作为一个返回值来进行接收，可以通过下面的三种方式来进行调用，这里展示第一种：async

async

在这个函数当中，就是一个经典的调用异步操作来执行的操作，对于函数参数来说，Fn表示的是这是一个要执行的函数，后面的args表示的是这个函数的参数，而对于这个函数来说，它存在一种函数的重载，这个函数的重载可以在最前面加上一个调用的策略，可以使得是立刻进行执行和获取函数的返回值，或者是在调用get函数再进行函数返回值的获取，下面使用一个实例代码来进行演示

std::async是⼀种将任务与std::future关联的简单⽅法。它创建并运⾏⼀个异步任务，并返回⼀个与该任务结果关联的std::future对象。默认情况下，std::async是否启动⼀个新线程，或者在等待future时，任务是否同步运⾏都取决于你给的参数。这个参数为std::launch类型：

• std::launch::deferred表明该函数会被延迟调⽤，直到在future上调⽤get()或者wait()才会开始
  执⾏任务
• std::launch::async 表明函数会在⾃⼰创建的线程上运⾏
• std::launch::deferred | std::launch::async 内部通过系统等条件⾃动选择策略

#include <chrono>
#include <iostream>
#include <future>
#include <thread>
using namespace std;// 模拟一个加法的环境
int add(int num1, int num2)
{cout << "加法" << endl;return num1 + num2;
}void deferred_solve()
{cout << "deferred" << endl;cout << "------1------" << endl;future<int> fut = async(launch::deferred, add, 10, 20);cout << "------2------" << endl;this_thread::sleep_for(chrono::seconds(1));cout << "------3------" << endl;int res = fut.get();cout << "------4------" << endl;cout << "运行结果" << res << endl;
}void async_solve()
{cout << "async" << endl;cout << "------1------" << endl;future<int> fut = async(launch::async, add, 10, 20);cout << "------2------" << endl;this_thread::sleep_for(chrono::seconds(1));cout << "------3------" << endl;int res = fut.get();cout << "------4------" << endl;cout << "运行结果" << res << endl;
}int main()
{cout << "deferred: " << endl;deferred_solve();cout << endl;cout << "async: " << endl;async_solve();cout << endl;return 0;
}

运行结果如下所示：

deferred: 
deferred
------1------
------2------
------3------
加法
------4------
运行结果30async: 
async
------1------
------2------
加法
------3------
------4------
运行结果30

从上可以看出一些端倪，对于deferred这种策略来说，它的策略是在进行get方法的时候再进行资源的计算，而对于async这样的策略来说，更多的是在进行调用之后就会进行计算，在这种调用之后，会立刻再开一个工作线程把内容计算完毕后传递回主函数，这是两个基本的调用逻辑。

promise

std::promise提供了⼀种设置值的⽅式，它可以在设置之后通过相关联的std::future对象进⾏读取。换
种说法就是之前说过std::future可以读取⼀个异步函数的返回值了，但是要等待就绪，⽽std::promise就提供⼀种⽅式⼿动让std::future就绪

#include <iostream>
#include <thread>
#include <future>using namespace std;void add(int num1, int num2, promise<int>& pro)
{pro.set_value(num1 + num2);return;
}int main()
{promise<int> pro;future<int> fut = pro.get_future();thread th(add, 10, 20, ref(pro));int res = fut.get();cout << "执行结果: " << res << endl;th.join();return 0;
}

这个场景本质上就是利用了一个promise对象来和future对象建立了关系，如果在获取future对象的时候并没有发生值改变，就会阻塞等待，保证了异步的基本进行

package task

对于这种调用的方式，可以把它生成的对象当成是一个可调用对象，下面演示其基本用法

#include <iostream>
#include <thread>
#include <future>
#include <memory>
using namespace std;int add(int num1, int num2)
{return num1 + num2;
}int main()
{auto ptask = make_shared<packaged_task<int(int, int)>>(add);future<int> fut = ptask->get_future();thread th([ptask](){(*ptask)(10, 20);});int sum = fut.get();cout << sum << endl;th.join();return 0;
}

C++11线程池实现

基于线程池执⾏任务的时候，⼊⼝函数内部执⾏逻辑是固定的，因此选择std::packaged_task加上std::future的组合来实现。
线程池的⼯作思想：

• ⽤⼾传⼊要执⾏的函数，以及需要处理的数据（函数的参数）,由线程池中的⼯作线程来执⾏函数完成任务

实现：

1. 管理的成员

• 任务池：⽤vector维护的⼀个函数任务池⼦
• 互斥锁&条件变量：实现同步互斥
• ⼀定数量的⼯作线程：⽤于不断从任务池取出任务执⾏任务
• 结束运⾏标志：以便于控制线程池的结束。

2. 管理的操作：

• ⼊队任务：⼊队⼀个函数和参数
• 停⽌运⾏：终⽌线程池

#include <features.h>
#include <iostream>
#include <functional>
#include <memory>
#include <thread>
#include <future>
#include <mutex>
#include <condition_variable>
#include <vector>
using namespace std;class threadpool
{using Functor = function<void(void)>;
public:threadpool(int count = 1) : _stop(false){for(int i = 0; i < count; i++)_threads.emplace_back(&threadpool::entry, this);}~threadpool(){ stop();}void stop(){if(_stop == true)return;_stop = true;// 唤醒线程_cv.notify_all();// 回收线程for(auto& thread: _threads)thread.join();}// 对于push函数，传入的是一个用户要执行的函数，还有函数的参数// push函数的内部，会把这些传入的函数和参数封装为一个packaged_task// 然后使用lambda表达式生成一个可调用对象，放到任务池中，让工作线程取出执行template<typename F, typename ...Args>auto push(const F&& func, Args&& ...args) -> future<decltype(func(args...))>{// 1. 把传入的函数封装为一个packaged任务// 把返回类型获取出来using return_type = decltype(func(args...));// 把函数对象和函数参数绑定到一起auto tmp_func = bind(forward<F>(func), forward<Args>(args)...);// 把整体的tmp_func绑定成一个任务auto task = make_shared<packaged_task<return_type()>>(tmp_func);future<return_type> fut = task->get_future();// 2. 构造一个lambda匿名函数，函数内执行任务对象{unique_lock<mutex> lock(_mutex);// 3. 把匿名函数对象放到任务池中_taskpool.push_back([task](){ (*task)(); });_cv.notify_one();}return fut;}
private:void entry(){while(!_stop){vector<Functor> tmp_taskpool;{unique_lock<mutex> lock(_mutex);_cv.wait(lock, [this](){ return _stop || !_taskpool.empty(); });tmp_taskpool.swap(_taskpool);}for(auto& task : tmp_taskpool)task();}}
private:atomic<bool> _stop;vector<Functor> _taskpool;mutex _mutex;condition_variable _cv;vector<thread> _threads;
};