Boost.Asio 发送队列为何发生数据竞争，如何用 strand 修复？

2026-11-06数据小陈

结论

竞争条件来自 _send_queue 缺少同步保护。同步连接的测试会在启动 io_context::run() 线程前完成所有初始入队。异步连接的回调则可能在 io_context 运行期间入队，而发送完成处理器也可能同时在另一个工作线程上运行。因此，两条路径可能同时访问队列。

代码在 main 函数中创建了一个局部 strand，但没有传给 sender，也没有用于发送队列或 async_write。这个 strand 不会自动保护 socket 或队列。

为什么两种测试表现不同

async_initiate 会在发起操作时调用初始化函数。即使操作使用异步 API，初始化函数也不会因此自动安排到 socket 的 executor 或 strand 上。所以，当前 async_send_message_impl() 中的 _send_queue.emplace() 会直接在调用 async_send_message() 的线程上执行。

同步连接会在启动工作线程前调用 run_connection()。初始发送由同一线程依次入队；等到 ctx.run() 开始后，异步写入的完成处理器才会运行。初始入队期间没有并发访问。

异步连接的 async_connect 完成处理器会在 ctx.run() 的工作线程上运行，并在其中发起多条发送。与此同时，先前的写操作可能已经完成，handle_send_complete() 可能正在另一个工作线程上处理队列。这样一来，它就可能与后续的 _send_queue.emplace() 并发执行，造成数据竞争。是否触发取决于具体时序。

用同一个 strand 管理队列和写入

将 strand 保存为 sender 的成员。所有访问 _send_queue 的操作都应在这个 strand 上执行，包括入队、检查队列状态、移除队首和启动下一次写入。async_write 的完成处理器也应关联到同一个 strand。由于 async_write 是组合异步操作，关联 executor 会用于它的中间处理器。Boost.Asio 关于 strands 的说明

关键改动如下。这里的 asio::any_io_executor 适用于使用该 executor 类型的 Boost.Asio 版本。若项目使用较旧版本，应将成员类型改为 socket 对应的 executor 类型。

class sender : public std::enable_shared_from_this<sender>
{
public:
  explicit sender(std::shared_ptr<tcp::socket> sock)
      : _socket(std::move(sock)),
        _strand(asio::make_strand(_socket->get_executor()))
  {
  }

private:
  template <typename CompletionHandler>
  void async_send_message_impl(asio::const_buffer buf, CompletionHandler&& h)
  {
    // 先把所有队列操作安排到同一个 strand。
    asio::post(_strand,
        [self = shared_from_this(), buf,
         h = std::forward<CompletionHandler>(h)]() mutable
        {
          self->_send_queue.emplace(buf, std::move(h));
          if (self->_send_queue.size() == 1)
            self->send_queued_message();
        });
  }

  void send_queued_message()
  {
    asio::async_write(
        *_socket,
        _send_queue.front().first,
        asio::bind_executor(
            _strand,
            [self = shared_from_this()](system::error_code ec, std::size_t)
            {
              self->handle_send_complete(ec);
            }));
  }

  void handle_send_complete(system::error_code ec)
  {
    auto h = std::move(_send_queue.front().second);
    _send_queue.pop();

    // 先更新队列状态，再调用用户处理器，避免回调重入时仍看到旧队首。
    const auto& ex = asio::get_associated_executor(h, _socket->get_executor());
    asio::dispatch(ex, [h = std::move(h), ec]() mutable { h(ec); });

    if (!_send_queue.empty())
      send_queued_message();
  }

  std::shared_ptr<tcp::socket> _socket;
  asio::strand<asio::any_io_executor> _strand;
  std::queue<std::pair<
      asio::const_buffer,
      asio::any_completion_handler<void(system::error_code)>>> _send_queue;
};

async_send_message_impl() 通过 post(_strand, ...) 将队列操作安排到同一个 strand 上。这样，即使调用方来自任意线程，也不会直接修改队列。async_write 的处理器通过 bind_executor(_strand, ...) 绑定到同一个 strand，完成路径上的队列访问也会因此串行执行。async_write 进行期间，不要对同一个 stream 发起其他写操作；这里的队列确保同一时间最多只有一个活动写操作。Boost.Asio async_write 文档

注意事项

main() 中的局部 strand 没有被使用。应移除它，或将它传给 sender，并确保整个发送状态机都使用这个实例。
只有进入同一个 strand 的操作才会互不并发。若调用方从多个线程直接访问队列或启动写入，仍会产生竞争。
asio::const_buffer 只保存内存地址和长度，并不拥有数据。缓冲区指向的数据必须至少保持有效，直到对应的发送完成处理器被调用。示例中的 msg 具有静态存储期；如果传入临时字符串或可变容器，需要让发送队列持有相应数据，或另行保证其生命周期。
示例保留了原有的关联 executor 完成回调处理思路。实际工程中，应根据所用的 Boost 版本和 completion token 支持情况检查类型与接口。代码片段展示的是同步边界上的关键调整，并非对整份类的逐行替换。