使用Asio并行执行并等待运行时数量的C++20协程

原代码的核心问题 #

• 冗余的线程池调度：asio::co_spawn本身会将协程提交到指定执行器（thread_pool），无需额外通过asio::post包装，这既增加了调度开销，还引入了不必要的线程安全锁逻辑。
• parallel_group使用错误：make_parallel_group需要配合协程环境使用，但原代码在非协程的main函数中直接调用async_wait，无法正确处理异步等待逻辑；同时存储co_spawn返回的awaitable的方式不符合parallel_group的使用要求。
• 执行器类型错误：任务协程中强制将this_coro::executor转换为io_context&是错误的，thread_pool的执行器并非io_context类型，协程直接使用拿到的执行器即可。

修正后的实现代码 #

#include <asio/co_spawn.hpp>
#include <asio/experimental/awaitable_operators.hpp>
#include <asio/experimental/parallel_group.hpp>
#include <asio/steady_timer.hpp>
#include <asio/thread_pool.hpp>
#include <asio/use_awaitable.hpp>
#include <iostream>
#include <vector>

// 单个协程任务，接收ID用于区分，返回示例结果
asio::awaitable<int> task(int id) {
    auto ex = co_await asio::this_coro::executor;
    // 模拟协程异步工作：定时器等待
    asio::steady_timer timer(ex, std::chrono::milliseconds(100 * id));
    co_await timer.async_wait(asio::use_awaitable);
    
    std::cout << "任务 " << id << " 执行完成\n";
    co_return id * 2;
}

int main() {
    const int task_count = 5; // 可运行时配置的任务数量
    asio::thread_pool pool(4); // 4线程的线程池

    // 方式1：用when_all等待所有协程完成（简洁易用）
    asio::co_spawn(pool, [&]() -> asio::awaitable<void> {
        using namespace asio::experimental::awaitable_operators;

        std::vector<asio::awaitable<int>> tasks;
        for (int i = 0; i < task_count; ++i) {
            tasks.push_back(task(i));
        }

        // 等待所有协程完成，获取结果集合
        auto results = co_await when_all(tasks.begin(), tasks.end());
        
        std::cout << "\n所有任务完成，结果：";
        for (auto res : results) {
            std::cout << res << " ";
        }
        std::cout << "\n";
    }, asio::detached);

    // 方式2：用parallel_group（支持灵活等待策略，可获取完成顺序）
    // asio::co_spawn(pool, [&]() -> asio::awaitable<void> {
    //     std::vector<asio::awaitable<int>> tasks;
    //     for (int i = 0; i < task_count; ++i) {
    //         tasks.push_back(task(i));
    //     }

    //     auto group = asio::experimental::make_parallel_group(std::move(tasks));

    //     // 等待所有任务完成，拿到完成顺序和结果
    //     auto [completion_order, results] = co_await group.async_wait(
    //         asio::experimental::wait_for_all(),
    //         asio::use_awaitable
    //     );

    //     std::cout << "\n完成顺序：";
    //     for (auto idx : completion_order) {
    //         std::cout << idx << " ";
    //     }
    //     std::cout << "\n结果：";
    //     for (auto& res : results) {
    //         std::cout << res.value() << " ";
    //     }
    //     std::cout << "\n";
    // }, asio::detached);

    pool.join();
    return 0;
}

关键实现说明 #

• 协程任务设计：将任务封装为独立的awaitable函数，直接使用协程获取的执行器进行异步操作，避免类型转换错误。
• 两种等待方案：
  • when_all：属于asio::experimental::awaitable_operators，语法简洁，适合只需要等待所有任务完成并获取结果的场景。
  • parallel_group：支持多种等待策略（如等待第一个完成、等待指定数量完成），还能获取任务完成的顺序，适合需要精细控制的场景。
• 主逻辑协程化：通过asio::co_spawn将主等待逻辑包装成协程提交到线程池，避免在非协程环境中处理异步等待的复杂逻辑。
• 无冗余调度：直接创建协程任务并等待，省去不必要的post操作，减少线程切换开销。

内容的提问来源于stack exchange，提问作者Davidbrcz