如何在等待async_read时解除boost::asio::io_context阻塞？

问题场景 #

我用Boost.Asio和Beast实现WebSocket客户端，需要每10秒发送一次心跳，但写请求会卡在队列里，只有服务器主动发消息后，队列里的心跳任务才会被处理。

核心代码片段 #

消息监听逻辑 #

this->ioContext.run();

thread heartbeatThread(&client::heartbeatCycle, this);

while (this->p->is_socket_open()) {
    this->ioContext.restart();
    this->p->asyncQueue("", true);
    this->ioContext.run();
}

任务队列处理函数asyncQueue #

void session::asyncQueue(const string& payload, const bool& madeAfterLoop)
{
    if(!payload.empty())
    {
        queue_.emplace_back(payload);
    }

    if(payload.empty() && madeAfterLoop)
    {
        queue_.emplace_back("KEEPALIVE");
    }

    // 如果有消息要发送，执行异步写
    if(!currentlyQueued_ && !queue_.empty() && queue_.at(0) != "KEEPALIVE")
    {
        currentlyQueued_ = true;
        ws_.async_write(
                net::buffer(queue_.at(0)),
                beast::bind_front_handler(
                        &session::on_write,
                        shared_from_this()));
        queue_.erase(queue_.begin());
    }

    // 如果没有可发送的消息，执行异步读保持连接
    if(!currentlyQueued_ && !queue_.empty())
    {
        currentlyQueued_ = true;
        ws_.async_read(
                buffer_,
                beast::bind_front_handler(
                        &session::on_read,
                        shared_from_this()));
        queue_.erase(queue_.begin());
    }
}

心跳线程逻辑 #

void client::heartbeatCycle()
{
    while(this->p->is_socket_open())
    {
        this->p->asyncQueue(bot::websocket::sendEvents::getHeartbeatEvent(cache_), true );
        this_thread::sleep_for(chrono::milliseconds(10000));
    }
}

on_read回调 #

currentlyQueued_ = false;
asyncQueue();

问题细节 #

• 无任务时，asyncQueue会调用async_read，导致io_context一直阻塞在读取状态，只有服务器发消息触发on_read后，currentlyQueued_才会被设为false，队列里的心跳任务才能被处理。
• 心跳线程尝试发送心跳时，任务卡在队列；如果强制发送，会触发断言错误：Assertion failed: (id_ != T::id), function try_lock, file soft_mutex.hpp, line 89.

疑问 #

1. 能不能在服务器不发消息的情况下解除io_context的阻塞？
2. 如果不能，有没有办法模拟服务器发送消息来触发队列处理？

解决方案建议 #

1. 修复线程安全问题，避免断言错误 #

断言错误的根源是心跳线程直接操作了非线程安全的队列和状态变量。所有对queue_、currentlyQueued_的操作必须切换到io_context线程执行，用io_context::post来同步：

void client::heartbeatCycle()
{
    while(this->p->is_socket_open())
    {
        auto heartbeat = bot::websocket::sendEvents::getHeartbeatEvent(cache_);
        // 将队列操作post到io_context线程执行
        this->p->get_io_context().post([self = this->p, heartbeat]() {
            self->asyncQueue(heartbeat, true);
        });
        this_thread::sleep_for(chrono::milliseconds(10000));
    }
}

2. 重构队列逻辑，避免io_context被async_read永久阻塞 #

当前逻辑在无写任务时，会用async_read占住io_context，导致心跳任务无法调度。调整为优先处理写任务，仅在队列空时保持异步读：

// 抽离队列处理逻辑，单独调用
void session::processQueue()
{
    currentlyQueued_ = true;
    // 优先处理所有写任务（包括心跳，不再区分KEEPALIVE）
    if(!queue_.empty())
    {
        auto msg = queue_.front();
        queue_.erase(queue_.begin());
        ws_.async_write(
            net::buffer(msg),
            beast::bind_front_handler(&session::on_write, shared_from_this())
        );
        return;
    }

    // 队列空了，再发起异步读监听服务器消息
    ws_.async_read(
        buffer_,
        beast::bind_front_handler(&session::on_read, shared_from_this())
    );
}

// 修改asyncQueue逻辑
void session::asyncQueue(const string& payload, const bool& madeAfterLoop)
{
    if(!payload.empty())
    {
        queue_.emplace_back(payload);
    }
    if(payload.empty() && madeAfterLoop)
    {
        queue_.emplace_back("KEEPALIVE");
    }

    // 如果当前没有正在执行的IO任务，立即处理队列
    if(!currentlyQueued_)
    {
        processQueue();
    }
}

// 修改on_write回调，写完后立即检查队列
void session::on_write(beast::error_code ec, std::size_t bytes_transferred)
{
    boost::ignore_unused(bytes_transferred);
    if(ec) return handle_error(ec);

    currentlyQueued_ = false;
    processQueue();
}

// 修改on_read回调，读完后检查队列
void session::on_read(beast::error_code ec, std::size_t bytes_transferred)
{
    boost::ignore_unused(bytes_transferred);
    if(ec) return handle_error(ec);

    // 处理服务器消息的逻辑...

    currentlyQueued_ = false;
    processQueue();
}

3. 关于解除io_context阻塞的疑问 #

io_context::run()会在没有待处理异步任务时返回，你当前的问题是async_read一直处于挂起状态，导致io_context有未完成任务无法退出。可以用boost::asio::steady_timer设置定时任务，定期唤醒io_context检查队列状态，避免永久阻塞在async_read上。

模拟服务器发送消息完全没必要，解决线程安全和任务调度的根本问题即可。

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