Flask WebSockets 入门指南(Flask-SocketIO 实战)
用 Flask-SocketIO 给 Flask 应用加上 WebSocket 实时能力:项目配置、事件处理、房间与广播、前端对接,一步步实现即时通信。
直接回答:Flask 本身不含 WebSocket,但 Flask-SocketIO 扩展用 Socket.IO 协议为它补上实时双向通信:事件驱动的编程模型、房间分组广播、自动重连降级——即时消息、实时通知、协同编辑都能快速落地,同时保持 Flask 的轻量哲学。
为什么是 Socket.IO 而非裸 WebSocket
Socket.IO 在 WebSocket 之上加了一层"贴心":断线自动重连、不支持 WebSocket 的环境自动降级为长轮询、事件命名空间、房间管理。Flask-SocketIO 把这套能力带进 Flask。
项目配置
pip install flask-socketio
from flask import Flask, render_template
from flask_socketio import SocketIO
app = Flask(__name__)
import os
app.config["SECRET_KEY"] = os.environ["FLASK_SECRET_KEY"]
socketio = SocketIO(app) # 默认同源检查,不开放任意 Origin
if __name__ == "__main__":
socketio.run(app, debug=True) # 注意:用 socketio.run 而不是 app.run
生产环境配消息队列(Redis)支撑多进程广播:SocketIO(app, message_queue="redis://...")。
第一个事件处理器
Socket.IO 的世界里,通信单位是"事件"而非裸消息:
from flask_socketio import emit
@socketio.on("connect")
def on_connect():
print("客户端接入")
@socketio.on("chat")
def on_chat(data):
emit("chat", {"msg": data["msg"], "echo": True}) # 回给发送者
@socketio.on("disconnect")
def on_disconnect(reason=None):
print("客户端离开")
前端需安装与服务端兼容的 Socket.IO JavaScript 4.x 客户端,将固定版本构建文件放在 static/socket.io.min.js;不能使用浏览器原生 WebSocket 代替 Socket.IO 客户端:
<script src="/static/socket.io.min.js"></script>
<script>
const socket = io();
socket.emit("chat", { msg: "大家好" });
socket.on("chat", (d) => console.log(d));
</script>
房间与广播
from flask_socketio import join_room, leave_room
@socketio.on("join")
def on_join(data):
join_room(data["room"])
emit("notice", {"msg": "欢迎进群"}, to=data["room"])
@socketio.on("chat")
def on_chat(data):
emit("chat", {"user": data["user"], "msg": data["msg"]}, to=data["room"])
房间示例中的用户和 room 仅作演示,生产必须从认证身份取用户名,并在 join 与 chat 时验证房间权限。启动代码放在所有事件处理器注册之后。to=房间名 实现组播,broadcast=True 发给所有人(默认包含发送者;排除时显式指定 include_self=False)。聊天室、游戏房间、协同文档的骨架就此成型。
生产要点
- 多 worker/多副本必须配 Redis 消息队列,否则广播只覆盖本进程连接;
- 长连接需要负载均衡开启会话保持;
- 新部署优先评估 threading + simple-websocket 或兼容的 gevent;不要默认引入已进入维护退场的 eventlet。Gunicorn 每实例用 -w 1,多实例在支持粘性会话的代理后扩展,并配置消息队列。
常见问题(FAQ)
Q:Flask-SocketIO 与 FastAPI 原生 WebSocket 怎么选?
A:存量 Flask 项目直接用 Flask-SocketIO,无需换框架;新实时项目且想要原生 WebSocket/异步,FastAPI 更现代。Socket.IO 协议的自动重连与降级是它的独家价值。
Q:连接数上不去卡在哪?
A:检查消息频率、阻塞调用、线程/连接资源和慢客户端,再测量并选择后端;没有可通用的连接数或数量级提升。
Q:怎么做连接鉴权?
A:connect 事件接收 auth 数据或验证 Flask-Login 会话,避免把长期 token 放 URL,拒绝时 return False 断开。结合 Flask-Login 的会话也可直接复用。
官方参考
本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。