Axum 设计与实现

第17章 Body 处理与流式响应

作者 杨艺韬 · 13,567 字 · 发布于 · 更新于

第 15、16 章讲 axum::serve 的连接层。这一章下沉一层——到数据层。Body 是 HTTP 请求和响应的"主要载体"——URL 和 header 只是元信息，真正的负载在 body。axum 里 Body 的类型设计、流式机制、缓冲工具是整个数据流的中枢。

前面章节零散提到过 Body：

  • 第 6 章 body 单次消费：FromRequest 消费 Body、FromRequestParts 借用 Parts
  • 第 8 章 Multipart：Body 按 stream 分 field 流式读
  • 第 10 章 SSE：Sse::new(stream) 把 Stream<Item = Result<Event, E>> 包装成 Body
  • 第 10 章 文件下载：Body::from_stream(ReaderStream::new(file)) 流式返回
  • 第 15 章 serve：hyper 通过 Body 读字节、写字节

这一章把这些分散的使用统一起来——讲清楚 Body 的 API、类型结构、流式机制。读完之后你应该能回答：为什么 Body 能既是缓冲的又是流式的？Body::from_stream 内部做了什么？到底什么时候该 to_bytes 缓冲、什么时候该 stream？

为什么 body 是 HTTP server 框架的中枢

HTTP server 框架的工作本质上就是"字节进字节出"——所有路由、提取器、middleware 最终都是"对字节的某种处理"。body 是这些字节的 载体——一个框架对 body 的抽象质量基本决定了它能做什么。

axum 的 body 设计反映了几个成熟的工程取舍：

  • 不造新 trait：用 http_body::Body——避免生态分裂
  • 类型擦除但高效：UnsyncBoxBody + try_downcast——简单 API + 零重复包装开销
  • 流式和缓冲对称：from_stream / into_data_stream / to_bytes——用户按场景选
  • Bytes 零拷贝：整条路径共享引用、不做字节 memcpy

这些取舍让 axum 的 body 处理既语言好写（handler 里直接 Body::from(...)）又运行时高效（零拷贝 + 流式）。常见的用法是 body 先全缓冲到内存再交给 handler（比如 Express 里挂 body-parser 这类中间件之后）——axum 则把流式和缓冲都做成 body 的一等用法。

Body 的两重身份

axum 的 Body 有两种"身份"：

作为类型（axum::body::Body）：用户 handler 里见到的、响应和请求里携带的——就是这一个具体类型。没有泛型、没有 trait 对象、handler 签名里直接写 Body 即可。

作为协议（http_body::Body trait）：hyper 读写 HTTP body 的通用接口——poll_frame、size_hint、is_end_stream 三件套。所有 HTTP 相关 crate（axum、hyper、tonic、reqwest）都遵守这个 trait。

两个身份的关系：axum::Body（具体类型）实现了 http_body::Body（trait）。axum 提供一个"开箱即用的 body 类型"——背后是标准 trait 的具体包装。

这让 axum 的 body 处理 API 简单（就一个类型）但生态兼容（符合 http_body 规范）——两全其美。

Body：类型擦除的 http_body

Body 的定义在 axum-core/src/body.rs:39，依赖第 13 行的 BoxBody 类型别名：

// axum-core/src/body.rs:13
type BoxBody = http_body_util::combinators::UnsyncBoxBody<Bytes, Error>;

// axum-core/src/body.rs:37-39
#[must_use]
#[derive(Debug)]
pub struct Body(BoxBody);

一个 newtype 包装 UnsyncBoxBody。层层分解：

http_body_util::combinators::UnsyncBoxBody<Bytes, Error>：http_body::Body 的 trait object wrapper。http_body::Body 是 HTTP body 的标准 trait（hyper / tonic / reqwest 都基于它）——axum 用类型擦除存储任意实现者。

UnsyncBoxBody vs BoxBody：前者不要求 Sync、后者要求。axum 选 UnsyncBoxBody 因为：

  • 大多数 body 是从单个 task 处理的——Send 够、不需要 Sync
  • Sync bound 会把一些合法的 body 类型排除（比如内部持有 !Sync Stream 的 body——第 10 章 SSE 的 SseBody 正是为此才在 Stream 外包了一层 SyncWrapper）

Bytes 作为 data type：http_body::Body::Data 关联类型固定为 bytes::Bytes——最通用的字节容器、零拷贝共享、hyper / h2 等都用 Bytes。

Error = axum_core::Error：第 12 章讲过的 axum 通用错误 wrapper——BoxError 的 newtype。body 的任何错误统一成 axum::Error、downstream 代码只处理一种错误类型。

flowchart TB
    classDef axum fill:#efe,stroke:#080
    classDef std fill:#fef,stroke:#808
    classDef impl fill:#eef,stroke:#066
    User["用户侧<br/>handler 返回 / 接收 Body"]:::axum
    Body["axum::body::Body"]:::axum
    BoxBody["http_body_util::UnsyncBoxBody&lt;Bytes, Error&gt;"]:::std
    HttpBody["http_body::Body trait<br/>poll_frame / size_hint / is_end_stream"]:::std
    Impls["实现者"]:::impl
    Full["Full&lt;Bytes&gt; (全缓冲)"]:::impl
    Empty["Empty"]:::impl
    StreamBody["StreamBody&lt;S&gt; (用户 Stream)"]:::impl
    Incoming["hyper::body::Incoming (来自网络)"]:::impl
    User --> Body
    Body --> BoxBody
    BoxBody -.实现.-> HttpBody
    BoxBody -.擦除.-> Impls
    Impls --> Full
    Impls --> Empty
    Impls --> StreamBody
    Impls --> Incoming

用户看到一个统一的 Body 类型——下层实际是多种具体 body 类型的类型擦除包装。

Body::new 的 try_downcast 优化

axum-core/src/body.rs:41-49：

impl Body {
    pub fn new<B>(body: B) -> Self
    where
        B: http_body::Body<Data = Bytes> + Send + 'static,
        B::Error: Into<BoxError>,
    {
        try_downcast(body).unwrap_or_else(|body| Self(boxed(body)))
    }
}

try_downcast<T, K>(k: K) -> Result<T, K> 的实现在 axum-core/src/body.rs:23-34：

pub(crate) fn try_downcast<T, K>(k: K) -> Result<T, K>
where T: 'static, K: Send + 'static,
{
    let mut k = Some(k);
    if let Some(k) = <dyn std::any::Any>::downcast_mut::<Option<T>>(&mut k) {
        Ok(k.take().unwrap())
    } else {
        Err(k.unwrap())
    }
}

核心：尝试把 K downcast 成 T。如果 K 本身就是 T（TypeId 相等），直接返回——不做任何包装。

在 Body::new 的语境下：如果用户传入的 body 本来就是 Body 类型（比如中间件里 Body::new(existing_body)），try_downcast 直接返回那个 Body——不重新 box 一层。

这个优化避免了"不必要的类型擦除嵌套"——Body::new(Body::new(Body::new(inner))) 等价于 Body::new(inner)、不会层层 box。生产里中间件栈可能多次 Body::new 同一个 body——原样传进来的 Body 不会被重复装箱；但中间件若先套一层别的 wrapper（比如 with_limited_body 的 Body::new(Limited::new(b, limit))），那一层照样要装箱一次。

Body::empty 的实现

axum-core/src/body.rs:52-54：

pub fn empty() -> Self {
    Self::new(http_body_util::Empty::new())
}

用 http_body_util::Empty——一个 ZST，poll_frame 永远返 None。Body::empty() 是零大小、零分配的空 body——适合作为"响应无 body"的默认（比如 204 No Content）。

From 各种基础类型

axum-core/src/body.rs:93-111 的宏：

macro_rules! body_from_impl {
    ($ty:ty) => {
        impl From<$ty> for Body {
            fn from(buf: $ty) -> Self {
                Self::new(http_body_util::Full::from(buf))
            }
        }
    };
}

