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 的会话也可直接复用。

官方参考

本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。

延伸阅读

获取专属方案

联系我们

加入社区

微信扫码
加入官方交流群

立即体验

在线开通,按量计费,真正的云服务!

立即开始

选择观测云版本

代码托管平台