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

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
<p>app = FastAPI()</p><p>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()</p><h1>加 sleep(0) 可提升 chunk 及时性，非必需但有时有用</h1><pre class="brush:php;toolbar:false"><code>    await asyncio.sleep(0)</code>

@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学习网公众号，一起学习编程~