body_from_impl!(&'static [u8]);
body_from_impl!(std::borrow::Cow<'static, [u8]>);
body_from_impl!(Vec<u8>);
body_from_impl!(&'static str);
body_from_impl!(std::borrow::Cow<'static, str>);
body_from_impl!(String);
body_from_impl!(Bytes);

7 种基础类型自动 Into<Body>——http_body_util::Full 是"全内容一次性 body"的包装。让 Body::from("hello".to_string())、Body::from(vec![1, 2, 3]) 都直接工作。

注意：都是 Full——一次返回整个 body——不是流式。字符串和字节数组天然是内存里的、不需要流。Full 在 poll_frame 里一次返 Frame::data(整个 Bytes)、下一次返 None——两次 poll 就结束。

try_downcast 的类型系统技巧

try_downcast 是 Rust 类型系统的小把戏。原理：

pub(crate) fn try_downcast<T, K>(k: K) -> Result<T, K>
where T: 'static, K: Send + 'static,
{
    let mut k = Some(k);
    if let Some(k) = <dyn std::any::Any>::downcast_mut::<Option<T>>(&mut k) {
        Ok(k.take().unwrap())
    } else {
        Err(k.unwrap())
    }
}

几个微妙点：

一、Option<T> 作为桥梁：downcast_mut::<Option<T>>(&mut k) 要求 k 的类型是 Option<T>——类型完全匹配。把 K 先 wrap 成 Option<K> 再 downcast——如果 K == T，downcast 成功，可以从 &mut Option<T> 里 .take() 拿出值。

二、'static bound：Any::downcast_mut 要求类型是 'static——TypeId 不区分生命周期，非 'static 类型无法安全地 downcast。所以 Body 类型和所有相关的 body impl 都要求 'static——这也是 axum 所有地方 'static bound 如此普遍的原因。

三、零运行时类型信息：Rust 的 TypeId 是编译期确定的——运行时 downcast 只是比较 TypeId（一次整数相等检查）。成本极低。

这个 pattern 在 axum 里不止用一处——axum/src/util.rs:99 有一份同样的 try_downcast，Router::route_service 用它识别「传进来的服务本身就是 Router」并直接 panic、提示改用 nest（axum/src/routing/mod.rs:205）。在 Body::new 这里，目的是提供一个统一的 constructor API，但内部对已经是目标类型的情况做短路——避免不必要的包装。

Body::from_stream：流式构造

axum-core/src/body.rs:59-68：

pub fn from_stream<S>(stream: S) -> Self
where
    S: TryStream + Send + 'static,
    S::Ok: Into<Bytes>,
    S::Error: Into<BoxError>,
{
    Self::new(StreamBody {
        stream: SyncWrapper::new(stream),
    })
}

接收一个 TryStream——Stream<Item = Result<Ok, Err>>——Ok 能变成 Bytes、Err 能变成 BoxError。输出一个可以作为 http_body::Body 使用的流式 body。

SyncWrapper 的作用：sync_wrapper::SyncWrapper<T> 的关键性质是 unsafe impl<T> Sync for SyncWrapper<T> {}（sync_wrapper 1.0.2 的 lib.rs 第 167 行，对 T 不加任何约束）——无论 T 是什么，包装后都是 Sync。这条断言之所以安全，是因为 SyncWrapper 只提供 &mut self 形式的访问器（get_mut / get_pin_mut），拿不到 &T，也就不可能从多个线程同时读同一个 T。用户传进来的 Stream 未必是 Sync，包一层之后 StreamBody<S>（以及 SSE 那边的 SseBody<S>）无论 S 是什么都是 Sync——body 类型不会把用户 Stream 的 !Sync 传染出去。注意这是一层"保险"而不是硬性要求：Body::new 的 bound 只有 Send + 'static，UnsyncBoxBody 内部的 trait object 也只写了 + Send（BoxBody 才要求 + Send + Sync）。SyncWrapper 本身是零成本 newtype，所以加上不花钱。

StreamBody 的 http_body::Body impl

axum-core/src/body.rs:193-220：

pin_project! {
    struct StreamBody<S> {
        #[pin]
        stream: SyncWrapper<S>,
    }
}

impl<S> http_body::Body for StreamBody<S>
where
    S: TryStream, S::Ok: Into<Bytes>, S::Error: Into<BoxError>,
{
    type Data = Bytes;
    type Error = Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        let stream = self.project().stream.get_pin_mut();
        match ready!(stream.try_poll_next(cx)) {
            Some(Ok(chunk)) => Poll::Ready(Some(Ok(Frame::data(chunk.into())))),
            Some(Err(err)) => Poll::Ready(Some(Err(Error::new(err)))),
            None => Poll::Ready(None),
        }
    }
}

别把它和历史上的 axum::body::StreamBody 搞混：这里的 StreamBody 是 axum-core/src/body.rs 里的私有 struct，你在用户代码里引用不到，唯一入口就是 Body::from_stream。而曾经公开导出的 axum::body::StreamBody（axum 0.2.0 加入，见 axum/CHANGELOG.md 的 # 0.2.0 (23. August, 2021) 段与 PR #237）在 axum 0.7.0 随 http-body 1.0 迁移一并移除了——老代码里的 StreamBody::new(stream) 现在一律写成 Body::from_stream(stream)。

每次 hyper 调 poll_frame，StreamBody 就 poll 一次底层 Stream——把 Ok(chunk) 转成 Frame::data(chunk.into())、Err(err) 包成 axum_core::Error。

pin_project! { #[pin] stream }：需要 pin projection 让 self.stream 能作为 Pin<&mut S> 访问——try_poll_next 需要这个形式。

chunk.into()：S::Ok 只要求 Into<Bytes>——比如可以是 Vec<u8>、String、&'static [u8]——在这里统一转成 Bytes。

这是 axum 把"任意 Rust Stream"接入"HTTP body 协议"的具体 bridge。类似于第 10 章 SSE 的 SseBody——但 SseBody 需要把 Event 编码成 SSE 帧格式，StreamBody 只是透明转发 Bytes。

Body::into_data_stream：反向转换

axum-core/src/body.rs:76-78：

pub fn into_data_stream(self) -> BodyDataStream {
    BodyDataStream { inner: self }
}

BodyDataStream 的定义（axum-core/src/body.rs:141-143）加 Stream impl（axum-core/src/body.rs:145-168）：

pub struct BodyDataStream { inner: Body }

impl Stream for BodyDataStream {
    type Item = Result<Bytes, Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        loop {
            match ready!(Pin::new(&mut self.inner).poll_frame(cx)?) {
                Some(frame) => match frame.into_data() {
                    Ok(data) => return Poll::Ready(Some(Ok(data))),
                    Err(_frame) => {}  // 跳过非 data frame（比如 trailer）
                },
                None => return Poll::Ready(None),
            }
        }
    }
}

把 Body 当 Stream<Item = Result<Bytes, Error>> 用——每次 poll 从 body 拿一帧、如果是 data frame 返回 bytes、其他类型帧（trailer）丢弃。

用 loop 是因为可能连续几个 non-data frame——要 skip 到第一个 data frame 才 return。

实际场景：

async fn echo(request: Request) -> impl IntoResponse {
    let (parts, body) = request.into_parts();
    let stream = body.into_data_stream();
    // stream 是 Stream<Item = Result<Bytes, Error>>, 可以做各种 stream 操作
    let doubled = stream.map_ok(|bytes| {
        // 把每块数据大写
        Bytes::from(bytes.to_ascii_uppercase())
    });
    Response::new(Body::from_stream(doubled))
}

把请求 body 转 stream、每块数据转大写、重新包成响应 body。整个过程流式——不缓冲到内存——适合大 body。

from_stream 和 into_data_stream 是一对互逆操作——让 Body 在 http_body::Body 和 futures::Stream 两个生态间自由转换。

flowchart LR
    classDef axum fill:#efe,stroke:#080
    classDef stream fill:#fef,stroke:#808
    Stream["futures::Stream&lt;Item=Result&lt;Bytes,Err&gt;&gt;"]:::stream
    Body["axum::Body"]:::axum
    Stream -->|"Body::from_stream"| Body
    Body -->|"body.into_data_stream()"| Stream
    HYPER["hyper poll_frame"] -.读.-> Body
    USER["用户代码"] -.读.-> Stream

to_bytes：缓冲工具

流式 body 有时需要缓冲——比如签名验证、JSON 反序列化、文件写盘前的验证。axum 提供 to_bytes（axum/src/body/mod.rs）：

pub async fn to_bytes(body: Body, limit: usize) -> Result<Bytes, axum_core::Error> {
    Limited::new(body, limit)
        .collect()
        .await
        .map(|col| col.to_bytes())
        .map_err(axum_core::Error::new)
}

三行代码做三件事：

一、Limited::new(body, limit)：http_body_util::Limited 包装 body、限制总字节数。超过 limit 后 poll_frame 返 Err——不继续读。

二、.collect().await：http_body_util::BodyExt::collect 的方法——一次性读完 body、返回 http_body_util::Collected（一个 bytes + 可选 trailer 的容器）。

三、.to_bytes()：从 Collected 里拿 Bytes——所有 data frame 拼接成一块。

整个流程：流式 body → Limited wrap → 全部读完（受 limit 控制）→ 返回 Bytes。

limit 参数的重要性：to_bytes(body, usize::MAX) 是"不限"——接受客户端发送的任意大 body。生产里这是 DoS 漏洞——攻击者发个 10GB 的 body 耗尽内存。正确做法：

let bytes = to_bytes(body, 10 * 1024 * 1024).await?;  // 10MB 上限

到限失败后，to_bytes 返回的 axum_core::Error 里包着 http_body_util::LengthLimitError（识别方法见下文）——handler 据此返 413 Payload Too Large。

要澄清一个容易想当然的点：axum::body::to_bytes 是给用户用的便利函数，提取器内部并不调它。Bytes 的 FromRequest 走的是 req.into_limited_body().collect().await?.to_bytes()（axum-core/src/extract/request_parts.rs:100-116），BytesMut 走手写的 body_to_bytes_mut 逐帧 put，Multipart 走 req.with_limited_body().into_body()。Json / Form 则先 Bytes::from_request 再反序列化。两条路子形状一样（先加 limit、再收集），但限额来源不同：提取器用 DefaultBodyLimit，to_bytes 用你传的参数。

Limited 深入：防超大请求

http_body_util::Limited<B> 是一个 body wrapper——限制 body 总字节数。内部实现：

// http-body-1.0.1/http-body-util/src/limited.rs:15（节选；工作树版本 0.1.3）
pin_project! {
    pub struct Limited<B> {
        remaining: usize,   // 注意：是「还能读多少」，倒着减
        #[pin]
        inner: B,
    }
}

impl<B> Body for Limited<B>
where
    B: Body,
    B::Error: Into<Box<dyn Error + Send + Sync>>,
{
    type Data = B::Data;
    type Error = Box<dyn Error + Send + Sync>;   // 类型擦除，不是专用错误枚举

    fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>>
    {
        let this = self.project();
        let res = match this.inner.poll_frame(cx) {
            Poll::Pending => return Poll::Pending,
            Poll::Ready(None) => None,
            Poll::Ready(Some(Ok(frame))) => {
                if let Some(data) = frame.data_ref() {
                    if data.remaining() > *this.remaining {
                        *this.remaining = 0;
                        Some(Err(LengthLimitError.into()))
                    } else {
                        *this.remaining -= data.remaining();
                        Some(Ok(frame))
                    }
                } else {
                    Some(Ok(frame))   // trailer 帧不计入长度
                }
            }
            Poll::Ready(Some(Err(err))) => Some(Err(err.into())),
        };
        Poll::Ready(res)
    }
}

三个容易看漏的细节：

一、计数方向：字段是 remaining（剩余额度），每帧减去 data.remaining()，不是累加 consumed。 二、判定粒度是「整帧」：data.remaining() > *this.remaining 一旦成立，这一帧整个被拒绝——不会先吐出前面能装下的一部分再报错。所以超限时你拿到的是错误而不是截断数据。 三、错误类型是 Box<dyn Error + Send + Sync>：LengthLimitError 被装进这个 box，识别方式是 downcast（见下）。

每次 poll 检查一次额度——超限立即 Err——上层停止读、body 不会真的完成缓冲。这让"10GB 上传"在读到第一个越界帧时就被拒绝——而不是"等缓冲完 10GB 才发现超限"。

Limited 还改写了 size_hint：把上界收窄到 min(inner 上界, remaining)——下游因此能提前知道"这个 body 最多这么大"。

错误类型 LengthLimitError 能通过 downcast 识别：

match to_bytes(body, 1024).await {
    Ok(bytes) => { /* ... */ }
    Err(err) => {
        let source = std::error::Error::source(&err).unwrap();
        if source.is::<http_body_util::LengthLimitError>() {
            return (StatusCode::PAYLOAD_TOO_LARGE, "body too large").into_response();
        }
        // 其他错误
    }
}

这让 handler 能区分"body 过大"和"其他 body 错误"——分别返 413 和 500。

Limited 和 DefaultBodyLimit

第 6 章讨论过 DefaultBodyLimit——请求 body 的默认大小限制（axum-core/src/ext_traits/request.rs:319 写死 const DEFAULT_LIMIT: usize = 2_097_152，即 2 MB）。

这里有个常见误解要澄清：DefaultBodyLimit 这个 Layer 本身不 wrap body。它只往请求 extensions 里塞一个 DefaultBodyLimitKind（axum-core/src/extract/default_body_limit.rs:182-184 的 apply），而这个枚举只有 Disable 和 Limit(usize) 两个变体（axum-core/src/extract/default_body_limit.rs:83-86）。

真正的 Limited wrap 发生在提取的那一刻——RequestExt::with_limited_body（axum-core/src/ext_traits/request.rs:316-328）读这个 extension 决定用哪个额度：

    fn with_limited_body(self) -> Request {
        // update docs in `axum-core/src/extract/default_body_limit.rs` and
        // `axum/src/docs/extract.md` if this changes
        const DEFAULT_LIMIT: usize = 2_097_152; // 2 mb

        match self.extensions().get::<DefaultBodyLimitKind>().copied() {
            Some(DefaultBodyLimitKind::Disable) => self,
            Some(DefaultBodyLimitKind::Limit(limit)) => {
                self.map(|b| Body::new(http_body_util::Limited::new(b, limit)))
            }
            None => self.map(|b| Body::new(http_body_util::Limited::new(b, DEFAULT_LIMIT))),
        }
    }

这个设计的好处：限额是"延迟应用"的——只有真正要缓冲 body 的提取器才付出 Limited 这层包装，纯流式 handler（自己拿 Body 做 into_data_stream）完全不受影响。代价是：DefaultBodyLimit 只对走 with_limited_body 的提取器生效，你自己手写的 body 消费逻辑得自己限。

如果你在这样的提取器之后再手工调 to_bytes(body, n)，就是套了第二层 Limited——两层都生效，实际上限是更严格的那个。

http_body::Body trait 内部

axum::Body 的底层是 http_body::Body——HTTP body 的标准 trait（第三方 crate http-body）：

// 等价于 http_body crate 的定义（http-body-1.0.1/http-body/src/lib.rs:38-88）
pub trait Body {
    type Data: Buf;
    type Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>>;

    fn is_end_stream(&self) -> bool { false }
    fn size_hint(&self) -> SizeHint { SizeHint::default() }
}

Frame：数据帧或 trailer

Frame<T> 是 http_body 的基本单元——可以是 数据帧（一块 bytes）或 trailer（HTTP/2 的尾部 header）：

// http-body-1.0.1/http-body/src/frame.rs:9-17 —— Frame<T> 是 struct，内部包着这个私有 enum
enum Kind<T> {
    Data(T),
    Trailers(HeaderMap),
}

注意 Frame<T> 本身是 struct、Kind<T> 是它私有的字段类型，外部拿不到这个 enum——所以你不能对 Frame 直接 match，只能走 is_data() / into_data() / into_trailers()（http-body-1.0.1/http-body/src/frame.rs:50、58、94）以及 data_ref() / trailers_ref() 这类方法。

Data frame：body 的实际字节。HTTP/1.1 的 chunked encoding 里 chunk 的数据以 data frame 交出（一个大 chunk 可能被切成多帧，取决于 hyper 每次读到多少）；HTTP/2 的 DATA frame 也是。

Trailer frame：body 结束后额外发一批 header（比如 gRPC 的 status code）。HTTP/2 上最常用；HTTP/1.1 只有 chunked 编码能带 trailer。

http_body 统一两种形态——用户 poll 同一个方法、根据 Frame 类型分发。这个设计让 HTTP/1.1 和 HTTP/2 的 body 模型统一——上层代码不用分别处理。

axum::BodyDataStream 主动丢弃 trailer frame——poll_next 里 match frame.into_data() { Ok(data) => .., Err(_frame) => {} }，拿不到 data 就什么都不做、继续循环（axum-core/src/body.rs:145-158）——因为大多数 axum handler 不关心 trailers。想要 trailer 的场景（比如 gRPC）要用 http_body_util::BodyStream。

SizeHint：body 大小的预测

http_body::SizeHint 是 body 大小的（可能粗糙的）预估——lower: u64（最少多少字节）和 upper: Option<u64>（最多多少字节，None 表示未知）。

用途：

  • Content-Length header：如果 upper == lower、hyper 能自动加 Content-Length
  • 流量预估：客户端库（比如 reqwest）用 size_hint 判断是否能 stream
  • 内存预分配：调用方可以据此提前 reserve（注意 http_body_util::Collected 本身并不看 size_hint——它把各帧堆进一个 BufList，to_bytes() 时才按实际总长度做一次 copy_to_bytes；单帧时甚至是零拷贝直接返回）

http_body_util::Full<Bytes>（它属于 http-body-util、不是 axum 自己的类型）返回精确的 size_hint：SizeHint::with_exact(data.remaining())（http-body-1.0.1/http-body-util/src/full.rs:68-72）——让响应自动带 Content-Length。StreamBody 的 Body impl 根本没有重写 size_hint，落回 trait 默认的 SizeHint::default()（lower=0, upper=None）——因为 stream 的大小未知——响应用 chunked encoding 而不是 Content-Length。（别被 http-body-1.0.1/http-body-util/src/stream.rs:53 那个 fn size_hint(&self) -> (usize, Option<usize>) 骗了——那是 http-body-util 自己的同名公开类型 http_body_util::StreamBody 实现的 Stream::size_hint，和 axum 私有的 StreamBody 无关，返回类型也不一样，和 Body::size_hint 是两回事。）

is_end_stream：body 已经结束吗

is_end_stream(&self)——返回 true 表示"确定下一次 poll_frame 返 None"，默认实现返 false（http-body-1.0.1/http-body/src/lib.rs:77-79）。用途是让 hyper 不 poll 已知结束的 body——小优化。

Empty 的 is_end_stream 恒返 true——零次 poll 就知道结束（http-body-1.0.1/http-body-util/src/empty.rs:37-39）。Full 是 self.data.is_none()——内容发完后 data 被 take 成 None，才返 true（http-body-1.0.1/http-body-util/src/full.rs:64-66）。自定义 body 可以智能返回——精确实现让 hyper 行为更高效。

流式 vs 缓冲的具体差异

Body 既能缓冲也能流式——两种形态的性能和语义差异：

维度 缓冲（Full / Vec<u8>） 流式（StreamBody）
内存占用 O(body size) O(chunk size)
首字节延迟 等到 body 构造完 第一个 chunk 就绪就开始发
总耗时 = 构造 + 传输 = 构造和传输并行
Content-Length 精确 None（chunked）
适用场景 小 body（几 KB 到几 MB） 大 body（文件、日志流）
典型类型 Json、Html 响应 SSE、下载、proxy

内存优势：对 1GB 文件下载，流式版本 peak 内存只是一个 chunk（几 KB 到几 MB）——缓冲版本需要 1GB。

延迟优势：流式可以"边生成边发送"——客户端收到第一字节的时间减去服务端开始处理的时间更短（time-to-first-byte）。对 LLM streaming 这类 user-facing 场景尤其重要。

为什么默认缓冲：大多数 API 响应小（几 KB 的 JSON），缓冲更简单且有 Content-Length。只有明确是"大 body" 或 "需要实时推送"才用流式。

实战：三种流式 body 构造

实战一：文件下载（tokio File → Body）

use tokio::fs::File;
use tokio_util::io::ReaderStream;
use axum::body::Body;

async fn download(Path(name): Path<String>) -> Result<Response, AppError> {
    let file = File::open(format!("/data/{}", sanitize(&name))).await?;
    let stream = ReaderStream::new(file);
    let body = Body::from_stream(stream);
    Ok((
        [
            ("content-type", "application/octet-stream"),
            ("content-disposition", &format!("attachment; filename=\"{name}\"")),
        ],
        body,
    ).into_response())
}

核心链条：

  1. File::open：tokio 的异步文件句柄——实现 AsyncRead
  2. ReaderStream::new(file)：把 AsyncRead 转成 Stream<Item = io::Result<Bytes>>——tokio-util 提供
  3. Body::from_stream(stream)：包装成 axum body

hyper 每次 poll_frame 时从 stream 拉一块，stream 从 file 读一块——完整的流式链。1GB 文件下载的 peak 内存只有 ReaderStream 的 buffer（默认几 KB）。

实战二：proxy 转发（upstream body → downstream body）

async fn proxy(request: Request) -> Result<Response, AppError> {
    let client = reqwest::Client::new();
    let upstream_url = build_upstream_url(&request);

    // 转发请求到 upstream——body 流式传过去
    let (parts, body) = request.into_parts();
    let upstream_body = reqwest::Body::wrap_stream(body.into_data_stream());

    let resp = client
        .request(parts.method, &upstream_url)
        .headers(parts.headers)
        .body(upstream_body)
        .send()
        .await?;

    // 转发 response——body 流式回来
    let status = resp.status();
    let headers = resp.headers().clone();
    let downstream_body = Body::from_stream(resp.bytes_stream());

    Ok((status, headers, downstream_body).into_response())
}

两次 Stream ↔ Body 转换：

  1. 入站：axum Body → data stream → reqwest Body
  2. 出站：reqwest bytes_stream → axum Body from_stream

整条 proxy 路径完全流式——两边都不缓冲。1GB 文件经过 proxy 只占几 MB 内存（两边的 chunk buffer）。

示例为了突出流式，把请求和响应的 headers 原样转发；生产代理还应剔除 connection、transfer-encoding 这类逐跳（hop-by-hop）头，并按上游改写 host。

实战三：定时事件源（interval → SSE → Body）

use std::time::Duration;
use tokio_stream::wrappers::IntervalStream;
use tokio_stream::StreamExt;

async fn heartbeat_stream() -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let interval = tokio::time::interval(Duration::from_secs(1));
    let stream = IntervalStream::new(interval).map(|_| {
        Ok(Event::default()
            .event("heartbeat")
            .data(chrono::Utc::now().to_rfc3339()))
    });

    Sse::new(stream).keep_alive(KeepAlive::default())
}

Stream 来源是 IntervalStream (tokio-stream crate)——每秒 tick 一次。map 转成 Event。Sse::new 把 Event stream 包装成 SSE 格式的 Body。

这展示了 axum 流式生态的组合力：任何 Stream 都能作为 Body 源——定时、broadcast channel、数据库游标、LLM token 流——只要能产生 Bytes（或能转成 Bytes 的事件类型）。

实战四：LLM token 流式输出

LLM 应用最经典的流式场景——模型一个 token 一个 token 输出、前端一个一个展示：

use axum::body::Body;
use futures::StreamExt;

async fn chat(Json(req): Json<ChatRequest>) -> Result<Response, AppError> {
    let llm_stream = llm_client.stream_completion(req.messages).await?;

    // llm_stream: Stream<Item = Result<Token, LlmError>>
    let sse_stream = llm_stream.map(|token_result| {
        let data = match token_result {
            Ok(token) => format!("data: {}\n\n", serde_json::to_string(&token).unwrap()),
            Err(e) => format!("event: error\ndata: {}\n\n", e),
        };
        Ok::<_, std::convert::Infallible>(data.into_bytes())
    });

    Ok((
        [
            ("content-type", "text/event-stream"),
            ("cache-control", "no-cache"),
            ("x-accel-buffering", "no"),  // 禁用 Nginx buffering
        ],
        Body::from_stream(sse_stream),
    ).into_response())
}

几个生产细节：

一、x-accel-buffering: no：Nginx 默认对响应做 proxy buffering——SSE 流会被 Nginx 缓冲等完再一次性发给客户端——失去流式意义。这个 header 告诉 Nginx 不要缓冲。Cloudflare 则会把这个 header 从回给访客的响应里去掉（见 Cloudflare 文档 HTTP headers 一页），它自己的响应缓冲要在 Configuration Rules 里关。

二、直接用 Body::from_stream 而不是 Sse::new：第 10 章的 Sse<S> helper 专门做 SSE 语义（Event + KeepAlive）——但 LLM 流通常自己格式化（OpenAI 格式 data: {...}\n\n），Body::from_stream 更直接。

三、Error 的 Infallible：stream 内部 map 已经把 llm error 转成 SSE event: error 字节——再无其他错误。用 Infallible 告诉编译器这个 stream 永不真正失败。

四、token JSON 格式：沿用了 OpenAI API 那种 data: {json}\n\n 的 SSE 行格式；想让 openai 的客户端 SDK 直接解析，Token 序列化出的 JSON 结构也得和 OpenAI 的流式 chunk 对象一致。

这个模式是每个 LLM 应用的基础——从 stream 来源（模型 backend）到 stream 响应（前端 SSE）的直接链。流式性让用户感觉到"模型在工作"——首 token 延迟比等完整响应好太多。

实战五：chunked upload 与部分下载

大文件上传时用 multipart/form-data（第 8 章讨论过）是一种选——另一种是chunked transfer encoding——客户端分块发 body，axum 流式接收：

use futures::StreamExt;
use tokio::io::AsyncWriteExt;

async fn upload(request: Request) -> Result<Response, AppError> {
    let (_, body) = request.into_parts();
    let mut stream = body.into_data_stream();

    let path = format!("/uploads/{}", uuid::Uuid::new_v4());
    let mut file = tokio::fs::File::create(&path).await?;
    let mut total = 0usize;

    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        total += chunk.len();
        if total > 100 * 1024 * 1024 {  // 100MB 上限
            let _ = tokio::fs::remove_file(&path).await;
            return Err(AppError::PayloadTooLarge);
        }
        file.write_all(&chunk).await?;
    }

    file.flush().await?;
    Ok(Json(json!({"path": path, "size": total})).into_response())
}

边读边写——内存只占单个 chunk 的大小。100MB 文件的 peak 内存是单个 chunk（HTTP/1 下由 hyper 的读缓冲决定：从 8 KB 起按需增长，默认上限约 408 KB，即 hyper 的 INIT_BUFFER_SIZE 与 DEFAULT_MAX_BUFFER_SIZE）+ file OS cache——可控。

对应的断点续传（客户端用 Range header 请求部分）：

use tokio::io::{AsyncReadExt, AsyncSeekExt};

async fn download_with_range(
    Path(name): Path<String>,
    headers: HeaderMap,
) -> Result<Response, AppError> {
    let path = format!("/files/{}", sanitize(&name));
    let file = File::open(&path).await?;
    let metadata = file.metadata().await?;
    let size = metadata.len();

    // parse Range: bytes=X-Y
    let range = headers.get("range").and_then(|v| parse_range(v.to_str().ok()?, size));

    match range {
        Some((start, end)) => {
            let mut file = file;
            file.seek(std::io::SeekFrom::Start(start)).await?;
            let limited = file.take(end - start + 1);  // 只读部分
            let stream = ReaderStream::new(limited);
            Ok((
                StatusCode::PARTIAL_CONTENT,
                [
                    ("content-range", format!("bytes {}-{}/{}", start, end, size)),
                    ("content-length", (end - start + 1).to_string()),
                ],
                Body::from_stream(stream),
            ).into_response())
        }
        None => Ok((
            [("content-length", size.to_string())],
            Body::from_stream(ReaderStream::new(file)),
        ).into_response()),
    }
}

客户端挂机重启时用 Range header 续传——节省网络。这对大视频流或大文件下载场景必要。

CompressionLayer 与 body 的互动

tower-http::compression::CompressionLayer 自动给响应 body 压缩（gzip / brotli / deflate）——透明、用户不用改 handler：

let app = Router::new()
    .route("/", get(handler))
    .layer(CompressionLayer::new());

CompressionLayer 内部的工作：

  1. 检查客户端的 Accept-Encoding header（默认 predicate 还会跳过 32 字节以下、图片、gRPC 和 text/event-stream 响应，见 tower-http-0.6.8/tower-http/src/compression/predicate.rs:121-130）
  2. 包装响应 body——把每个 chunk 经过压缩 encoder
  3. 设 Content-Encoding: gzip、删除 Content-Length（压缩后长度变了）

和 axum::Body 的互动：body 本身不参与压缩——CompressionLayer 用 body 的 poll_frame 读原始 chunk、压缩、写进它自己的 stream body。对 axum::Body 来说完全透明——它不关心数据是否被压缩。

这层解耦是 http_body trait 设计的价值——body 只管"产出字节"、压缩只管"变换字节"——各自独立、组合工作。想自己写 encryption layer、encoding adapter 都能按相同思路做。

Frame 的 trailer 细节

HTTP/2 的 trailer 是body 之后的额外 header——gRPC 用它传 status code、metadata。轮廓：

HTTP/2 DATA frame (body chunk)
HTTP/2 DATA frame (body chunk)
...
HTTP/2 HEADERS frame (trailer, end_stream=true)

axum 的 Body 能产出 trailer——但 handler 的典型用法不涉及：

use http_body::{Body, Frame};
use http::HeaderMap;

struct MyBody { /* ... */ }

impl Body for MyBody {
    // ...
    fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Option<Result<Frame<Bytes>, Error>>>
    {
        if self.trailers_sent {
            return Poll::Ready(None);  // trailer 之后结束
        }
        if self.done_sending_data() {
            // 发一个 trailer
            self.trailers_sent = true;
            let mut trailers = HeaderMap::new();
            trailers.insert("x-status", "ok".parse().unwrap());
            Poll::Ready(Some(Ok(Frame::trailers(trailers))))
        } else {
            // 发 data
            Poll::Ready(Some(Ok(Frame::data(self.next_chunk()))))
        }
    }
}

自定义 Body 产出 Frame::trailers(...)——hyper 会在 HTTP/2 上正确编码。BodyDataStream 过滤掉 trailer——into_data_stream 的 user 不用看到。想要 trailer 用 http_body_util::BodyStream。

生产里 axum handler 几乎不产出 trailer——只有 gRPC（tonic）需要。tonic 在 axum 上的集成依赖这个能力——把 gRPC status 通过 trailer 传出去。

HttpBody 和 axum Body 的互操作

axum::body::HttpBody 直接 re-export http_body::Body——axum 用户不需要单独 import http-body crate。但底层类型互相兼容：

use axum::body::{Body, HttpBody};

// 任何 http_body::Body 都能变成 axum Body
fn from_any_body<B>(b: B) -> Body
where B: HttpBody<Data = Bytes> + Send + 'static, B::Error: Into<BoxError>,
{
    Body::new(b)
}

// 反过来——axum Body 本身实现了 HttpBody
fn body_is_http_body(body: Body) -> impl HttpBody<Data = Bytes> {
    body  // axum::Body 实现了 http_body::Body
}

这种"双向兼容"让 axum 能接收任何 http_body::Body 实现——hyper 的 Incoming、tonic 的 gRPC body、自定义 body 都能塞进 Body::new。反过来 axum Body 也能作为 http_body 交给下游（hyper、reqwest）。

深入：Body 的 pin_project 结构

Body 本身是一个 newtype：

pub struct Body(BoxBody);

没有 pin_project——UnsyncBoxBody 内部处理 pin。但 axum 用 Pin::new(&mut self.0) 直接 pin Box 的 inner（axum-core/src/body.rs:113-134）：

impl http_body::Body for Body {
    type Data = Bytes;
    type Error = Error;

    #[inline]
    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        Pin::new(&mut self.0).poll_frame(cx)
    }

    // …（后面还有 size_hint 与 is_end_stream，都是直接转发给 self.0，与本节论点无关，略去）
}

