异步读写同时运行时读操作被挤占的原因及解决方法问询

现象描述 #

单独运行异步读或异步写功能时均正常无延迟，但同时运行两者时，Socket写操作会占据大部分时间，读操作频次大幅降低。

相关代码 #

测试主函数 #

void AddData(std::shared_ptr<std::queue<std::string>> stores)
{
    stores->push("A\n");
    stores->push("B\n");
    stores->push("C\n");
    stores->push("D\n");
    stores->push("E\n");
}

int main()
{
    std::shared_ptr<std::queue<std::string>> stores = std::make_shared<std::queue<std::string>>();
    AddData(stores);

    boost::asio::io_context io_context;
    ServerTools::Server srv(io_context, 15004);

    std::shared_ptr<ServerTools::session> sess = srv.accept_one();
    sess->SetStoreForServer(stores);
    sess->StartAsyncRead();
    sess->StartAsyncSend();

    io_context.run();
}

异步写函数实现 #

void ServerTools::session::StartAsyncSend()
{
    if (!p_messageStore->empty())
    {
        std::string data = p_messageStore->front();

        boost::asio::async_write(socket, boost::asio::buffer(data),         
        [this](boost::system::error_code ec, std::size_t bytes_transferred)
        {
            if (!ec) 
            {
                std::cout << "Sending " << bytes_transferred << " bytes" << '\n';
                this->p_messageStore->pop();
            }
            else
            {
                std::cerr << "Mistake answer: " << ec.message() << std::endl;
            }

            std::this_thread::sleep_for(std::chrono::seconds(3));
            StartAsyncSend();
        });
    }
}

异步读函数实现 #

void ServerTools::session::StartAsyncRead()
{
    std::vector<char> buffer(100);   
    boost::asio::async_read(socket, boost::asio::buffer(buffer), [this](const boost::system::error_code& ec, std::size_t size_transferred)
    {
        if (!ec)
        {
            std::cout << "Received " << size_transferred << " bytes" << std::endl;
        }
        StartAsyncRead();
    });
}

原因分析 #

1. 阻塞调用占用IO线程：StartAsyncSend的回调中使用了std::this_thread::sleep_for，这是阻塞当前线程的同步调用。默认情况下io_context单线程运行，这个sleep会直接卡住整个IO事件循环，导致读操作的回调无法被及时调度，表现为读频次大幅降低。
2. 读操作的不合理等待：async_read要求读取满指定的buffer大小（100字节），如果客户端未发送足够数据，读操作会一直处于等待状态，进一步加剧了读写任务的调度不平衡。

修复方法 #

1. 替换阻塞sleep为异步定时器 #

不要在IO线程中执行阻塞操作，改用Boost.Asio提供的异步定时器实现延迟，避免卡住IO事件循环：

void ServerTools::session::StartAsyncSend()
{
    if (!p_messageStore->empty())
    {
        std::string data = p_messageStore->front();

        boost::asio::async_write(socket, boost::asio::buffer(data),         
        [this](boost::system::error_code ec, std::size_t bytes_transferred)
        {
            if (!ec) 
            {
                std::cout << "Sending " << bytes_transferred << " bytes" << '\n';
                this->p_messageStore->pop();
            }
            else
            {
                std::cerr << "Mistake answer: " << ec.message() << std::endl;
            }

            // 用异步定时器替代阻塞sleep
            boost::asio::steady_timer timer(socket.get_executor(), std::chrono::seconds(3));
            timer.async_wait([this](boost::system::error_code ec) {
                if (!ec) {
                    StartAsyncSend();
                }
            });
        });
    }
}

2. 修正读操作的等待逻辑 #

将async_read改为async_read_some，只要有可用数据就触发回调，避免因等待满buffer而长时间阻塞：

void ServerTools::session::StartAsyncRead()
{
    std::vector<char> buffer(100);   
    boost::asio::async_read_some(socket, boost::asio::buffer(buffer), [this](const boost::system::error_code& ec, std::size_t size_transferred)
    {
        if (!ec)
        {
            std::cout << "Received " << size_transferred << " bytes" << std::endl;
            StartAsyncRead();
        }
        else if (ec != boost::asio::error::eof) {
            std::cerr << "Read error: " << ec.message() << std::endl;
            StartAsyncRead(); // 非EOF错误可尝试继续读取
        }
        // 遇到EOF则停止读操作，避免无限调用
    });
}

3. 确保队列线程安全（可选） #

如果后续有其他线程操作p_messageStore，需要给队列加锁（比如std::mutex），避免并发访问导致的未定义行为。

总结 #

不需要刻意组织异步读写的执行序列，只要避免在IO线程中执行阻塞操作，改用Asio的异步原语，同时修正读操作的等待逻辑，就能让读写任务在IO事件循环中正常并发调度，解决读写不平衡的问题。

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