本文深入解析了 FastAPI 中 StreamingResponse 实现流式响应的核心机制与实战要点，揭示其默认即通过 chunked transfer encoding 实时推送数据的原理，并强调真正实现“流式”的关键不在配置开关，而在于生成器设计、中间件兼容性（如 Nginx 必须禁用缓冲）、前端读取方式（需用 ReadableStream 而非 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 参数。

本篇关于《FastAPI流式响应chunked传输方法》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！