【C++ 标准项目】发布订阅消息队列 (篇五)：C++11 异步编程，手写线程池，async/promise/packaged_task/future 实战

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

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

1.std::future

a.介绍

链接如下：

cplusplus.com/reference/future/future/

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

b.应用场景

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

c.用法示例

• 使用 std::async 关联异步任务

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 内部通过系统等条件自动选择策略

来看代码：

运行结果：

• 使用 std::packaged_task 和 std::future 配合

std::packaged_task 就是将任务和 std::future 绑定在一起的模板，是一种对任务的封装。我们可以通过 std::packaged_task 对象获取任务相关联的 std::future 对象，通过调用 get_future()方法获得。std::packaged_task 的模板参数是函数签名。可以把 std::future 和 std::async 看成是分开的， 而 std::packaged_task 则是一个整体。

来看代码：

运行结果：

• 使用 std::promise 和 std::future 配合

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

来看代码：

2.c++11 线程池实现

来看代码：

运行结果：

代码解析如下：

a.整体结构

简答说就是：push 负责"交任务"，工作线程负责"取任务执行"，future 负责"传结果"。

b.核心流程

push() 的四步

template<typename F, typename …Args>
auto push(F &&func, Args&& …args) -> std::future<decltype(func(args…))>
{
// 1. 把函数 + 参数绑成一个整体，并推断返回类型
using return_type = decltype(func(args…));
auto tmp_func = std::bind(std::forward<F>(func), std::forward<Args>(args)…);

// 2. 用 packaged_task 包装它，并用 shared_ptr 持有
auto task = std::make_shared<std::packaged_task<return_type()>>(tmp_func);
std::future<return_type> fu = task->get_future(); // 取出配对凭证

// 3. 构造一个只捕获 shared_ptr 的 lambda，塞进队列
{
std::unique_lock<std::mutex> lock(_mutex);
_taskpool.push_back([task](){ (*task)(); });
_cv.notify_one(); // 叫醒一个工作线程
}
return fu; // 4. 把凭证交给调用方
}

entry() 的三步

void entry() {
while(!_stop)
{
std::vector<functor> tmp_taskpool;
{
std::unique_lock<std::mutex> lock(_mutex);
// 1. 等：任务池非空 或 要停了
_cv.wait(lock, [this](){ return _stop || !_taskpool.empty(); });
// 2. 把整批任务 swap 到本地
tmp_taskpool.swap(_taskpool);
}
// 3. 【锁已经释放】逐个执行
for (auto &task : tmp_taskpool) task();
}
}

c.四个关键设计

1. 为什么必须用 shared_ptr 包 packaged_task

这是整段代码最关键的一处，很多人写线程池就卡在这里：

std::vector<functor> _taskpool; // functor = std::function<void()>

std::function 要求存进去的可调用对象是可拷贝的。但 std::packaged_task 是 move-only（只能移动，不能拷贝）：

std::packaged_task<int()> t1(…);
std::packaged_task<int()> t2 = t1; // 错误 拷贝构造被删除

_taskpool.push_back(task); // 错误 packaged_task 无法放进 std::function

解法：用 shared_ptr 包一层。shared_ptr 是可拷贝的，lambda 捕获它也只是增加引用计数：

auto task = std::make_shared<std::packaged_task<return_type()>>(tmp_func);
_taskpool.push_back([task](){ (*task)(); }); // 正确 shared_ptr 可拷贝

同时它还顺手解决了生命周期问题：

• task 被 lambda 捕获 –> 引用计数 +1
• 任务执行完，lambda 被销毁 –> 引用计数 -1 –> 自动释放
• 不用手动 delete，也不用担心"任务还在队列里对象就没了"

小结：shared_ptr 在这里干了两件事——让 move-only 的对象能被 std::function 持有，以及自动管理生命周期。

2.为什么要把任务 swap 到本地再执行

{
std::unique_lock<std::mutex> lock(_mutex);
tmp_taskpool.swap(_taskpool); // <- 只做这一件事，然后立刻解锁
}
for (auto &task : tmp_taskpool) task(); // <- 在锁【外面】执行

如果直接在锁里执行任务会怎样？

// 反例
{
std::unique_lock<std::mutex> lock(_mutex);
while(!_taskpool.empty()) {
_taskpool.front()(); // 执行任务可能耗时很久
_taskpool.pop_front();
}
}

后果：某个任务跑了 5 秒，这 5 秒内其他线程一个都 push 不进来（锁被占着）—— 线程池变成了"串行池"。

swap 的妙处：加锁的时间只有"交换两个 vector 指针"那么短（O(1)），任务的实际执行完全不持锁。

