Flask需用生成器函数+Response返回text/event-stream流式响应,设content_type、禁用缓存,每条消息以data:开头并以\n\n结尾;须按用户ID映射连接(如字典存queue.Queue),避免广播;上线须用gunicorn+gevent等支持长连接的服务器。

Flask 怎么启动一个支持 SSE 的长连接响应
Server-Sent Events 本质是客户端用 EventSource 发起 GET 请求,服务端保持连接打开、持续流式返回 text/event-stream 格式的数据。Flask 默认视图函数返回后就关闭响应,所以必须绕过 Flask 的常规响应流程,手动控制响应流。
关键点:不能用 return 返回字符串或 JSON,得用生成器函数 + Response 构造流式响应,并显式设置 content_type='text/event-stream' 和禁用缓存。
- 必须在生成器函数里每条消息末尾加两个换行符
\n\n,否则浏览器不会触发message事件 - 每条消息推荐包含
data:前缀,可选id:和event:,例如:data: {"status": "done"}\nid: 123\nevent: progress\n\n - Flask 开发服务器(
app.run())默认不支持长连接并发,上线必须用gunicorn+gevent或uwsgi,否则第二个连接就会卡住第一个
怎么安全地把后台任务进度推给指定用户
SSE 是单向下行通道,没有内置的“按用户推送”机制。你得自己维护连接与用户身份的映射关系,否则所有连接都会收到同一份广播。
常见错误是直接在全局变量里 append 所有 client 的 generator,结果一个任务更新导致所有人收到相同数据。正确做法是按 session ID 或 token 区分连接,并只往目标用户的连接写入。
- 用字典如
active_connections = {}存储{user_id: queue.Queue()},后台任务往对应队列put(),SSE 视图从队列get()并 yield - 别用线程共享 list 或 dict 直接 append —— 多线程下不安全,且无法通知阻塞中的生成器
- 客户端断开时,Flask 不会自动触发异常,得靠心跳检测或捕获
GeneratorExit/BrokenPipeError清理连接 - 如果用户登录态靠 cookie,记得在 SSE 路由里校验
request.cookies.get('session_id'),防止未授权连接
为什么前端 EventSource 一直 reconnect 或报错 net::ERR_FAILED
这不是前端代码写错了,大概率是服务端响应头、连接中断处理或反向代理配置没对。
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
典型现象:页面加载后立刻重连、控制台显示 EventSource failed to connect、Chrome Network 面板里状态码是 200 但响应体为空。
- 检查响应头是否含
Cache-Control: no-cache和Connection: keep-alive;Nginx 默认会缓存text/event-stream,需加配置:proxy_buffering off;和proxy_cache off; - Flask 生成器里不能有长时间阻塞操作(比如
time.sleep(5)),否则连接超时断开;改用queue.get(timeout=1)或asyncio.to_thread()(Py3.11+) -
EventSource不支持自定义 header,所以带认证的场景只能走 query 参数(如?token=xxx)或 cookie,别试图在headers里传Authorization - 某些浏览器(特别是 Safari)对空响应体敏感,首次 yield 前至少输出一条带
data:的消息,哪怕只是data: ping\n\n
要不要用 async def 写 SSE 视图
Flask 2.0+ 支持异步视图,但对 SSE 来说,异步不是必须的,甚至可能增加复杂度。
同步生成器(def + yield)配合线程安全队列就能满足大多数场景;强行上 async 要求整个调用链都 async(包括数据库、缓存),而多数 SSE 场景只是转发后台任务状态,瓶颈不在 I/O。
- 如果你用的是
celery或redis pub/sub,同步消费消息再 yield 更简单稳定 - 若真要用
async,必须搭配await queue.get()(如asyncio.Queue),不能混用同步queue.Queue - 异步视图在 gunicorn 下需用
gevent或eventletworker class,syncworker 会直接报错 - 调试时注意:
print()在异步生成器里可能乱序,优先用logging并设flush=True
最常被忽略的是连接生命周期管理——没人关它,它就不会停。后台任务结束、用户关闭标签页、网络闪断,这三种情况都要有对应的清理逻辑,否则连接堆积、内存泄漏、重复推送都会发生。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