BoxBody 内部是 Pin<Box<dyn Body>>——pin 已经通过 Box 实现了。Pin::new(&mut self.0) 直接拿 Pin<&mut BoxBody>——因为 BoxBody 是 Unpin（它内部的 Pin<Box> 是 Unpin）。这让代码不需要 pin_project——简洁。

同样的技巧用在 BodyDataStream：内部的 Body 是 Unpin、所以外层不需要 pin_project。只有 StreamBody 的 Stream（可能 !Unpin）需要 pin_project。

SizeHint 深入

SizeHint 是 body 对"我有多少字节"的预估——http_body::SizeHint 有 lower: u64 和 upper: Option<u64>。几种典型情况：

Body 类型 lower upper 含义
Empty 0 Some(0) 精确空
Full<Bytes> n Some(n) 精确 n 字节
StreamBody<S> 0 None 未知——任意大
Limited<B> inner.lower（若 inner.lower ≥ 剩余额度则收成精确值） min(inner.upper, 剩余额度) 受限上界
自定义 counting body n None 至少 n 字节

SizeHint 影响：

HTTP/1.1 的 Content-Length 或 chunked：精确 size 用 Content-Length、不精确用 chunked transfer encoding。chunked 每块前加长度字节、略有开销；Content-Length 更高效但要求预知。

