如何在两个线程中分离Boost异步读与异步写操作？

1. 线程分工与IO上下文的合理设计 #

不需要给读、写操作各开独立线程，更高效的方式是用单个io_context搭配多工作线程，再通过strand序列化读写操作避免竞态；如果一定要强制读写在不同线程执行，可以给读、写分别创建独立的io_context，每个上下文绑定一个工作线程。
定时发送逻辑可以直接用steady_timer绑定到负责写操作的io_context或strand上，定时触发写任务即可。

2. 数据预处理与线程同步方案 #

预处理线程和读回调之间的交互，用线程安全队列实现数据传递是最稳妥的方式：

• 读回调收到服务器消息后，将原始数据或初步处理结果推入输入队列；
• 预处理线程从输入队列取数据，生成待发送内容后推入发送队列；
• 定时写任务从发送队列取数据执行异步写操作，如果队列为空则发送预设的旧数据。

3. 关键代码示例 #

简易线程安全队列实现 #

#include <queue>
#include <mutex>
#include <condition_variable>

template<typename T>
class ThreadSafeQueue {
public:
    void push(T item) {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_queue.push(std::move(item));
        m_cv.notify_one();
    }

    bool try_pop(T& item) {
        std::lock_guard<std::mutex> lock(m_mutex);
        if (m_queue.empty()) return false;
        item = std::move(m_queue.front());
        m_queue.pop();
        return true;
    }

    void wait_and_pop(T& item) {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_cv.wait(lock, [this] { return !m_queue.empty(); });
        item = std::move(m_queue.front());
        m_queue.pop();
    }

    bool empty() const {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_queue.empty();
    }

private:
    std::queue<T> m_queue;
    mutable std::mutex m_mutex;
    std::condition_variable m_cv;
};

定时发送逻辑实现 #

void start_periodic_write(boost::asio::io_context& write_io, boost::asio::ip::tcp::socket& socket, ThreadSafeQueue<std::string>& send_queue, std::chrono::seconds interval) {
    auto timer = std::make_shared<boost::asio::steady_timer>(write_io, interval);
    timer->async_wait([timer, &socket, &send_queue, interval](const boost::system::error_code& ec) {
        if (!ec) {
            std::string data;
            if (send_queue.try_pop(data)) {
                boost::asio::async_write(socket, boost::asio::buffer(data),
                    [](const boost::system::error_code& ec, std::size_t) {
                        if (ec) {
                            // 处理发送错误，比如断开重连、日志记录
                        }
                    });
            } else {
                // 发送预设旧数据
                std::string old_data = "default_stale_data";
                boost::asio::async_write(socket, boost::asio::buffer(old_data),
                    [](const boost::system::error_code& ec, std::size_t) {
                        if (ec) {
                            // 处理发送错误
                        }
                    });
            }
            // 循环启动定时器，维持定时发送
            start_periodic_write(write_io, socket, send_queue, interval);
        }
    });
}

4. 核心注意事项 #

• strand的强制使用：如果多个线程共享同一个socket，必须用strand序列化所有异步操作，避免读写操作交叉导致的未定义行为；
• 资源优雅释放：关闭客户端时要先停止io_context，等待所有工作线程退出，避免内存泄漏或资源残留；
• 全链路错误处理：每个异步操作的回调都要处理错误码，比如连接断开、读写失败等场景，及时做资源清理或重连逻辑。

内容的提问来源于stack exchange，提问作者And.Re