3. 为什么 wait 必须带谓词

_cv.wait(lock, [this](){ return _stop || !_taskpool.empty(); });

等价于：

while (!(_stop || !_taskpool.empty())) { // <— 必须用 while，不是 if
_cv.wait(lock);
}

两个原因：

不带谓词的写法是错的：

_cv.wait(lock); // 错误的 醒来后不检查条件，可能任务池是空的

4. 为什么 _stop 要用 std::atomic<bool>

_stop 被多个线程同时读写（工作线程读、stop() 写），必须保证可见性和原子性：

std::atomic<bool> _stop; // 正确
bool _stop; // 错误 数据竞争，工作线程可能永远看不到变化

d.五个有待改进得问题

 问题 1：main 里 push 完立刻 get，等于没有并发

for(int i = 0; i < 10; i++)
{
std::future<int> fu = pool.push(Add, 11, i);
std::cout << fu.get() << std::endl; // <– 立刻阻塞等待
}

每个任务都是"提交 –> 等它做完 –> 再提交下一个"，线程池退化成单线程顺序执行。 想真正看到并发，应该先把任务都提交完，再统一取结果：

std::vector<std::future<int>> fus;
for (int i = 0; i < 10; i++)
fus.push_back(pool.push(Add, 11, i)); // 先全部提交

for (auto& f : fus)
std::cout << f.get() << std::endl; // 再统一取

这样 10 个任务能同时被多个工作线程并行处理。

 问题 2：std::bind 会拷贝参数，丢失移动语义

auto tmp_func = std::bind(std::forward<F>(func), std::forward<Args>(args)…);

std::bind 存参数时会 decay（退化）——std::forward 白写了：

• 传进来的右值会被拷贝而不是移动；
• move-only 的参数（比如 unique_ptr、packaged_task）根本传不进去，直接编译报错。

C++14 起有更好的写法（lambda + 初始化捕获）：

auto task = std::make_shared<std::packaged_task<return_type()>>(
[func = std::forward<F>(func), args = std::make_tuple(std::forward<Args>(args)…)]() mutable {
return std::apply(func, std::move(args));
});

（std::apply 是 C++17。你编译用 -std=c++11，所以 std::bind 是这个版本的标准做法，不算错，但要知道它的局限

 问题 3：stop() 可能丢弃剩余任务

while(!_stop) // <– 工作线程在这里检查
{
…
_cv.wait(lock, [this](){ return _stop || !_taskpool.empty(); });
tmp_taskpool.swap(_taskpool);
…
}

如果 _stop 在工作线程执行完一批任务、准备回到 while 判断时被置位，而队列里还有新任务 —— 它会直接退出循环，那些任务永远不会被执行。

这个"丢任务"是静默的，不会报错。想保证"停之前把活干完"，需要在退出前再清一次队列：

void entry() {
for (;;)
{
std::vector<functor> tmp;
{
std::unique_lock<std::mutex> lock(_mutex);
_cv.wait(lock, [this]{ return _stop || !_taskpool.empty(); });
tmp.swap(_taskpool);
// 只有在"要停 且 队列已空"时才退出
if (tmp.empty() && _stop) return;
}
for (auto &t : tmp) t();
}
}

（不过这里得 main 里是"push 一个 get 一个"，退出时队列本来就是空的，所以实际不会触发。）

问题 4：stop() 的 check-then-set 不是原子的

if(_stop == true) return; // <– 检查
_stop = true; // <– 设置

两个线程同时调 stop()，可能都通过检查。更好的写法是：

if (_stop.exchange(true)) return; // 一步完成"检查并设置"

问题 5：tmp_taskpool 每轮循环都重建

while(!_stop)
{
std::vector<functor> tmp_taskpool; // <— 每次都重新分配内存

把它移到循环外面，就能复用已分配的内存（swap 之后 vector 的容量会保留在另一方）：

std::vector<functor> tmp_taskpool; // 放外面
while(!_stop)
{
{ … tmp_taskpool.swap(_taskpool); }
for (auto &task : tmp_taskpool) task();
tmp_taskpool.clear();
}

e.运行结果

因为 main 是"push 一个、get 一个"，所以顺序输出——这反而证明了 future.get() 的同步作用：每次 get 都会阻塞到对应任务执行完。想验证多线程真的在跑，可以在 Add 里打印线程 ID：

int Add(int a, int b) {
std::cout << "线程 " << std::this_thread::get_id() << " 处理任务\\n";
return a + b;
}

然后用上面"问题 1"里那个"先全部提交、再统一取结果"的写法——你会看到任务被不同线程接手。

f.小结