怎样在Python中实现WebSocket实时通信功能_通过Flask-SocketIO-Python教程-PHP中文网

flask-socketio不能直接用原生websocket协议，因其基于自有socket.io协议（含握手、心跳、包编码、命名空间等），与标准websocket不兼容；必须使用socket.io-client等官方客户端，而非new websocket()。

怎样在python中实现websocket实时通信功能_通过flask-socketio

Flask-SocketIO 为什么不能直接用原生 WebSocket 协议

因为 Flask-SocketIO 底层不依赖浏览器原生 WebSocket，而是自动降级支持 long-polling、eventsource 等多种传输方式。这意味着你写的服务端逻辑对客户端是否真正走 WebSocket 是无感的——但反过来，如果你强行在前端用 new WebSocket() 连接 Flask-SocketIO 启动的地址（比如 ws://localhost:5000/socket.io/），会立刻收到 400 Bad Request 或连接后立即断开。

根本原因：Socket.IO 是一个自有协议，包含握手、心跳、包编码（如 0{"sid":"abc"}40）、命名空间、事件名封装等，和标准 WebSocket 完全不兼容。

必须用官方客户端 socket.io-client（JS）或 python-socketio（Python）
服务端启动时默认开启 /socket.io/ 路由，不是根路径 /
若需纯 WebSocket 支持（如对接硬件设备、第三方系统），应换用 websockets 或 FastAPI + websockets，而非 Flask-SocketIO

如何正确初始化 Flask-SocketIO 并避免 Eventlet/Gevent 冲突

Flask-SocketIO 默认依赖异步服务器（eventlet 或 gevent）来处理并发连接。如果没装或版本不匹配，运行时可能静默降级为同步模式（只支持 1 个连接），或抛出类似 RuntimeError: You need to use the eventlet server 的错误。

推荐做法是显式指定异步模式，并统一安装配套依赖：

执行 pip install flask-socketio[eventlet]（非 pip install eventlet 单独装）
初始化时传入 async_mode='eventlet'，并确保 socketio.run(app, ...) 调用，而不是 app.run()
若项目已用 requests 等阻塞库，注意 eventlet.monkey_patch() 必须在所有 import 之前调用，否则 HTTPS 请求可能卡死

示例关键初始化代码：

python 查询技能

查询客流数据，输出JSON格式，可直接导入Bitable等可视化工具

下载

import eventlet
eventlet.monkey_patch()  # 必须最前
<p>from flask import Flask
from flask_socketio import SocketIO</p><p>app = Flask(<strong>name</strong>)
socketio = SocketIO(app, async_mode='eventlet', cors_allowed_origins='*')</p><p>if <strong>name</strong> == '<strong>main</strong>':
socketio.run(app, host='0.0.0.0', port=5000)
</p>

客户端连接失败的三个高频原因及验证方式

前端连不上？先别改代码，按顺序排查这三项：

服务端是否真在监听 http://localhost:5000（不是 http://127.0.0.1:5000）？浏览器同源策略会把二者视为不同源，导致跨域被拒；建议启动时明确写 host='0.0.0.0' 并配 cors_allowed_origins='*'
前端 JS 是否用了匹配版本的 socket.io-client？v4+ 客户端必须配 v5+ 服务端（Flask-SocketIO ≥5.3.0），旧版 socket.io-client@2.x 会卡在 connecting 状态；查法：console.log(io.version)
是否漏了 socket.on('connect_error', (err) => console.error(err))？很多失败其实返回了具体错误（如 "timeout"、"transport error"），但默认不打印

最小可测前端代码：

<script src="https://cdn.socket.io/4.7.2/socket.io.min.js"></script><script>
  const socket = io('http://localhost:5000', { transports: ['websocket'] });
  socket.on('connect', () => console.log('connected'));
  socket.on('connect_error', (err) => console.error('connection failed:', err));
</script>

发送消息时要注意命名空间、事件名和数据序列化的边界

Flask-SocketIO 的事件分发基于三元组：namespace + event + data。其中 event 是字符串标识符，不是函数名；data 会被 JSON 序列化，因此不能传 datetime、bytes、自定义类实例等。

服务端监听用 @socketio.on('my_event', namespace='/chat')，客户端触发用 socket.emit('my_event', {'msg': 'hi'}, {'namespace': '/chat'})
若未指定 namespace，默认为 '/'，前后端必须一致，否则事件收不到
传 datetime 会报 TypeError: Object of type datetime is not JSON serializable；应提前转成 ISO 字符串：dt.isoformat()
大对象（如图片 base64）建议分片或改用二进制事件（socketio.send(..., binary=True)），否则可能触发默认 1MB 消息限制

服务端接收示例：

@socketio.on('send_message')
def handle_message(data):
    # data 是 dict，已从 JSON 解析完成
    if 'text' not in data:
        return
    # 广播给同 namespace 所有人（不含发送者）
    socketio.emit('receive_message', {
        'sender': 'user1',
        'text': data['text'],
        'ts': datetime.now().isoformat()
    }, include_self=False)

实际部署时最容易忽略的是反向代理配置：Nginx 需显式透传 Upgrade 和 Connection 头，且 location 要匹配 /socket.io/ 路径，否则长连接会在代理层被切断。