下游的容量决策：调用方可以据 upper hint 决定是否值得缓冲、要 reserve 多少。要注意 to_bytes 自己并不看 SizeHint——http_body_util::Collected 把各帧攒进 BufList，to_bytes() 时按实际总长度做一次 copy_to_bytes（只有一帧时直接返回那一帧，零拷贝）。

内存压力估计：客户端库可以用 upper 判断是否值得缓冲——太大的响应选择流式处理而非 to_bytes。

自己写 Body impl 时，尽量提供准确的 SizeHint——让 hyper 和上层代码做出更好的 I/O 决策。

Body 的 FAQ

Q：handler 里先 to_bytes 再手动 parse JSON 对不对？

不对——应该用 Json<T> 提取器。to_bytes + 手动 parse 失去了 Json 提取器的 Content-Type 校验、错误分类（422 vs 400）、body limit 等。只有在"需要自定义行为"时才绕过提取器。

Q：Body::from_stream 的 stream 已 exhausted 后再 poll 会怎样？

poll_frame 返 None。如果再继续 poll，行为取决于 Stream 实现——大多数情况下持续返 None。不应该依赖这个——FusedStream 明确定义 exhausted 后的行为。

Q：Body 能 clone 吗？

不能——Body 的 UnsyncBoxBody 不是 Clone（body 可能是一次性的 stream）。想"body 复用"需要先 to_bytes 缓冲成 Bytes、然后 Bytes 能 clone（Arc 语义）、再各自 Body::from(bytes.clone())。

