基于asio-wrapper：如何不依赖sleep_for解决Boost.Asio数据合并问题？

我正在尝试使用Boost.Asio，测试代码基于一个asio-wrapper实现。目前我只能通过在多次调用client.Write后添加std::this_thread::sleep_for来避免数据合并，但希望找到不依赖该方法的解决方案。
此外我发现，当boost::asio的write函数处于忙碌状态时，该wrapper的Write函数会导致第二条消息丢失。

测试代码 #

#include<iostream>
#include <vector>
#include<thread>
#include<chrono>
#include "asio_wrapper.hpp"


class WritableHandlerImplServer : public cppeng::tcp::WritableHandler {
    public:
        void HandleCallback(std::vector<uint8_t> data,
                std::shared_ptr<cppeng::tcp::Writable> writable) override {
                // 转换为字符串
                std::string received(data.begin(), data.end());
                std::cout << "Server Rx: " << received << std::endl;

            }
            void NotifyClosed(std::shared_ptr<cppeng::tcp::Writable> ptr) override {
                std::cout << "Server: Connection closed!" << std::endl;
            }
        };


class WritableHandlerImplClient : public cppeng::tcp::WritableHandler {
    public:
        void HandleCallback(std::vector<uint8_t> data,
            std::shared_ptr<cppeng::tcp::Writable> writable) override {
            // 转换为字符串
            std::string received(data.begin(), data.end());
            std::cout << "Client Rx: " << received << std::endl;
            
            rx_flag_ = true;
        }
        void NotifyClosed(std::shared_ptr<cppeng::tcp::Writable> ptr) override {
            std::cout << "Client: Connection closed!" << std::endl;
        }
        bool rx_flag_{ false };
};


int main()
{
    /*测试代码*/
    try {

        WritableHandlerImplServer writable_handler_server;
        WritableHandlerImplClient writable_handler_client;

        cppeng::tcp::Server server(3000, writable_handler_server);
        server.Start();
        cppeng::tcp::Client client("127.0.0.1", 3000, writable_handler_client);
        client.Start(1000);

        
        client.Write("abcd");
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        client.Write("efgh");
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        client.Write("ijkl");
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        client.Write("mnop");
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        client.Write("qrst");
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        client.Write("uvwx");
        std::this_thread::sleep_for(std::chrono::milliseconds(5));
        client.Write("yz");
            
        
        server.Stop();
        client.Stop();

    }
    catch (std::exception& e) {
        std::cout << "exception :" << e.what() << "\n";
    }

    return EXIT_SUCCESS;
}

读写函数实现 #

void Connection::DoRead() {

    auto self(shared_from_this());
    boost::asio::async_read(socket_,
        boost::asio::buffer(read_buffer_.data(), read_buffer_.size()),
        boost::asio::transfer_at_least(1),
                    [this, self](std::error_code ec, std::size_t length) {
                        if (!ec) {
                        std::vector<uint8_t> rx_data(read_buffer_.begin(), read_buffer_.begin() + length);
                        writable_handler_.HandleCallback(std::move(rx_data), shared_from_this());
                        DoRead();
                        } else {
                        Close();
                        }
                    });
}

void Connection::DoWrite() {
    auto self(shared_from_this());
    boost::asio::async_write(socket_,
                    boost::asio::buffer(write_queue_.front().data(),
                                    write_queue_.front().size()),
                    [this, self](std::error_code ec, std::size_t /*length*/) {
                        if (!ec) {
                        write_queue_.pop();
                        if (!write_queue_.empty()) {
                            DoWrite();
                        } else {
                            write_busy_ = false;
                        }
                        } else {
                        Close();
                        }
                    });
}

Wrapper的Write函数实现 #

void Connection::Write(const std::string &data) {
    // 将数据加入队列
    write_queue_.emplace(data.begin(), data.end());
    // 仅当未在写入时启动写流程
    if (!write_busy_) {
        write_busy_ = true;
        DoWrite();
    }
    else {
         
    }
}

1. 修复消息丢失问题 #

