为boost::asio::ip::tcp::iostream实现unget类功能的现代C++方案

要解决TCP流无法回退字节的问题，最直接的方案是在tcp::iostream之上套一层自定义缓存流缓冲区，把从TCP流读取的字节先暂存到本地缓冲区里，所有读取操作都针对这个本地缓冲区进行，回退操作直接操作缓冲区即可。下面用现代C++（C++11及以上）实现这个方案：

实现自定义可回退的streambuf #

C++标准库的std::streambuf是所有iostream的底层核心，我们可以继承它实现自己的缓存逻辑：

#include <streambuf>
#include <vector>
#include <boost/asio/ip/tcp.hpp>
#include <cstring>

class BackableTcpStreambuf : public std::streambuf {
public:
    explicit BackableTcpStreambuf(boost::asio::ip::tcp::iostream& tcp_stream, std::size_t buffer_size = 4096)
        : tcp_stream_(tcp_stream), buffer_(buffer_size) {
        // 初始化缓冲区指针：初始为空，读指针和尾指针重合
        setg(buffer_.data(), buffer_.data(), buffer_.data());
    }

    // 自定义方法：将指定字节放回缓冲区（支持多字节回退）
    void putback_bytes(const char* data, std::size_t size) {
        if (size == 0) return;

        // 检查缓冲区头部是否有足够空间容纳回退字节
        const std::ptrdiff_t available_space = gptr() - eback();
        if (available_space < static_cast<std::ptrdiff_t>(size)) {
            // 空间不足时，将已读数据后移，腾出头部空间
            const std::size_t used_bytes = egptr() - gptr();
            std::memmove(buffer_.data() + size, gptr(), used_bytes);
            // 更新指针：调整读起始位置，保持已读数据的可访问性
            setg(buffer_.data(), buffer_.data() + size, buffer_.data() + size + used_bytes);
        }

        // 复制回退字节到缓冲区头部，调整读指针位置
        std::memcpy(gptr() - size, data, size);
        gbump(-static_cast<int>(size));
    }

protected:
    // 缓冲区空时，从底层TCP流读取数据填充
    int_type underflow() override {
        if (gptr() < egptr()) {
            return traits_type::to_int_type(*gptr());
        }

        // 从TCP流读取数据到缓冲区
        const std::streamsize bytes_read = tcp_stream_.readsome(buffer_.data(), buffer_.size());
        if (bytes_read <= 0) {
            return traits_type::eof();
        }

        // 更新缓冲区可读区域指针
        setg(buffer_.data(), buffer_.data(), buffer_.data() + bytes_read);
        return traits_type::to_int_type(*gptr());
    }

private:
    boost::asio::ip::tcp::iostream& tcp_stream_;
    std::vector<char> buffer_; // 本地缓存缓冲区
};

使用方式 #

把自定义streambuf和标准std::istream绑定，所有读取操作通过这个istream进行，需要回退时调用putback_bytes方法即可：

#include <iostream>
#include <cstdint>
#include <arpa/inet.h> // 用于字节序转换（跨平台可替换为boost的字节序工具）

// 假设你的Message结构体定义
struct Message {
    uint32_t length;
    std::vector<char> payload;
};

// 重载>>运算符，处理半消息时回退已读字节
std::istream& operator>>(std::istream& stream, Message& msg) {
    // 读取长度字段（网络字节序转主机字节序）
    uint32_t network_length;
    if (!stream.read(reinterpret_cast<char*>(&network_length), sizeof(network_length))) {
        return stream;
    }
    msg.length = ntohl(network_length);

    // 尝试读取payload
    msg.payload.resize(msg.length);
    const std::streamsize bytes_read = stream.readsome(msg.payload.data(), msg.payload.size());
    
    // 若读取字节数不足，回退长度字段，标记流失败以便重试
    if (bytes_read < static_cast<std::streamsize>(msg.length)) {
        auto* backable_buf = dynamic_cast<BackableTcpStreambuf*>(stream.rdbuf());
        if (backable_buf) {
            backable_buf->putback_bytes(reinterpret_cast<const char*>(&network_length), sizeof(network_length));
        }
        stream.setstate(std::ios::failbit);
    }
    return stream;
}

int main() {
    boost::asio::ip::tcp::iostream tcp_stream("example.com", "1234");
    if (!tcp_stream) {
        std::cerr << "连接失败: " << tcp_stream.error().message() << std::endl;
        return 1;
    }

    BackableTcpStreambuf backable_buf(tcp_stream);
    std::istream input_stream(&backable_buf);

    Message msg;
    while (true) {
        if (input_stream >> msg) {
            // 处理完整消息
            std::cout << "收到消息，长度: " << msg.length << std::endl;
        } else {
            // 清除错误状态，等待更多数据后重试
            input_stream.clear();
        }
    }

    return 0;
}

关键细节说明 #

• 缓冲区管理：默认用4096字节的std::vector<char>作为缓存，可根据业务场景调整大小；当缓冲区空时，通过underflow()从TCP流批量读取数据填充。
• 回退逻辑：putback_bytes()会自动处理缓冲区空间不足的情况，通过移动已读数据腾出头部空间，确保回退的字节能被下一次读取正确获取。
• 兼容性：自定义streambuf完全兼容标准istream接口，现有operator>>重载无需大幅修改，只需替换底层流对象即可。

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