Q：Body 支持 Seek 吗？

不支持。http_body::Body 是 forward-only 的——只能按顺序 poll_frame。想要 seek（比如断点续传）要在 body 外层处理——比如文件层用 file.seek 再包 body。

Q：返回超大 body 会导致什么？

hyper 按 poll_frame 速率发——backpressure 自动工作。客户端慢、TCP buffer 满——hyper 停止 poll body——body 停止生成。内存不会爆。唯一需要注意是 body 的生成本身不要提前"产出一大堆待发"——应该按 poll 节奏生成。

Q：Body 在 HTTP/2 上有什么不同？

HTTP/2 的 body 经过 flow control——客户端宣告 window size、服务端发到上限就停。这在 hyper 层处理、对 axum::Body 完全透明。body 该怎么写就怎么写——HTTP/1.1 和 HTTP/2 行为一致。

body 在 axum 内部流转

HTTP 请求从 TCP 到 handler、响应从 handler 到 TCP 的完整 body 流转：

sequenceDiagram
    participant Client
    participant Hyper
    participant Layer as Tower Layer 栈
    participant Handler
    Client->>Hyper: HTTP request with body (chunked/content-length)
    Hyper->>Hyper: 解析 HTTP、构造 Request<Incoming>
    Note over Hyper: body 类型是 hyper::body::Incoming
    Hyper->>Layer: req.map(Body::new)  <br/>(axum/src/serve/mod.rs:594)
    Note over Layer: body 变成 axum::Body<br/>(类型擦除 + 可流式 poll_frame)
    Layer->>Handler: Service::call(Request<Body>)
    alt handler 全缓冲处理
        Handler->>Handler: to_bytes(body, limit).await
        Handler->>Handler: 业务逻辑 on Bytes
        Handler->>Layer: Response<Body>
    else handler 流式处理
        Handler->>Handler: body.into_data_stream().map(...)
        Handler->>Layer: Response<Body::from_stream(...)>
    end
    Layer->>Hyper: Response<Body>
    Hyper->>Hyper: 根据 SizeHint 决定 Content-Length 或 chunked
    Hyper->>Client: 响应 body (chunked/content-length)

几个关键点：

  1. hyper::Incoming → axum::Body 在 axum/src/serve/mod.rs:594 的 .map_request(|req: Request<Incoming>| req.map(Body::new))——一次性类型统一
  2. handler 有两条路径：to_bytes 缓冲 + 流式 map
  3. hyper 根据 SizeHint 选编码：精确 size → Content-Length；未知 → chunked transfer encoding

