如何用Boost.Beast处理WebSocket多写请求避免重叠async_write调用

问题背景 #

使用Boost.Beast实现WebSocket客户端时，WebSocket连接同一时间只能存在一个活跃的async_write调用。但现有代码的main循环会持续从队列取出请求并调用send_message，直接发起async_write，这会导致重叠的写操作，引发错误甚至消息丢失。需要设计可靠的机制确保所有待发送消息都能有序、安全地被发送。

现有核心代码片段：

// 原Websocket类的send_message实现
void send_message(std::string_view message)
{
    m_ws.async_write(
        boost::beast::net::buffer(std::string { message }),
        [self = this->shared_from_this()](boost::system::error_code error_code, size_t) {
            if (error_code)
            {
                SPDLOG_ERROR("Failed to send message");
                SPDLOG_ERROR("    error_code.message: {}", error_code.message());
                SPDLOG_ERROR("    error_code.what: {}", error_code.what());
            }
        }
    );
}

// main循环
auto main() -> int32_t {
    std::cout << "entrypoint" << std::endl;
    boost::asio::io_context ioctx;
    auto entry_point_ws = std::make_shared<EntryPoint>(ioctx);
    entry_point_ws->async_start("example.com", "443", "/ws/private/orders/control");
    
    while (true) {
        auto pseudo_request = "pseudo_request_popped_from_queue_pushed_by_another_thread";
        entry_point_ws->send_message(pseudo_request);
        ioctx.poll();
    }
}

核心解决方案：内部消息队列+串行化写操作 #

在WebSocket类内部维护一个待发送消息队列，同时标记当前是否有活跃的写操作。当调用send_message时：

1. 如果当前没有正在进行的写操作，直接发起async_write
2. 如果已有活跃写操作，将消息加入队列
3. 每一次async_write完成后，自动从队列取出下一条消息继续发送，直到队列为空

修改后的Websocket类关键代码 #

template<typename Derived>
class Websocket: public std::enable_shared_from_this<Websocket<Derived>>
{
    // ... 原有成员 ...
private:
    // 添加消息队列和写操作状态标记
    std::queue<std::string> m_write_queue;
    std::atomic<bool> m_writing_in_progress{false};
    std::mutex m_queue_mutex; // 多线程调用send_message时保护队列

    // 新增：启动下一次写操作
    void start_next_write()
    {
        std::lock_guard<std::mutex> lock(m_queue_mutex);
        if (m_write_queue.empty() || m_writing_in_progress)
            return;

        m_writing_in_progress = true;
        auto message = std::move(m_write_queue.front());
        m_write_queue.pop();

        auto self = this->shared_from_this();
        m_ws.async_write(
            boost::beast::net::buffer(message),
            [self](boost::system::error_code error_code, size_t) {
                if (error_code)
                {
                    SPDLOG_ERROR("Failed to send message");
                    SPDLOG_ERROR("    error_code.message: {}", error_code.message());
                    SPDLOG_ERROR("    error_code.what: {}", error_code.what());
                    // 可选：处理错误，比如清空队列、重连等
                }

                // 标记写操作完成，启动下一次
                self->m_writing_in_progress = false;
                self->start_next_write();
            }
        );
    }

public:
    // 修改send_message实现
    void send_message(std::string_view message)
    {
        {
            std::lock_guard<std::mutex> lock(m_queue_mutex);
            m_write_queue.emplace(message);
        }
        // 尝试启动写操作（如果当前没有活跃写操作）
        start_next_write();
    }

    // ... 原有方法 ...
};

方案优势 #

1. 贴合Asio异步模型：无需额外线程，利用Asio的事件循环串行处理写操作，避免线程切换开销
2. 自动消息排队：所有消息按顺序发送，不会出现重叠写操作
3. 线程安全：通过互斥锁保护队列，支持多线程调用send_message
4. 简洁可靠：逻辑集中在WebSocket类内部，无需额外的跨线程队列和线程管理

针对多线程场景的优化 #

如果是单生产者单消费者场景（比如只有主线程推送消息，Asio线程处理写操作），可以用无锁队列替代互斥锁，进一步提升性能：

#include <boost/lockfree/queue.hpp>

// 替换队列和锁
boost::lockfree::queue<std::string> m_write_queue{1024}; // 预分配容量
std::atomic<bool> m_writing_in_progress{false};

// 修改send_message和start_next_write中的队列操作，去掉互斥锁

关于初步方案的补充 #

你提到的"单独线程运行io_context::run，主线程推消息到无锁队列"方案是可行的，但相比内部队列方案，需要额外管理线程和跨线程队列，复杂度更高。内部队列方案更符合Asio的异步设计理念，代码更简洁，维护成本更低。

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