C++11 异步操作实现线程池

文章目录

• 1. std::future
• 2. C++11 线程池实现

1. std::future

介绍

std::future 是 C++11 标准库中的一个模板类，它表示一个异步操作的结果。当我们在多线程编程中使用异步任务时，std::future 可以帮助我们在需要的时候获取任务的执行结果。std::future 的一个重要特性是能够阻塞当前线程，直到异步操作完成，从而确保我们在获取结果时不会遇到未完成的操作。

应用场景

• 异步任务： 当我们需要在后台执行一些耗时操作时，如网络请求或计算密集型任务等，std::future 可以用来表示这些异步任务的结果。通过将任务与主线程分离，我们可以实现任务的并行处理，从而提高程序的执行效率
• 并发控制： 在多线程编程中，我们可能需要等待某些任务完成后才能继续执行其他操作。通过使用 std::future，我们可以实现线程之间的同步，确保任务完成后再获取结果并继续执行后续操作
• 结果获取：std::future 提供了一种安全的方式来获取异步任务的结果。我们可以使用 std::future::get()函数来获取任务的结果，此函数会阻塞当前线程，直到异步操作完成。这样，在调用 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 <iostream>
#include <thread>
#include <future>
#include <chrono>

int Add(int num1, int num2)
{
std::cout << "加法！！1111\\n";
std::this_thread::sleep_for(std::chrono::seconds(5));
std::cout << "加法！！2222\\n";
return num1 + num2;
}

int main()
{
//std::async(func, …) std::async(policy, func, …)
std::cout << "——–1———-\\n";
//std::launch::deferred 在执行get获取异步结果的时候，才会执行异步任务
//std::launch::async 内部会创建工作线程，异步的完成任务
std::future<int> result = std::async(std::launch::deferred, Add, 11, 22);
//std::future<int> result = std::async(std::launch::async, Add, 11, 22);
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "——–2———-\\n";
int sum = result.get();
std::cout << "——–3———-\\n";
std::cout << sum << std::endl;
return 0;
}

先来试一下std::launch::deferred：

ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ make
g++ -o async async.cc -std=c++11 -lpthread
ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ ./async
——–1———-
——–2———-
加法！！1111
加法！！2222
——–3———-
33

再来试一下std::launch::async：

ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ make
g++ -o async async.cc -std=c++11 -lpthread
ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ ./async
——–1———-
加法！！1111
——–2———-
加法！！2222
——–3———-
33

虽然使用std::launch::async参数，会启动一个新线程和主线程并行处理，但是当主线程（或调用线程）执行到 std::future::get()函数时，如果异步任务的结果尚未就绪，调用线程都会阻塞在get()处，直到结果可用。

两者的核心区别不在于 get()是否阻塞，而在于任务在何时、何地被执行，这决定了线程会在 get()处等待多久。

说白了，其实就是std::launch::deferred在调用 get() 时才会去执行任务，而std::launch::async则是启动一个新线程和主线程并行处理，主线程在 get() 获取异步任务的结果时，如果新线程还没处理完返回结果，那么主线程就会在 get() 处阻塞等待，直到新线程处理完毕后将结果返回

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

或者可以换一种方式理解：

std::promise和 std::future共同构成了一个经典的生产者-消费者模型：

• 生产者端 (std::promise)：在一个线程（如工作线程）中创建一个 std::promise对象。当计算出结果或发生异常时，通过该对象“设置”结果。
• 消费者端 (std::future)：在主线程（或任何需要结果的线程）中，通过 promise.get_future()获取与之关联的 std::future对象。您可以随时检查或等待（get()/wait()）这个 future来获取生产者设置的结果。

它们之间通过一个共享状态（Shared State） 连接，这是一个内部数据结构，负责安全地存储结果和同步线程

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

// 通过在线程中对promise对象设置数据，其他线程中通过future获取设置数据的方式实现获取异步任务执行结果的功能
void Add(int num1, int num2, std::promise<int> &prom)
{
std::this_thread::sleep_for(std::chrono::seconds(3));
prom.set_value(num1 + num2);
return;
}

int main()
{
std::promise<int> prom;

std::future<int> fu = prom.get_future();

std::thread thr(Add, 11, 22, std::ref(prom));
int res = fu.get();
std::cout << "sum: " << res << std::endl;
thr.join();
return 0;
}

运行结果：

ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ make
g++ -o promise promise.cc -std=c++11 -lpthread
ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ ./promise
sum: 33

注意：一个 promise只能有一个关联的 future。