整条路径保持类型一致（都是 axum::Body），但实际数据流可以是全缓冲或全流式——取决于 body 内部的具体实现。

Content-Length 与 Transfer-Encoding 的交互

body 在 HTTP wire 上有两种编码：

Content-Length：头里声明 body 长度——客户端读 N 字节就知道结束。要求提前知道长度。

Transfer-Encoding: chunked：body 按 chunk 发送——每 chunk 前是长度字节。不需要提前知道总长度。

axum body 的 SizeHint 决定用哪种——hyper 读 SizeHint：

  • 精确（upper == Some(n) 且 lower == n）→ Content-Length: n
  • 不精确 → Transfer-Encoding: chunked

手动覆盖：

// 强制 Content-Length（如果精确）
async fn explicit_length() -> impl IntoResponse {
    let bytes: Bytes = generate_data().await;
    let len = bytes.len();
    ([("content-length", len.to_string())], Body::from(bytes))
}

// 强制 chunked（流式）
async fn chunked() -> impl IntoResponse {
    let stream = generate_stream();
    Body::from_stream(stream)  // SizeHint 是 None, hyper 自动用 chunked
}

客户端偏好：现代 HTTP client 都支持 chunked——但有些极老系统（嵌入式、L4 proxy）偏好 Content-Length。实际看场景选。

HTTP/2 没有这个区分：HTTP/2 全部用 DATA frame——Content-Length 作为 header 传但不影响传输机制。上面的讨论主要是 HTTP/1.1。

Body 的生命周期

一个请求响应周期里 body 的状态变化：

stateDiagram-v2
    [*] --> Received: hyper 解析字节流入 buffer
    Received --> AxumBody: Body new wrap
    AxumBody --> HandlerRead: handler 读 body
    HandlerRead --> Buffered: to_bytes 全内存
    HandlerRead --> Streamed: into_data_stream 流式
    Buffered --> HandlerDone: 处理完
    Streamed --> HandlerDone: 处理完
    HandlerDone --> ResponseBody: 返回 Body
    ResponseBody --> Sent: hyper poll_frame 写 TCP
    Sent --> [*]
    note right of AxumBody
        Body 是可消费的——只能读一次
    end note
    note right of Streamed
        背压: hyper 慢时 stream 也慢
    end note

关键 state transitions：

  • Received → AxumBody：第 15 章讲的 req.map(Body::new)——一次类型转换
  • HandlerRead → Buffered：to_bytes + 处理 bytes——常见简单路径
  • HandlerRead → Streamed：into_data_stream + map/filter——高级流式路径
  • ResponseBody → Sent：hyper 的 connection task poll_frame 一个一个 chunk 写 TCP

整个过程body 永远是一次消费——不能读完了回到 Received 重读。这条约束深植于 HTTP 协议本身（第 6 章深入讨论过）。

实战六：tar 流式打包

把多个文件打包成 tar 流式下载——不缓冲整个 tar 到内存：

use axum::body::Body;
use futures::Stream;
use tokio::io::AsyncReadExt;

async fn download_tar(Path(dir): Path<String>) -> Result<Response, AppError> {
    let entries = list_files(&dir).await?;

    Ok((
        [("content-type", "application/x-tar"),
         ("content-disposition", "attachment; filename=\"archive.tar\"")],
        Body::from_stream(tar_stream(entries)),
    ).into_response())
}

// 构造 async stream: 对每个文件产出 tar header + content
// try_stream! 里才能用 `?`；错误类型靠返回类型标注推断
fn tar_stream(entries: Vec<Entry>) -> impl Stream<Item = std::io::Result<Bytes>> {
    async_stream::try_stream! {
        let mut tar = TarWriter::new(Vec::new());

        for entry in entries {
            let path = entry.path;
            let metadata = tokio::fs::metadata(&path).await?;

            // 先 yield tar header
            let header = tar.write_header(&path, metadata.len()).await?;
            yield Bytes::from(header);

            // 再 yield 文件内容
            let mut file = tokio::fs::File::open(&path).await?;
            let mut buf = vec![0u8; 64 * 1024];
            loop {
                let n = file.read(&mut buf).await?;
                if n == 0 { break; }
                yield Bytes::copy_from_slice(&buf[..n]);
            }

            // tar 要求每个文件的内容补零到 512 字节的整数倍
            let pad = (512 - metadata.len() % 512) % 512;
            yield Bytes::from(vec![0u8; pad as usize]);
        }

        // tar 结束 marker
        yield Bytes::from(tar.finish());
    }
}

核心：用 async_stream::try_stream! 宏构造一个 generator-like 的 stream——每次 yield 一块 bytes（TarWriter、list_files 是示意用的辅助类型和函数）。注意要用 try_stream! 而不是 stream!：后者的块里不能用 ?。文件内容按 64KB 块流式读、写——总内存占用恒定、几 GB 的 tar 都能流。

这种 pattern 对"按需聚合"的下载场景（zip / tar / 合并多个响应）很有用——所有都遵循 "generate → bytes chunk → yield → body" 的套路。

body 的 DoS 防线

body 处理是很多 DoS 攻击的目标——送大 body 耗尽内存、慢速 body 占用连接、分块 body 里藏非法数据。axum 的防御：

一、DefaultBodyLimit：默认 2MB 请求 body 限制。所有 Bytes::from_request、Json::from_request 等都经过这层——超过直接 413。

二、to_bytes 显式 limit：自己调 to_bytes 时第二参数要写合理值——不用 usize::MAX。

三、hyper 的 header_read_timeout：第 15 章提到的、axum::serve 默认启用的 header 读取超时（builder.http1().timer(TokioTimer::new())，#3478）。注意这是主线上的改动，CHANGELOG 里还在 Unreleased 段，未包含在 crates.io 的 0.8.9 里——用 0.8.9 的话这道防线得自己搭。注意它防的是慢速发 header 那一类 slowloris，并不限制 body 的读取时长——慢速发 body 得靠 tower_http::timeout 之类的超时中间件或反向代理去兜。

四、request body limit layer (tower-http)：全局限制 request body——比 DefaultBodyLimit 更 strict（不依赖提取器触发）。

五、Content-Length 验证：hyper 会校验 Content-Length 的 body 不超过头声明的大小。

这五层配合能挡住常见的 body-level DoS（慢速发 body 仍要另加超时，见第三条）。生产的 checklist：

  • 配合前置 LB（nginx / ALB）做第一层 body size 限制（硬上限按业务定）
  • axum 层用 DefaultBodyLimit 或 RequestBodyLimitLayer 做第二层（按业务上限）
  • handler 里 to_bytes 调用带合理 limit
  • 监控 413 响应数——异常增加可能是攻击

body 的性能维度

Body 的各种形态有不同的性能 profile。几个常见场景的量化：

小 JSON 响应（几 KB）：

  • Json(data).into_response() → Body::from(bytes) → Full<Bytes>
  • 开销：serde_json 序列化 + 一次 Full<Bytes> 包装（Body::from 不拷贝字节，只为类型擦除装箱一次）
  • bottleneck: 序列化本身

大文件下载（100 MB）：

  • Body::from_stream(ReaderStream::new(file))
  • 开销：每次 poll 从 file 读一块（ReaderStream 默认容量 4096 字节，见 tokio-util reader_stream.rs 的 DEFAULT_CAPACITY）、转成 Bytes、返回
  • bottleneck: 磁盘 I/O，内存 = 单 chunk 大小

SSE 推送：

  • Body::from_stream(event_stream)
  • 开销：每次 event 一次 poll、格式化成 bytes、返回
  • bottleneck: event 生成速率

proxy 转发：

  • request.body.into_data_stream() → reqwest body → upstream
  • 开销：两次 data_stream ↔ body 转换（零拷贝）、网络 I/O
  • bottleneck: 上游延迟

关键观察：body 本身的开销通常不是 bottleneck——要么是序列化、要么是 I/O。优化方向应该针对这些——Body 层只要"不乱加开销"就够。axum 的 body 设计确实做到了：每帧只多一次 UnsyncBoxBody 的动态派发，from_stream 除了把 Stream move 进一个 Box 里（类型擦除那一次）不再额外分配，SyncWrapper 和 StreamBody 本身都是零成本 newtype。

body 的零拷贝路径

Bytes 的核心特性是零拷贝共享——底层 Arc<[u8]> 或类似结构，clone 只是引用计数加一。Body 内部的 data frame 都是 Bytes——意味着：

  1. hyper 从网络收到字节进 Bytes buffer——不拷贝
  2. Bytes 经过提取器、handler、响应 body——只传引用
  3. hyper 把 Bytes 写到网络 socket——不拷贝

整条路径基本零拷贝。例外是需要一块连续内存的地方：Bytes 提取器和 to_bytes 收到多帧时，Collected::to_bytes() 要拼接一次（见前文 SizeHint 一节）；此外只有在用户代码里 .to_vec()、copy_from_slice 这类操作时才真正复制字节（Bytes::clone() 只加引用计数）。这让大 body 的处理 CPU 成本低——瓶颈在网络和业务逻辑、不在 byte shuffling。

