如何在Rust中结合BufMut与BufReader<TcpStream>使用？

我有一个需要在不同网络协议处理程序间共享的解码函数，签名如下：

pub fn handle_decode(buf: &mut bytes::BytesMut) -> anyhow::Result<DataType> {}

选择bytes库是因为它能很好处理字节序、浅拷贝，支持按序读取字节且克隆成本低。

在TCP处理流程中，我希望用BufReader<TcpStream>减少反复调用read()系统调用的开销，但不知道怎么把BufReader和bytes::BytesMut结合起来，兼顾两者优势。下面是我的最简示例，其中handle_decode(&mut bufreader)这行无法运行：

use anyhow::Context;
use std::io::BufReader;
use std::net::TcpListener;

fn bind_accept() -> anyhow::Result<()> {
    fn handle_decode(buf: &mut bytes::BytesMut) -> anyhow::Result<()> {
        unimplemented!()
    }

    let mut l = TcpListener::bind("1234")?;

    let (sock, _) = l.accept().context("should bind to provided port")?;

    let bufreader = BufReader::new(sock);
    let mut buf = bytes::BytesMut::with_capacity(1024);

    // 当前状态下该行代码无法运行
    handle_decode(&mut bufreader)
}

核心思路是把BufReader中的数据读取到BytesMut中，再传给handle_decode，以下是两种可行实现方式：

方式一：手动管理缓冲区读取 #

利用std::io::Read trait的read方法，将BufReader的数据填充到BytesMut的可写区域，需要手动标记已写入的数据长度：

use anyhow::Context;
use bytes::BytesMut;
use std::io::{Read, BufReader};
use std::net::TcpListener;

fn bind_accept() -> anyhow::Result<()> {
    fn handle_decode(buf: &mut BytesMut) -> anyhow::Result<()> {
        // 示例解码逻辑：需处理数据不足的情况
        if buf.len() < 4 {
            return Ok(()); // 数据不够组成完整包，等待后续读取
        }
        let len = u32::from_be_bytes(buf[0..4].try_into()?) as usize;
        if buf.len() < 4 + len {
            return Ok(());
        }
        let data = buf.split_to(4 + len);
        // 处理解析出的data...
        Ok(())
    }

    let mut l = TcpListener::bind("1234")?;
    let (sock, _) = l.accept().context("failed to accept connection")?;
    let mut bufreader = BufReader::new(sock);
    let mut buf = BytesMut::with_capacity(1024);

    loop {
        // 获取BytesMut的可写切片
        let spare = buf.spare_mut();
        let n = bufreader.read(spare)?;
        if n == 0 {
            break; // 连接已关闭
        }
        // 标记已写入的字节数
        buf.advance_mut(n);
        
        // 循环处理所有完整数据包
        while handle_decode(&mut buf).is_ok() {}
    }

    Ok(())
}

方式二：使用bytes库的BufReadExt扩展 #

bytes库提供的BufReadExt trait封装了更便捷的读取逻辑，自动处理BytesMut的扩容，无需手动管理缓冲区：
首先确保Cargo.toml中bytes依赖启用std特性（默认已启用）：

[dependencies]
bytes = "1.5"
anyhow = "1.0"

然后修改代码：

use anyhow::Context;
use bytes::{BufReadExt, BytesMut};
use std::io::BufReader;
use std::net::TcpListener;

fn bind_accept() -> anyhow::Result<()> {
    fn handle_decode(buf: &mut BytesMut) -> anyhow::Result<()> {
        unimplemented!()
    }

    let mut l = TcpListener::bind("1234")?;
    let (sock, _) = l.accept().context("failed to accept connection")?;
    let mut bufreader = BufReader::new(sock);
    let mut buf = BytesMut::with_capacity(1024);

    loop {
        // 自动读取数据到BytesMut，自动扩容
        let n = bufreader.read_buf(&mut buf)?;
        if n == 0 {
            break;
        }

        // 循环处理所有完整数据包
        while handle_decode(&mut buf).is_ok() {}
    }

    Ok(())
}

关键注意点 #

• handle_decode必须处理数据不足的场景：当BytesMut中的字节数不足以组成完整数据包时，需返回Ok(())，等待后续读取更多数据。
• 循环调用handle_decode是因为单次读取可能包含多个完整数据包，需全部处理完毕再继续读取新数据。
• 两层缓冲（BufReader的系统调用缓冲 + BytesMut的字节操作缓冲）结合，既减少了系统调用开销，又保留了bytes库的灵活字节处理能力。

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