std::packaged_task 就是将任务和 std::future 绑定在一起的模板，它内部封装了一个可调用对象（任务）和一个共享状态，该共享状态用于存储任务执行的结果。我们可以通过 std::packaged_task 对象获取任务相关联的 std::future 对象，通过调用 get_future()方法获得。std::packaged_task 的模板参数是函数签名（如 int(int, int)）。

可以把 std::future 和 std::async 看成是分开的， 而 std::packaged_task 则是一个整体。

#include <iostream>
#include <thread>
#include <future>
#include <memory>
// pakcaged_task的使用
// pakcaged_task 是一个模板类，实例化的对象可以对一个函数进行二次封装，
// pakcaged_task可以通过get_future获取一个future对象，来获取封装的这个函数的异步执行结果

int Add(int num1, int num2)
{
std::this_thread::sleep_for(std::chrono::seconds(3));
return num1 + num2;
}

int main()
{
// std::packaged_task<int(int,int)> task(Add);
// std::future<int> fu = task.get_future();

// task(11, 22); task可以当作一个可调用对象来调用执行任务
// 需要注意的是，task 虽然重载了()运算符，但是它又不能完全的当作一个函数来使用
// std::async(std::launch::async, task, 11, 22); //–错误用法
// 所以导致它作为线程的入口函数时，语法上看没有问题，但是实际编译的时候会报错
// std::thread thr(task, 11, 22); //–错误用法

// 而 packaged_task 禁止了拷贝构造，
// 且因为每个 packaged_task 所封装的函数签名都有可能不同，因此也无法当作参数一样传递
// 传引用不可取，毕竟任务在多线程下执行存在局部变量声明周期的问题，因此不能传引用

// 因此我们可以把task定义成为一个指针，传递到线程中，然后进行解引用执行
// 但是如果单纯指针指向一个对象，存在生命周期的问题，很有可能出现风险
// 思想就是在堆上new对象，用智能指针管理它的生命周期
auto ptask = std::make_shared<std::packaged_task<int(int, int)>>(Add);
std::future<int> fu = ptask->get_future();
std::thread thr([ptask](){
(*ptask)(11, 22);
});

int sum = fu.get();
std::cout << sum << std::endl;
thr.join();
return 0;
}

运行结果：

ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ make
g++ -o package_task package_task.cc -std=c++11 -lpthread
ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ ./package_task
33

2. C++11 线程池实现

基于线程池执行任务的时候，入口函数内部执行逻辑是固定的，因此选择

std::packaged_task 加上 std::future 的组合来实现。

线程池的工作思想：

• 用户传入要执行的函数，以及需要处理的数据（函数的参数），由线程池中的工作线程来执行函数完成任务

实现：

• 管理的成员
  • 任务池：用 vector 维护的一个函数任务池子
  • 互斥锁 & 条件变量： 实现同步互斥
  • 一定数量的工作线程：用于不断从任务池取出任务执行任务
  • 结束运行标志：以便于控制线程池的结束。
• 管理的操作：
  • 入队任务：入队一个函数和参数
  • 停止运行：终止线程池

#include <iostream>
#include <vector>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>

class threadpool
{
public:
using Functor = std::function<void(void)>;
threadpool(int threadnums = 1) :_stop(false)
{
for(int i = 0; i < threadnums; i++)
{
_threads.emplace_back(&threadpool::entry, this);
}
}
~threadpool()
{
stop();
}

void stop()
{
if(_stop) 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) -> std::future<decltype(func(args...))>
{
//1. 将传入的函数封装成一个packaged_task任务
using return_type = decltype(func(args...));
auto tmp_func = std::bind(std::forward<F>(func), std::forward<Args>(args)...);
auto ptask = std::make_shared<std::packaged_task<return_type()>>(tmp_func);
std::future<return_type> fu = ptask->get_future();
//2. 构造一个lambda匿名函数（捕获任务对象），函数内执行任务对象
{
std::unique_lock<std::mutex> lock(_mutex);
//3. 将构造出来的匿名函数对象，抛入到任务池中，即生产者生产数据
_taskpool.push_back([ptask](){
(*ptask)();
});
// 唤醒消费者
_cv.notify_one();
}
return fu;
}
private:
//线程入口函数—内部不断的从任务池中取出任务进行执行。
void entry()
{
while(!_stop)
{
std::vector<Functor> tmp_taskpool;
{
// 加锁
std::unique_lock<std::mutex> lock(_mutex);
// 等待任务池不为空，或者_stop被置位返回，
_cv.wait(lock, [this](){
return _stop || !_taskpool.empty();
});
// 取出任务进行执行
tmp_taskpool.swap(_taskpool);
}
for(auto& task : tmp_taskpool)
{
task();
}
}
}
private:
std::atomic<bool> _stop;
std::vector<Functor> _taskpool; // 任务池
std::mutex _mutex;
std::condition_variable _cv;
std::vector<std::thread> _threads;
};