proxy body 的精细控制

前面的 proxy 例子是"流式转发"——还有更精细的控制场景。

场景：过滤 / 修改 body

proxy 时想对 body 做修改（比如加请求签名、改 URL references）——需要读完修改再转发：

async fn signed_proxy(request: Request) -> Result<Response, AppError> {
    let (parts, body) = request.into_parts();

    // 缓冲 body (受 limit)
    let bytes = to_bytes(body, 10 * 1024 * 1024).await?;

    // 加签名
    let signature = hmac_sign(&bytes, &SIGNING_KEY);
    let mut headers = parts.headers.clone();
    headers.insert("x-signature", signature.parse().unwrap());

    // 转发
    let resp = client.post(upstream)
        .headers(headers)
        .body(bytes)
        .send()
        .await?;

    // 响应流式回来 (这部分还是流式)
    Ok(axum_response_from_reqwest(resp))
}

这种 pattern 的 tradeoff：

  • 收益：能对 body 做任何处理
  • 成本：request body 全缓冲——1GB upload 要 1GB 内存

生产上 hybrid：对 small body（<1MB）缓冲再修改、对 large body（>1MB）流式不修改（或者用 streaming-capable 加密库边流边签）。具体策略看业务——axum 的 body API 足够灵活支持各种组合。

场景：response body 的条件修改

某些场景想根据 response 的内容决定是否透传——比如只 proxy 成功响应、失败替换为自定义错误页：

async fn conditional_proxy(request: Request) -> Result<Response, AppError> {
    let upstream_resp = client.send(request).await?;
    let status = upstream_resp.status();

    if status.is_success() {
        // 成功——流式转发
        let body = Body::from_stream(upstream_resp.bytes_stream());
        Ok(Response::builder().status(status).body(body).unwrap())
    } else {
        // 失败——用自定义错误替换
        let err_body = upstream_resp.text().await.unwrap_or_default();
        tracing::warn!(status = %status, body = %err_body, "upstream error");
        Ok(Response::builder()
            .status(502)
            .body(Body::from("upstream unavailable"))
            .unwrap())
    }
}

upstream 成功路径流式、失败路径缓冲 error body 后生成新 response——按需选择策略。

Body 的 StreamBody 内部细节

StreamBody 的 pin_project + SyncWrapper 组合值得深入：

pin_project! {
    struct StreamBody<S> {
        #[pin]
        stream: SyncWrapper<S>,
    }
}

为什么用 SyncWrapper？因为底层 Stream S 可能不是 Sync——但 http_body::Body 用在 UnsyncBoxBody 里不强制 Sync。SyncWrapper 的作用是标注类型——让包了它的 struct 是 Sync——即便 inner S 不是。

SyncWrapper 是怎么做到安全的？通过 API 设计：

  • 构造：SyncWrapper::new(s) - 任何 s 都能包
  • 访问：必须通过 &mut self 才能拿到 inner &mut S——不能通过 &self（单线程访问保证）

共享是对 &self 的——SyncWrapper 阻断这条路径——所以即便 S 非 Sync 也安全。代价是有 &mut self 的地方才能用——对 http_body 的 poll_frame(Pin<&mut self>) 签名完美兼容（&mut self）。

这是一种"用 API 限制换 trait bound 放松"的典型 Rust 技巧——牺牲一点灵活（必须 mut 访问）、获得范围更大的类型兼容性。

跨专栏关联：http_body 规范

http_body crate 是 Rust HTTP 生态的标准 body trait——hyper、axum、tonic、reqwest 都基于它（actix-web 走的是自己的 MessageBody，不在这个体系里）。定义简洁但经过几次演进：

  • http_body 0.x：两个方法——poll_data 和 poll_trailers 分别返回 data 和 trailer
  • http_body 1.0：合并成 poll_frame 返回 Frame<T>——统一接口。axum 0.7+ 用 1.0 版本

axum 紧跟 http_body 的最新版本——因为 hyper 1.0 也迁移到了这版。整个生态协同演进——但偶尔有版本兼容性问题（老项目用 http-body 0.4 会和新 crate 冲突）。

《Hyper 与 Tower：工业级 HTTP 栈》专栏第 10 章专门讨论 http_body——从 trait 设计到常见实现（Full、Empty、Limited、Combinators）。读那一章后再看 axum 的 Body，会发现 axum 只是在 http_body 上做最薄的 wrapper——UnsyncBoxBody 类型擦除 + 几个便利 constructor。

测试 body 的常见 pattern

测试 handler 的响应 body 几种写法：

use axum::body::to_bytes;
use tower::ServiceExt;

#[tokio::test]
async fn returns_expected_json() {
    let app = Router::new().route("/data", get(data_handler));
    let response = app
        .oneshot(Request::builder().uri("/data").body(Body::empty()).unwrap())
        .await
        .unwrap();

    assert_eq!(response.status(), StatusCode::OK);

    // 把 body 转成 bytes
    let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
    let parsed: MyData = serde_json::from_slice(&body).unwrap();

    assert_eq!(parsed.field, expected_value);
}

几个要点：

  • to_bytes(body, usize::MAX)：测试里不怕 body 大、用 MAX
  • serde_json::from_slice：bytes 直接 parse，省一次 String 转换
  • oneshot：直接跑 Service 不启 TCP、更快

测试流式 body：

use futures::StreamExt;

#[tokio::test]
async fn streams_tokens() {
    let app = Router::new().route("/stream", get(stream_handler));
    let response = app
        .oneshot(Request::builder().uri("/stream").body(Body::empty()).unwrap())
        .await
        .unwrap();

    let mut stream = response.into_body().into_data_stream();

    // 逐块验证
    let chunk1 = stream.next().await.unwrap().unwrap();
    assert_eq!(&chunk1[..], b"event: start\n\n");

    let chunk2 = stream.next().await.unwrap().unwrap();
    assert!(chunk2.starts_with(b"data: "));

    // 最终结束
    assert!(stream.next().await.is_none());
}

对 SSE / 分块响应，验证每个 chunk 符合预期——比 to_bytes 一次性更精细。

与其他 body 类型的 conversion

实际项目会涉及多种 body：

  • hyper::body::Incoming：hyper 服务端收到的 request body（从网络 parse 出来）
  • reqwest::Body：reqwest 客户端用于发请求/接响应的 body
  • http_body_util::Full<Bytes>：全缓冲
  • http_body_util::Empty：空
  • axum::body::Body：axum 用的

它们的 conversion：

// hyper::Incoming → axum::Body
let axum_body = Body::new(incoming);  // Incoming 不是 Body, 会走一次 boxed()
// 已经是 axum::Body 时才被 try_downcast 短路
let same = Body::new(axum_body);      // 原样返回, 不再多包一层

// Vec<u8> → axum::Body
let axum_body = Body::from(vec);  // From impl, Full wrap

// Stream → axum::Body
let axum_body = Body::from_stream(stream);  // StreamBody wrap

// axum::Body → reqwest::Body (for proxy)
let reqwest_body = reqwest::Body::wrap_stream(axum_body.into_data_stream());

// reqwest::Response → axum::Body
let axum_body = Body::from_stream(response.bytes_stream());

每对转换的具体 API 都在 axum / reqwest / http_body_util 的文档里——规则都很直白。只要两边都遵守 http_body 的 trait、conversion 通常是一行。

http_body 的版本演进

http_body 作为整个 Rust HTTP 生态的基础 trait，经过几轮大改：

  • http_body 0.1（2019-05）：初版就是 poll_data / poll_trailers 两个方法加 SizeHint
  • http_body 0.2～0.4：trait 形状基本不变，主要是跟进 Pin（0.2.0-alpha.1）、http 0.2 与 bytes 0.5（0.2.0）、bytes 1.0（0.4.0）等依赖升级（见 http-body 的 CHANGELOG.md）
  • http_body 1.0（2023-11-15）：合并 poll_data/poll_trailers 成 poll_frame + Frame<T> 帧类型——当前稳定版

axum 的版本对应：

  • axum 0.5/0.6：用 http_body 0.4
  • axum 0.7+：用 http_body 1.0（和 hyper 1.0 同步）

这次重大升级（0.4 → 1.0）让 axum 0.7 有大量不兼容改动——老项目升级 axum 要同时更新 http-body 依赖。好处是整个生态用同一个版本、互操作不用 adapter。

版本演进的经验：HTTP 生态的底层 crate（hyper、http、http-body）升级会波及所有上层。axum 跟进得很快——hyper 1.0.0 发布于 2023-11-15（hyper CHANGELOG.md），axum 0.7.0 在 2023-11-27 就跟上了（axum CHANGELOG.md）。

本章的几个心得

回头看本章内容可以总结几个关键 insight：

一、Body 是类型擦除的艺术：UnsyncBoxBody<Bytes, Error> 让一个具体类型承载多种 body 形态——用户 API 简单、生态兼容性好

二、流式和缓冲的选择是语义而非技术：axum 两种都支持、API 对称。选哪个看业务需求——小 body 缓冲简单、大 body 流式必要

三、pin 和 SyncWrapper 是 Rust 异步 HTTP 的底层术语：pin_project 处理 !Unpin future 的 safe access、SyncWrapper 处理 !Sync 类型的 across-thread 兼容。两者都是为了让更多的 Rust 类型能接入 HTTP body 接口。

