Flask原生不支持WebSocket,需用uvicorn+websockets等ASGI服务器替代;WebSocket服务应独立端口运行,通过Redis等共享数据,避免Flask上下文错误;Nginx需配置Upgrade和Connection头,前端须用wss://(HTTPS时)并等待onopen后通信。

Flask 本身不支持 WebSocket,别硬套 flask 启动服务
直接用 flask.run() 启动的服务器(比如开发时敲 python app.py)压根没有 WebSocket 能力——它底层是 Werkzeug 的 WSGI 服务器,只处理 HTTP 请求,ws:// 或 wss:// 连接一上来就会被拒绝,常见现象是浏览器控制台报 WebSocket connection to 'ws://localhost:5000/ws' failed。
必须换支持 ASGI 或原生 WebSocket 的服务器:
-
uvicorn+websockets库(推荐,轻量、调试友好) -
gunicorn配gevent-websocket(老项目兼容用,但 gevent 兼容性差,Python 3.12+ 已不建议) -
hypercorn(ASGI 兼容好,但对纯 WebSocket 路由支持不如 uvicorn 直观)
别在 Flask 路由里写 @app.route('/ws')——WS 不是 HTTP 路由,得交给专门的 WebSocket 服务器实例接管。
用 websockets 库手写 WebSocket 服务,和 Flask 共享数据要小心
最稳的路子是:Flask 负责 API 和页面渲染,另起一个 websockets.serve() 实例跑在另一个端口(如 8001),两者通过共享内存(threading.local)、Redis 或全局 dict 通信。别试图让 Flask 的 request 或 session 流进 WebSocket handler——WS 连接生命周期和 HTTP 完全无关。
立即学习“Python免费学习笔记(深入)”;
常见错误:在 websocket_handler 里调 current_app 或依赖 Flask 的上下文,结果抛 RuntimeError: Working outside of application context。
实操建议:
- 用
async def websocket_handler(websocket, path),不是普通函数 - 连接建立后,用
await websocket.recv()/await websocket.send(...)收发,别用print()模拟 - 若需关联用户身份,从首次
recv()解析 token 或 session_id,别依赖 Cookie 或 Header(WS 握手时虽能传,但后续无自动携带) - 共享状态用
asyncio.Queue或concurrent.futures.ThreadPoolExecutor包裹的 dict,避免竞态
前端连不上 ws://?检查协议、端口和反向代理配置
开发时浏览器访问 http://localhost:5000,但 WebSocket 却连 ws://localhost:5000/ws ——这大概率失败。因为 Flask 在 5000,而 WebSocket 服务其实在 8001,但前端代码没改端口。
更隐蔽的问题是 Nginx 反向代理。如果线上用 Nginx,必须显式透传 WebSocket 头:
location /ws {
proxy_pass http://backend_ws;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
}
漏掉 Upgrade 和 Connection 两行,Nginx 会把 WS 握手当普通 HTTP 代理,返回 400 或直接断连。
其他易踩点:
- HTTPS 页面(
https://)不能连ws://,必须用wss://,且后端证书得有效 - Chrome 控制台 Network 标签页里,WS 连接不会显示在 XHR 下,得切到 “WS” 分类看
- 用
new WebSocket('ws://...')后,立刻.send()可能报错,得等onopen触发后再发
并发连接多时,别让每个 WebSocket 协程阻塞事件循环
websockets 默认为每个连接启一个协程,看着很轻量,但一旦你在 handler 里写 time.sleep(5)、调同步数据库驱动(如 psycopg2 默认同步模式)、或做 CPU 密集计算,整个事件循环就卡住,其他连接收不到消息。
正确做法:
- 耗时 I/O(DB 查询、HTTP 调用)用异步驱动:
asyncpg替psycopg2,aiohttp替requests - CPU 密集任务扔给线程池:
await loop.run_in_executor(None, heavy_calc, data) - 加心跳保活:
ping_interval=20参数传给websockets.serve(),不然 NAT 网关可能静默断连 - 限制最大连接数:
websockets.serve(..., max_connections=100),防资源耗尽
真正难的不是“怎么连上”,是连接稳定后,怎么在不拖垮服务的前提下,把业务逻辑干净地塞进异步流里——这里没银弹,得按实际 IO 类型选适配的异步生态组件。


