int Add(int num1, int num2)
{
return num1 + num2;
}

int main()
{
threadpool pool;
for (int i = 0; i < 10; i++)
{
std::future<int> fu = pool.push(Add, 11, i);
std::cout << fu.get() << std::endl;
}
pool.stop();
return 0;
}

线程池的核心组件包括任务队列、工作线程组以及同步控制机制。

• 任务容器与同步机制：使用 std::vector<Functor>作为任务池 _taskpool来存储待执行的任务（这些任务是包装成 std::function<void(void)>的可调用对象）。线程同步通过 std::mutex（互斥锁 _mutex）和 std::condition_variable（条件变量 _cv）来实现，这是典型的生产者-消费者模型。
• 工作线程组：在构造函数中，根据指定的线程数量创建一组工作线程 _threads。每个工作线程在创建后立即执行 entry成员函数，在这个函数中循环等待并执行任务。
• 生命周期标志：std::atomic<bool> _stop是一个原子布尔标志，用于优雅地通知所有工作线程何时停止工作并退出。

工作流程详解

线程池的工作流程可以清晰地分为任务提交和工作线程执行两个部分。

a. 任务提交 (push方法)

push方法是一个模板函数，它负责接收用户想要执行的任何任务（函数、可调用对象等）及其参数，并将其打包后放入任务池。体现了生产者角色。

当调用 pool.push(Add, 11, i)时：

• 模板实例化：编译器会推导模板参数 F(函数类型，如 int(int, int)) 和 Args(参数类型，如 int, int&)。
• 返回类型推断：decltype(func(args…))在编译时确定任务的返回类型（本例中为 int），从而确定整个 push方法返回的 std::future的具体类型。

此步骤的目的是将五花八门的任务“标准化”。

• 绑定参数：std::bind(std::forward<F>(func), std::forward<Args>(args)…)将用户函数和其参数绑定在一起，生成一个可调用对象 tmp_func。这里使用 std::forward进行完美转发，确保参数的原值类别（左值/右值）得以保持，避免不必要的拷贝，提升效率。
• 创建异步任务：使用 std::make_shared创建一个封装了 tmp_func的 std::packaged_task<return_type()>对象 (ptask)。std::packaged_task的核心作用在于它将函数调用与其结果通道（std::future）绑定在一起。
• 获取结果通道：通过 ptask->get_future()立即获取与这个任务关联的 std::future对象 (fu)。后续可以通过这个 future异步获取任务的执行结果。

这是将任务安全地交到工作线程手中的关键步骤。

• 加锁：使用 std::unique_lock<std::mutex> lock(_mutex)锁定互斥量，确保在修改任务队列 _taskpool时，不会被其他线程（生产者或消费者）干扰，保证线程安全。
• 包装任务：将 ptask包装成一个无参的 Lambda 表达式 [ptask](){ (*ptask)(); }。这样做是因为 std::packaged_task不可拷贝，但可以被智能指针管理并通过引用捕获。更重要的是，线程池的任务队列 _taskpool定义为 std::vector<std::function<void(void)>>，它要求所有任务都是统一签名（void(void)）。这一步是实现类型擦除的关键，无论原始任务多复杂，最终都变成了一个简单的 void()调用。
• 任务入队：将包装好的 Lambda 任务放入任务队列 _taskpool中。
• 通知线程：调用 _cv.notify_one()，唤醒一个正在条件变量_cv上等待的空闲工作线程，告诉它有新任务可以处理了。

• push方法立即将之前获取到的 std::future对象返回给调用者。
• 调用者（例如上面的 main函数）可以在任何需要的时候，通过 fu.get()同步等待任务执行完成并获取结果。如果任务尚未完成，get()会阻塞当前线程直到结果就绪。

b. 工作线程执行 (entry方法)

每个工作线程在 entry函数中运行一个循环，充当消费者的角色，其执行流程如下：

运行结果：

ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ make
g++ -o threadpool threadpool.cpp -std=c++11 -lpthread
ltx@My-Xshell-8-Pro-Max-Ultra:~/rabbit-mq/demo/future$ ./threadpool
11
12
13
14
15
16
17
18
19
20