flask中可用response包裹生成器实现长轮询:返回流式响应,设content_type为'text/plain'或'text/event-stream',禁用缓冲,yield消息加换行,配合redis pub/sub唤醒而非sleep,需配置nginx proxy_buffering off及gunicorn协程worker。

Flask 中用 Response 配合生成器实现长轮询
Flask 本身不内置长轮询支持,但可以通过手动构造流式响应(streaming response)达成。核心是让视图函数返回一个生成器,并设置合适的响应头,使客户端保持连接直到有新数据或超时。
关键点在于:不能用 return jsonify() 或普通字符串返回,必须用 Response 包裹生成器,并禁用自动缓冲。
- 生成器每次
yield一条消息(通常以换行分隔),结尾加\n防止某些客户端卡住 - 必须设置
content_type='text/event-stream'或更通用的'text/plain',避免浏览器解析成 HTML 后提前关闭连接 - 务必在生成器开头写入一个空行或占位符(如
yield ''),部分 Nginx / Gunicorn 会因首块为空拒绝流式传输 - 使用
time.sleep()模拟等待逻辑时,注意不要阻塞整个 Flask worker;生产环境建议改用事件队列(如 Redis Pub/Sub)唤醒
为什么不能直接用 stream_with_context?
stream_with_context 只解决上下文延续问题(比如访问 request 对象),它不改变响应行为本身。真正起作用的是 Response 的流式能力 + WSGI 服务器对 chunked transfer 的支持。
常见误区是以为加上 @stream_with_context 就能“自动长轮询”,其实它只是防止生成器里拿不到 request.args 这类数据——如果没用到请求上下文,根本不需要它。
仅使用 Python 将 Claw Code 虚拟化框架与 OpenClaw 集成,提供 184 种工具和 200 多条命令,用于安全分析和代码质量...
- 开发时用 Flask 自带的 dev server 可以跑通,但不保证流式稳定(尤其 macOS 上)
- 部署时若用 Gunicorn,需加
--worker-class=gevent或--worker-class=eventlet,默认 sync worker 会等生成器结束才发响应 - Nginx 默认缓存 1k 响应体再转发,必须配置
proxy_buffering off;和proxy_cache off;,否则客户端收不到实时数据
客户端如何正确发起长轮询请求?
所谓“长轮询”本质是:客户端发一次请求 → 服务端 hold 住连接 → 有数据就返回 → 客户端收到后立刻发下一次请求。这不是 Server-Sent Events(SSE),不需要 EventSource,用原生 fetch 更可控。
- 必须设
keepalive: true(实际由底层 TCP 处理),重点是禁用超时:signal: AbortSignal.timeout(60_000)不够,得靠服务端超时控制 - 响应状态码建议用
200,避免某些代理把204当作连接终止 - 每次响应后,客户端要检查内容是否为空或含错误标识,再决定是否重试,而不是无条件循环调用
- 别忘了加防抖:连续失败 3 次后指数退避,否则网络抖动会打爆服务端连接数
用 Redis Pub/Sub 替代轮询 sleep 的真实场景
硬 sleep 等待只适合 demo。真实业务中,数据到来不可预测,应该让生成器被外部事件唤醒。Redis Pub/Sub 是最轻量的解耦方式。
服务端订阅一个 channel,客户端触发事件时 publish 到该 channel,生成器在 redis_pubsub.listen() 上阻塞等待 —— 这比 while + sleep 高效得多,且连接数不随等待时间线性增长。
- 注意
redis-py的listen()是阻塞迭代器,直接放在生成器里 yield 即可 - 每个请求独占一个 subscriber,别复用 connection,否则多个请求会互相干扰
- 记得在生成器退出时(比如超时或 client 断开)调用
pubsub.unsubscribe(),否则 Redis 连接泄漏 - 如果不用 Redis,
threading.Event或asyncio.Queue(配合 Quart)也可替代,但跨进程时 Redis 更稳
长轮询真正的复杂点不在代码几行,而在于连接生命周期管理:客户端异常断开时服务端是否及时释放资源、反向代理怎么透传流、超时时间该设成 30 秒还是 90 秒、重连时要不要携带 last_event_id 避免丢消息——这些细节不处理好,上线后第一波流量就会暴露问题。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










