当前位置：首页 > 文章列表 > 文章 > python教程 > FastAPI 实现 StreamingResponse 分块传输详解

FastAPI 实现 StreamingResponse 分块传输详解

2026-05-27 13:00:48 0浏览收藏

本文深入解析了 FastAPI 中 StreamingResponse 实现 HTTP 分块传输（chunked transfer encoding）的核心机制与实战要点，强调其默认即支持分块响应的关键前提——不设置 Content-Length 且 ASGI 服务器（如 Uvicorn）正常工作；同时直击生产环境常见陷阱：Nginx 缓冲导致流式失效、生成器 yield 频率与大小失衡、前端误用 response.text() 阻塞读取、中间件意外注入 Content-Length，以及超时配置不当引发连接中断，并辅以可验证代码示例和精准排查路径，助你真正掌握低延迟、高可控的实时数据流构建方法。

fastapi 如何实现 StreamingResponse 的 chunked transfer

StreamingResponse 默认就用 chunked transfer

只要你不设置 Content-Length，FastAPI 的 StreamingResponse 会自动启用 HTTP/1.1 chunked transfer encoding。这是它的默认行为，不需要额外配置 —— 但前提是底层 ASGI 服务器（如 Uvicorn）支持，并且你没手动干扰响应头。

Uvicorn 默认开启 chunked；Gunicorn + Uvicorn worker 也正常，但纯 Gunicorn（sync worker）不支持流式响应，会等整个生成器结束才发响应
如果用了 Nginx 反向代理，默认会缓冲响应体，导致 chunked 被吞掉、客户端收不到实时数据 —— 必须在 location 块里加 proxy_buffering off; 和 chunked_transfer_encoding on;
浏览器 fetch API 接收 chunked 响应时，response.body.getReader() 才能分块读取；用 response.text() 会等全部结束，失去流式意义

如何构造一个真正可分块的生成器

关键不是“怎么开启 chunked”，而是生成器 yield 的内容是否足够小、是否及时 flush。Uvicorn 每次收到一个 yield 就尝试发一个 chunk，但如果 yield 太大或太密集，可能被合并；如果 yield 太慢，TCP 包可能被延迟发送（Nagle 算法）。

每次 yield 最好控制在几 KB 内，例如 yield json.dumps(chunk).encode() + b"\n"
避免在生成器里做耗时同步操作（如 time.sleep、requests.get），改用 await asyncio.sleep()
如果需要强制刷新小 chunk，可在 yield 后加 await asyncio.sleep(0) 让出控制权，帮助 Uvicorn 尽快写出
不要在生成器里 raise 异常后继续 yield —— FastAPI 会关闭连接，后续 chunk 丢弃

常见 chunked 失效场景和排查方法

你以为在流，其实客户端看到的是全量响应，大概率是中间某层吃掉了 chunked。先看响应头：

用 curl -v 或浏览器 DevTools Network 面板确认响应头含 Transfer-Encoding: chunked，且没有 Content-Length
如果看到 Content-Length，说明你或中间件（如某些 CORS 中间件、gzip 中间件）提前计算了长度，禁用或绕过它们
如果响应头正确但前端收不到分块，抓包（tcpdump/wireshark）看 TCP 层是否真有多个小包；若只有 1 个大包，基本是 Nginx 或负载均衡器缓冲所致
Uvicorn 日志里出现 Connection lost before response written，常因客户端断连或超时，不是 chunked 问题，而是流产生太慢或客户端没保持连接

一个最小可验证的流式 endpoint 示例

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
async def event_stream():
for i in range(5):
await asyncio.sleep(1)  # 模拟异步等待
data = {"seq": i, "message": "hello"}
yield f"data: {json.dumps(data)}\n\n".encode()
加 sleep(0) 可提升 chunk 及时性，非必需但有时有用
    await asyncio.sleep(0)
@app.get("/stream")
async def stream():
return StreamingResponse(
event_stream(),
media_type="text/event-stream",  # 或 "application/json" 等
不要设 content_length！
)

注意：这里用 text/event-stream 是为了兼容浏览器 EventSource；若用 application/json，前端需用 fetch + ReadableStream 处理，且每个 chunk 应为独立 JSON 对象（不换行或用 \n 分隔）。

最易忽略的一点：流式响应的生命期依赖于生成器执行完或客户端断开。如果生成器卡在某个 await 上太久，Uvicorn 可能因 keep-alive 超时关闭连接，而这个超时通常不在 FastAPI 控制范围内，得去调 Uvicorn 的 --timeout-keep-alive 参数。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~