Flask中用yield实现服务端流式响应的关键是:返回生成器函数(非调用结果),设Content-Type和Cache-Control头,每条SSE消息以"data: ...\n\n"格式yield,禁用Nginx/Gunicorn缓冲,并避免同步阻塞(如time.sleep)。

Flask 中如何用 yield 实现服务端流式响应
Flask 本身不阻塞响应生成,只要返回一个可迭代对象(比如生成器),它就会边 yield 边发数据给客户端。关键不是“开启某个开关”,而是确保响应体是生成器、禁用自动缓冲、且客户端能持续读取。
常见错误是直接 return 一个 list 或字符串——这会等全部生成完才发;或者没设 Content-Type 和 Cache-Control,导致浏览器缓存整块响应或解析失败。
- 函数必须用
yield逐段产出字符串(注意:每段需为str或bytes,不能混用) - 在视图函数开头显式设置
response.headers["Content-Type"] = "text/event-stream"或"text/plain",取决于用途 - 加
response.headers["Cache-Control"] = "no-cache"防止中间代理缓存 - 生产环境务必配好 WSGI 服务器(如 Gunicorn 启动时加
--worker-class=gevent或用 uWSGI 的enable-threads),否则默认同步 worker 会卡住整个流
为什么用 text/event-stream 而不是 text/plain
text/event-stream 是浏览器原生支持的流式协议,自带自动重连、事件类型标记(event:)、数据分隔(data:)和 ID 管理。适合实时日志、通知推送等场景;而 text/plain 更轻量,适合 CLI 工具或 curl 直接消费,但需要客户端自己处理粘包和断连。
- 用
text/event-stream时,每段 yield 必须以"data: ...\n\n"结尾(两个换行),例如:yield "data: hello\n\n" - 若用
text/plain,只需 yield 原始内容,但客户端要用response.body.getdecoder()或按行读取(for line in response.iter_lines():) - 某些旧版 Nginx 默认会缓冲
text/event-stream响应,需在配置中加proxy_buffering off;和chunked_transfer_encoding on;
如何避免流被 WSGI 中间件提前关闭
很多 Flask 扩展(如 flask-compress、flask-talisman)会在响应写入前尝试读取整个 body,导致生成器被一次性耗尽,失去流式意义。Gunicorn 的默认 sync worker 也会因超时 kill 掉长时间运行的请求。
- 禁用压缩中间件:初始化时传参
app = Flask(__name__, static_folder=None),并在注册Compress前检查request.headers.get("Accept-Encoding") - Gunicorn 启动加参数:
--timeout 300 --keep-alive 5 --worker-class gevent(gevent 支持异步 I/O,避免阻塞) - 在生成器内部做心跳:每隔几秒
yield ":\n\n"(SSE 注释行),防止代理或浏览器因空闲断连 - 别在生成器里调
time.sleep()—— 应改用asyncio.sleep()+async def视图(需搭配Quart或 Flask 2.3+ 的异步支持)
客户端怎么稳定接收 Flask 流式响应
浏览器用 EventSource 最省事,但只支持 GET 且无法带认证头;curl 或 Python requests 则更灵活,但得手动处理 chunk 解析。
- 浏览器端推荐:
const es = new EventSource("/stream"); es.onmessage = e => console.log(e.data); - requests 客户端要设
stream=True,并用r.iter_lines()或r.iter_content(),注意默认decode_unicode=True可能出错,建议关掉再手动.decode("utf-8") - 如果后端 yield 的是 JSON 字符串,前端不要用
JSON.parse()直接套整行——先trim()去首尾空格,再判断是否为空行 - 移动端 WebView 或某些 Electron 版本对 SSE 支持不稳定,此时应 fallback 到轮询或 WebSocket
最易被忽略的是部署链路中的缓冲层:从 Nginx 到 CDN 再到客户端网络栈,每一层都可能攒包或超时。真要压测流式接口,得逐层确认缓冲行为,而不是只看 Flask 代码有没有 yield。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











