Axum 流式响应体输出

示例说明

生成完整 CSV 再返回会把大导出钉在内存与延迟上——客户端要等全部数据就绪才能收到第一个字节。Body::from_stream 把响应体本身变成流：服务端逐块产出，网络逐块传输，客户端逐块落盘，三段流水线并行。事件流端点则演示反向用法——把流收集成数组以 JSON 返回，同一数据源两种消费形态。

技术要点

• Body::from_stream 让响应体即数据流
• then 组合子做异步逐项变换
• filter_map 收集流为集合

代码解读

export_stream 用 stream::iter 驱动序号流，then 给每项叠加三十毫秒的异步延迟模拟真实查询，chain 追加收尾行。export_csv 把字符串块转字节流并声明 CSV 的内容类型与附件下载头；json_events 用 filter_map 过滤错误项后 collect 成数组，Json 包装直接序列化。

使用建议

流式响应要配上游取消——客户端断开时流应停止拉取。错误项映射到 Infallible 意味着该流不会失败，真实场景映射到运输层错误。超时控制加在流之外防止整体挂死。

代码示例

use axum::{
    body::Body,
    extract::State,
    http::{header, StatusCode},
    response::{IntoResponse, Response},
    routing::get,
    Json, Router,
};
use futures_util::stream::{self, StreamExt};
use std::convert::Infallible;
use std::time::Duration;

async fn export_stream() -> impl futures_util::Stream<Item = Result<String, Infallible>> {
    stream::iter(0..5)
        .then(|i| async move {
            tokio::time::sleep(Duration::from_millis(30)).await;
            Ok(format!("row-{i}\n"))
        })
        .chain(stream::iter(vec![Ok("done\n".to_string())]))
}

async fn export_csv() -> Response {
    let body_stream = export_stream().map(|chunk| {
        chunk
            .map(|text| text.into_bytes())
            .map_err(|err: Infallible| match err {})
    });

    let body = Body::from_stream(body_stream);

    (
        StatusCode::OK,
        [
            (header::CONTENT_TYPE, "text/csv; charset=utf-8"),
            (header::CONTENT_DISPOSITION, "attachment; filename=rows.csv"),
        ],
        body,
    )
        .into_response()
}

async fn json_events() -> Json<Vec<String>> {
    let events: Vec<String> = export_stream()
        .filter_map(|chunk| async move { chunk.ok() })
        .collect()
        .await;
    Json(events)
}

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/export", get(export_csv))
        .route("/events", get(json_events));

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}