四、零拷贝通过 Bytes 实现：整条 body 处理路径 Bytes 共享引用计数——不做实际字节拷贝。I/O 路径的 CPU 开销极低。

五、http_body 是生态级的 trait：axum 的 Body 是 http_body 的"友好封装"——不替代、不分叉、完全兼容。这让 axum 的 body 处理能和 tonic、reqwest、hyper 等其他 HTTP crate 无缝互操作。

和第 10 章 SSE 的回顾

第 10 章讲 SSE 时提到 SseBody<S> 是自定义 http_body::Body 实现。现在可以完整理解它：

// 第 10 章 sse.rs:110-135 (简化)
pin_project! {
    struct SseBody<S> {
        #[pin]
        event_stream: SyncWrapper<S>,
    }
}

impl<S, E> HttpBody for SseBody<S>
where S: Stream<Item = Result<Event, E>>,
{
    type Data = Bytes;
    type Error = E;

    fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>>
    {
        // poll stream, 每个 Event finalize 成 Bytes 再 Frame::data
    }
}

和本章讨论的 StreamBody 结构一模一样——都是 SyncWrapper<S> + pin_project。差异在 SseBody 要把 Event 编码成 SSE 帧格式（data: ...\n\n）、StreamBody 直接用 Bytes。同一套基础结构支持不同协议——是 http_body 设计的强大之处。

小结：Body 的两个维度

Body 在 axum 里承担两个维度的抽象：

一、类型擦除：UnsyncBoxBody<Bytes, Error> 隐藏具体 body 类型——用户只看到 Body，不用管是 Full、Empty、StreamBody 还是 Incoming。try_downcast 优化让重复 wrap 零成本。

二、流式/缓冲统一：同一个 Body 类型可以内部是缓冲（Full）或流式（StreamBody）——上层代码（hyper、handler）用 poll_frame 统一 API——SizeHint 告知具体形态。

两个维度合起来让 body 处理既类型简单（就一个 Body 类型）又功能灵活（能覆盖从空响应到 TB 级流式响应）。这是 axum 简洁 API 下的强大能力——关键都在 trait 设计和类型擦除的巧妙应用。

Body FAQ 补充

Q：能在 response 里同时返回 body 和 trailer 吗？

可以——自定义 Body 在 poll_frame 里先 yield 多个 data frame、最后 yield 一个 trailer frame。hyper 会在 HTTP/2 上正确发送 trailer。HTTP/1.1 上条件苛刻：响应必须是 chunked、请求带了 TE: trailers，而且只发响应 Trailer 头里声明过的字段，否则 trailer 被静默丢掉（hyper-1.9.0/src/proto/h1/conn.rs:328-332、hyper-1.9.0/src/proto/h1/conn.rs:729-733、hyper-1.9.0/src/proto/h1/role.rs:969-973；0.8.9 实测）。大多数场景不建议依赖。

Q：hyper 看到 upper: Some(n) 会严格 enforce n 字节吗？

hyper 只在 lower == upper（size_hint().exact()）时才据此设 Content-Length，光有 upper 不会。一旦按 exact 设了 Content-Length，HTTP/1 编码器就只写这么多字节：body 多 yield 的部分被截断丢弃，不会 panic（hyper-1.9.0/src/proto/h1/encode.rs:143-153、hyper-1.9.0/src/proto/h1/dispatch.rs:364-371；0.8.9 实测声明 5 字节、实发 10 字节，客户端只收到前 5 字节）；少于声明值则以 body 写中断的错误收场。所以 body 实现要保证 SizeHint 准确，拿不准就别给精确值。

Q：BodyDataStream 丢弃 trailer——有没有 stream 全帧类型？

http_body_util::BodyStream 保留所有 frame 类型——适合需要处理 trailer 的场景（比如 gRPC 实现）。axum 的 BodyDataStream 是"简化版"——大多数 HTTP handler 用它够。

Q：Full<Bytes> 和 Full<BytesMut> 哪个好？

Full 泛型到 T: Buf，两者都能构成 http_body::Body；但 axum::body::Body::new 要求 Data = Bytes，Full<BytesMut> 不能直接塞进去。Bytes 是 Arc-based 零拷贝、BytesMut 是可修改版本。构造 body 用 Bytes；内部还要修改用 BytesMut，最后 BytesMut::freeze() 变 Bytes 进 body。

Q：Body 能在多个 handler 间共享吗？

不能——Body 不是 Clone。每个 Body 是单次消费的。想"共享" body 数据要先 to_bytes 缓冲成 Bytes——Bytes 能 clone（引用计数）、各 handler 各自 Body::from(bytes.clone()) 独立消费。

Q：body 的 chunk 大小能控制吗？

由生成侧决定——ReaderStream::new(file) 的默认读缓冲是 4096 字节（tokio-util 的 DEFAULT_CAPACITY）、ReaderStream::with_capacity(file, 64 * 1024) 改成 64KB。大 chunk 减少 poll 次数、小 chunk 减少 per-chunk 延迟。根据网络条件权衡——几 KB 到几 MB 都合理。

Q：body 生成失败了怎么办？

Body::from_stream 的 Stream 返回 Err 会让 body 的 poll_frame 返 Err——hyper 看到 Err，HTTP/1 上断开连接，HTTP/2 上 reset 这条流（hyper-1.9.0/src/proto/h2/mod.rs:200-212）。客户端看到连接中断、stream reset 或 "incomplete message"。如果 body 生成可能失败——应该尽早检查（在 body 构造前）、失败返回 5xx、成功后再流式发送。一旦开始流式发送后就很难"优雅失败"了——HTTP 没有"响应开始后改主意"的机制。

Q：能从 stream 得到 size hint 吗？

不能自动——Stream trait 的 size_hint 是 (lower, upper) 元素数、不是字节数。需要自己的 body 实现手动计算并返回 SizeHint。比如对 File 可以先查 metadata 得到大小、然后 body 的 SizeHint 返回 (size, Some(size))——让响应自动带 Content-Length。

一个容易忽略的细节：body 的 Drop 行为

当 handler 早早 return（不读完 body）、或者客户端中途断开连接——body 被 drop。drop 会发生什么？

对缓冲 body（Full / Bytes）：drop 只是释放内存——没副作用。

对流式 body（StreamBody<S>）：drop 时内部 Stream 被 drop——Stream 内部的资源（File handle、tokio::sync receiver、数据库游标）随之释放。Rust 的 RAII 保证这层自动清理。

对来自 hyper 的 Incoming body：drop 通知 hyper "client 的后续字节我不读了"。HTTP/1 上 hyper 把手头已到的数据读一轮丢掉，body 若因此读完，连接照常复用；若还没读完就关闭读端，这条连接在响应后不再复用（hyper-1.9.0/src/proto/h1/conn.rs:849-865；0.8.9 实测：声明 1 MB 只发 10 字节，响应后连接被关）。HTTP/2 上则按 RST_STREAM 终止该 stream。两种情况都不会卡住连接。

这意味着用户不需要手动处理 "handler 没读完 body 怎么办"——Drop 完全兜底。只需要确保业务逻辑正确——不要让 body 被读一半再继续用（那会 panic 或 error）。

这条 Drop 语义也让第 10 章讲的 SSE 连接清理能工作——客户端断开时 body 被 drop、SSE 的 Stream 被 drop、它里面的 broadcast receiver / 数据库查询都随之释放。Rust 的所有权链让"资源清理"成为自动动作——不用写 try-finally 之类的显式清理。

几个工程级最佳实践

最后给几个生产级的 body 处理建议：

一、默认流式、特殊情况缓冲。对文件下载、SSE、proxy 等天然流式场景——不要用 to_bytes 缓冲。只有业务明确需要"整体访问"才缓冲（比如 HMAC 签名、JSON 反序列化）——并且总带合理 limit。

二、to_bytes 的 limit 按接口设置。不同 handler 的 body 合理上限不同：/api/chat 可能 16KB 够、/upload 可能 100MB。不要全局一刀切——在每个 handler 的 to_bytes 里写具体上限。

三、监控 body 相关指标：HTTP 413 响应数（超限）、body 读取失败数、body 平均大小——这些指标能发现 DDoS 尝试、upstream 异常、业务字段膨胀。

四、流式响应加超时：生产的流式 handler（SSE / download）要设上限（比如 tokio::time::timeout 或空闲检测）——防止客户端挂起不消费耗 server 资源。时长按场景定：下载按文件大小估，SSE 这种长连接要按空闲时间而不是总时长来算。

五、测试覆盖两路径：handler 的单元测试至少覆盖"小 body 成功"和"超限失败"两条路径——真实客户端会发两种都——测试里别只测前者。

做到这五点，body 层基本不会成为 axum 服务的稳定性问题源。剩下的注意力可以放到业务逻辑、数据库、外部调用上——那些是真正的挑战。

下一章讲 State 管理与 FromRef 子状态——第 7 章简单介绍过 State 提取器、第 18 章会深入类型状态、FromRef 的实际架构价值、多 state 项目的组织模式。那一章把 axum 的"依赖注入"完整机制讲清楚。