异步使用队列

15 分钟阅读 • 935 字 + 1551 词

1. 避免并发写入（数据竞争）

问题

直接发送的陷阱 ：若多个线程或回调同时调用异步发送（如 async_write ），会导致多个未完成的异步写操作同时操作同一个 Socket。由于 TCP 是流式协议，操作系统内核不保证多个并发写操作的顺序性，可能引发数据 交叉混乱 （如 ABC 和 123 可能被接收为 A1B2C3 ）。
示例场景 ：用户 A 的线程发送数据 "Hello" ，用户 B 的线程同时发送 "World" ，最终 Socket 实际发送的数据可能是 HWeolrllod 。

解决方案

队列序列化 ：将待发送数据按顺序存入队列，确保同一时间 只有一个异步写操作 在执行。当前写操作完成后，从队列中取出下一个数据继续发送。

2. 保证发送顺序性

问题

异步无序性 ：异步操作完成的顺序不确定。若直接连续调用 async_write ，实际发送顺序可能与调用顺序不一致。
示例场景 ：依次调用 async_write("A") 和 async_write("B") ，但由于网络延迟或调度， B 可能先于 A 发送完成。

解决方案

队列 FIFO 保证 ：发送队列按先进先出（FIFO）顺序处理数据，确保消息严格按照调用顺序发送。

3. 资源管理与生命周期安全

问题

数据生命周期风险 ： async_write 的缓冲区（如 boost::asio::buffer ）必须保证在异步操作完成前有效。若直接传递栈内存或临时对象，可能因作用域结束导致缓冲区被释放，引发内存错误。 错误示例 ：
```
void SendData() {
    char buffer[] = "Hello"; // 栈内存
    async_write(socket, boost::asio::buffer(buffer), [](...){}); 
    // 函数返回后 buffer 被销毁，但异步操作可能仍在进行！
}
```

解决方案

队列持有数据所有权 ：将待发送数据封装为堆对象（如 std::shared_ptr ）并存入队列，由队列管理其生命周期，确保异步操作期间数据有效。

4. 流量控制（背压，Backpressure）

问题

发送速率不匹配 ：若数据生产速率远高于网络发送速率，持续触发 async_write 会导致待发送数据在内存中无限堆积，最终内存耗尽。

解决方案

队列作为缓冲区 ：队列大小可设置上限，当队列满时拒绝新数据（或采取其他流控策略），避免资源耗尽。

实现模式详解

代码框架

class Session {
    std::queue<std::shared_ptr<Msg>> send_queue; // 发送队列
    std::mutex queue_mutex;                      // 互斥锁保护队列
    bool is_writing = false;                     // 标记是否正在发送

    void Send(std::shared_ptr<Msg> msg) {
        std::lock_guard<std::mutex> lock(queue_mutex);
        send_queue.push(msg);
        if (!is_writing) {
            StartAsyncWrite(); // 触发异步发送
        }
    }

    void StartAsyncWrite() {
        is_writing = true;
        auto& msg = send_queue.front();
        async_write(socket, boost::asio::buffer(msg->data), 
            [this, self=shared_from_this()](boost::system::error_code ec, size_t) {
                std::lock_guard<std::mutex> lock(queue_mutex);
                send_queue.pop(); // 发送完成，移除队列头部
                is_writing = false;
                if (!send_queue.empty()) {
                    StartAsyncWrite(); // 继续发送下一个
                }
            });
    }
};

关键步骤

数据入队 ：所有发送请求先将数据存入队列，而非直接调用 async_write 。
串行发送 ：检查当前是否正在发送（ is_writing 标记），若空闲则触发异步写操作。
回调处理 ：在异步写完成回调中：
- 移除已发送的队列头部数据。
- 若队列非空，继续发送下一个数据。

对比直接发送的缺陷

场景	直接发送	队列发送
并发写入	数据交叉混乱	顺序发送，无并发
数据生命周期	栈数据可能失效	队列持有数据所有权，保证有效性
流量控制	内存易失控	队列大小可控，支持背压
顺序性	无法保证	严格 FIFO 顺序

总结

通过发送队列管理异步写入操作，本质是 将并发操作转化为串行化任务链 ，从而：

避免数据竞争 ：同一时间只有一个异步写操作。
保证顺序性 ：严格按调用顺序发送。
管理资源 ：安全控制数据生命周期。
支持流量控制 ：防止内存溢出。

这是异步服务器设计中处理发送逻辑的标准模式，兼顾了性能、安全性和可维护性。

异步使用队列

目录