flask可通过response包裹生成器实现sse:yield格式为"data: {...}\n\n",需设content_type='text/event-stream'、headers={'cache-control': 'no-cache', 'connection': 'keep-alive'},并启用threaded=true避免阻塞。

Flask 本身不内置 SSE 支持,但用 Response 包裹生成器、设对 content_type='text/event-stream'、每条消息以 data: 开头并以 \n\n 结尾,就能跑通。关键不是“加个插件”,而是绕过 Flask 默认响应流程,手动控制流式输出。
怎么写一个能被 EventSource 正确解析的生成器?
浏览器只认标准 SSE 格式:每条完整消息必须以 data: 开头、内容后跟两个换行符 \n\n;单个 \n 不触发 message 事件,\r\n\r\n 在某些代理下也不稳定,统一用 \n\n。
-
yield的字符串必须是纯文本,不能是 dict 或 list —— 要先json.dumps()再拼进data:前缀里 - 如果想带事件类型(比如
event: progress),需在同一块中连续写多行:yield "event: progress\nid: 123\ndata: {...}\n\n" - 别在生成器里直接
time.sleep(1)—— 开发服务器默认单线程,会卡住所有请求;真要等,改用yield后立刻返回,让前端控制重试间隔 - 若生成器内访问
request.args或g.user,必须用stream_with_context包裹,否则运行时报RuntimeError: Working outside of application context
为什么本地测试时连接总断开或只收一条?
Werkzeug 开发服务器对长连接支持弱,且默认启用响应缓冲,导致数据攒着不发、超时断连。
安全 Python 代码沙箱:在隔离的沙箱中执行任意 Python 代码,预装 requests、NumPy、pandas 库。返回 stdout、stderr。
- 启动时务必加
app.run(threaded=True),否则第二个 SSE 请求就会阻塞第一个 - 禁用缓存:在
Response中显式加 headers:headers={'Cache-Control': 'no-cache', 'Connection': 'keep-alive'} - 避免中间件干扰:开发时临时关掉
flask-compress或其他 WSGI 压缩中间件,它们会吞掉流式响应 - 客户端监听要对路:
eventSource.onmessage接默认message类型;如果服务端发了event: heartbeat,得写eventSource.addEventListener('heartbeat', ...)
生产环境部署要注意哪些代理和超时配置?
Nginx、Gunicorn 这类代理默认把长连接当异常处理,50–60 秒就断开,SSE 必须主动放宽。
- Nginx 配置里加:
proxy_read_timeout 300;、proxy_buffering off;、chunked_transfer_encoding on; - Gunicorn 启动加
--timeout 300 --keep-alive 5,推荐搭配gevent或eventletworker 类型 - 别用纯同步 Werkzeug 服务器上线 —— 它无法并发维持多个 SSE 连接,第二个连接进来就会卡死第一个
- 如果要用 Redis 广播,生成器里别轮询
redis.blpop(),改用pubsub.listen()阻塞监听,更轻量也更实时
最易被忽略的一点:SSE 是单向通道,没有内置用户隔离。你用全局字典存连接、靠 user_id 查队列,就得自己保证线程安全和连接生命周期管理——客户端关闭时,GeneratorExit 不一定触发,得靠心跳 + try/except BrokenPipeError 主动清理。不然内存泄漏比超时还快。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