消息丢失的核心原因是多线程竞态条件：write_queue_和write_busy_的操作未加同步保护，多个线程同时调用Write时，可能同时进入!write_busy_分支，导致DoWrite被重复触发，队列消息被错误跳过。

修复代码 #

给Connection类添加互斥锁，同步队列和状态变量的访问：

#include <mutex>

class Connection {
    // ... 原有成员 ...
    std::mutex write_mutex_;
};

void Connection::Write(const std::string &data) {
    std::lock_guard<std::mutex> lock(write_mutex_);
    write_queue_.emplace(data.begin(), data.end());
    if (!write_busy_) {
        write_busy_ = true;
        DoWrite();
    }
}

void Connection::DoWrite() {
    auto self(shared_from_this());
    std::lock_guard<std::mutex> lock(write_mutex_);
    if (write_queue_.empty()) {
        write_busy_ = false;
        return;
    }
    // 先拷贝队列头部数据，避免后续pop销毁数据
    auto front_data = write_queue_.front();
    write_queue_.pop();
    lock.unlock(); // 释放锁，异步操作无需持有锁等待

    boost::asio::async_write(socket_,
                    boost::asio::buffer(front_data.data(), front_data.size()),
                    [this, self](std::error_code ec, std::size_t /*length*/) {
                        if (!ec) {
                            std::lock_guard<std::mutex> lock(write_mutex_);
                            if (!write_queue_.empty()) {
                                DoWrite();
                            } else {
                                write_busy_ = false;
                            }
                        } else {
                            Close();
                        }
                    });
}

2. 解决数据合并问题 #

TCP是字节流协议，本身不区分消息边界，数据合并是正常行为。解决方法是在应用层实现消息帧协议，让接收方能准确拆分消息，彻底摆脱对sleep的依赖。

推荐方案：长度前缀帧 #

发送消息前先传输消息长度（用固定字节数的整数，如4字节uint32_t），接收方先读取长度，再读取对应长度的内容作为完整消息。

发送端修改Write函数

void Connection::Write(const std::string &data) {
    std::lock_guard<std::mutex> lock(write_mutex_);
    // 构造带长度前缀的消息帧
    uint32_t length = htonl(data.size()); // 转换为网络字节序
    std::vector<uint8_t> frame;
    frame.reserve(sizeof(length) + data.size());
    frame.insert(frame.end(), reinterpret_cast<uint8_t*>(&length), 
                 reinterpret_cast<uint8_t*>(&length) + sizeof(length));
    frame.insert(frame.end(), data.begin(), data.end());
    write_queue_.emplace(std::move(frame));
    if (!write_busy_) {
        write_busy_ = true;
        DoWrite();
    }
}

接收端修改DoRead逻辑

// Connection类添加成员变量
uint32_t pending_length_ = 0;

void Connection::DoReadLength() {
    auto self(shared_from_this());
    boost::asio::async_read(socket_,
        boost::asio::buffer(&pending_length_, sizeof(pending_length_)),
        [this, self](std::error_code ec, std::size_t /*length*/) {
            if (!ec) {
                pending_length_ = ntohl(pending_length_); // 转换为主机字节序
                DoReadData(pending_length_);
            } else {
                Close();
            }
        });
}

void Connection::DoReadData(uint32_t length) {
    auto self(shared_from_this());
    read_buffer_.resize(length);
    boost::asio::async_read(socket_,
        boost::asio::buffer(read_buffer_.data(), length),
        [this, self](std::error_code ec, std::size_t /*length*/) {
            if (!ec) {
                std::vector<uint8_t> rx_data(std::move(read_buffer_));
                writable_handler_.HandleCallback(std::move(rx_data), shared_from_this());
                DoReadLength(); // 继续读取下一个消息的长度
            } else {
                Close();
            }
        });
}

// 替换原DoRead函数
void Connection::DoRead() {
    DoReadLength();
}

总结 #

• 消息丢失问题：通过std::mutex同步队列和状态变量的访问，解决多线程竞态条件。
• 数据合并问题：实现应用层消息帧协议（长度前缀方案），让接收方准确拆分消息，无需依赖sleep。

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