当前位置：首页 > 文章列表 > 文章 > python教程 > FastAPI流式响应chunked传输方法

FastAPI流式响应chunked传输方法

2026-02-16 15:03:48 0浏览收藏

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

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