Boost Asio Experimental Channel单线程性能问题排查与优化咨询

你的推测完全正确：单线程模式下，async_send每次完成都会立即调度消费者协程，导致协程频繁切换，这就是性能瓶颈的核心原因。在单线程的io_context中，所有异步操作的完成回调都在同一个线程执行——生产者每发送一条数据就触发一次调度，消费者刚挂起就被唤醒，来回切换的开销吃掉了大部分CPU资源，所以吞吐量卡在了3M条/秒左右。缓冲区设为0时性能没变化，是因为此时每次发送都必须等待消费者接收，协程切换的次数并没有减少。

验证猜想的方法 #

• 统计协程切换次数：在生产者每次co_await async_send前后、消费者每次co_await async_receive前后分别加计数器，运行后查看总切换次数。如果切换次数接近600万（300万发送+300万接收），就实锤了频繁切换的问题。
• 性能工具分析：用Linux的perf或Windows的VTune分析程序热点，会发现大部分CPU时间消耗在Asio的协程调度逻辑上（比如asio::detail::scheduler相关函数）。

性能优化方案 #

1. 批量处理数据（推荐） #

最有效的优化方式是减少协程切换次数——不要每次发送单条数据，而是攒一批数据再发送，消费者一次接收批量数据后统一处理。这样协程切换次数会从数百万级降到数千级，性能会大幅提升。

修改后的示例代码：

#include <boost/asio.hpp>
#include <boost/asio/experimental/awaitable_operators.hpp>
#include <boost/asio/experimental/channel.hpp>
#include <vector>

namespace asio = boost::asio;
using namespace asio::experimental::awaitable_operators;
// 改为传递批量数据的channel
using channel_t = asio::experimental::channel< void(boost::system::error_code, std::vector<uint64_t>) >;

constexpr size_t BATCH_SIZE = 1000; // 每批打包1000条数据

asio::awaitable< void >
producer(channel_t &ch)
{
    std::vector<uint64_t> batch;
    batch.reserve(BATCH_SIZE);
    for (uint64_t i = 0; i < 3'000'000; i++)
    {
        batch.push_back(i);
        if (batch.size() == BATCH_SIZE)
        {
            co_await ch.async_send(boost::system::error_code {}, std::move(batch), asio::use_awaitable);
            batch.reserve(BATCH_SIZE);
        }
    }
    // 发送剩余的零散数据
    if (!batch.empty())
    {
        co_await ch.async_send(boost::system::error_code {}, std::move(batch), asio::use_awaitable);
    }
    ch.close();
}

asio::awaitable< void >
consumer(channel_t &ch)
{
    for (;;)
    {
        auto batch = co_await ch.async_receive(asio::use_awaitable);
        // 此处添加批量数据处理逻辑
        (void)batch; // 占位符，替换为实际业务代码
    }
}

asio::awaitable< void >
experiment()
{
    channel_t ch { co_await asio::this_coro::executor, 10 }; // 缓冲区可适当缩小
    co_await (consumer(ch) && producer(ch));
}

int main()
{
    asio::io_context ctx {};
    asio::co_spawn(ctx, experiment(), asio::detached);
    ctx.run();
}

2. 多线程调度 #

如果业务场景允许，可以将io_context配置为多线程模式，让生产者和消费者在不同线程执行，实现发送和接收的并行处理，避免单线程内的调度开销。Asio的channel本身是线程安全的，跨线程使用无需额外同步。

修改后的main函数示例：

int main()
{
    asio::io_context ctx {2}; // 初始化2个线程的上下文
    asio::co_spawn(ctx, experiment(), asio::detached);
    
    // 启动两个线程运行io_context
    std::thread t1([&ctx]{ ctx.run(); });
    std::thread t2([&ctx]{ ctx.run(); });
    
    t1.join();
    t2.join();
}

3. 调整channel缓冲区策略 #

如果业务不适合批量处理，可以适当调大channel的缓冲区，让生产者尽可能填满缓冲区后再唤醒消费者，减少唤醒频率。不过这种优化的效果远不如批量处理明显。

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