Flask集成FlaskSocketIO报错：Cannot obtain socket from WSGI environment

问题根因 #

1. RuntimeError: Cannot obtain socket from WSGI environment报错的核心原因是Flask默认使用的WSGI服务器只支持普通HTTP的请求-响应模型，不兼容WebSocket全双工通信协议，无法完成WebSocket握手。浏览器显示连接成功是Socket.IO自动降级到了HTTP长轮询模式兜底，并非正常的WebSocket连接，消息推送阶段就会触发异常。
2. 连接成功但收不到test_message推送，是因为你的消息推送来自Celery独立进程，默认配置下SocketIO实例的连接池仅在当前进程内共享，跨进程的emit操作无法触达前端的Socket连接。

解决方案 #

第一步：修复WebSocket协议支持问题 #

二选一即可，不需要同时使用gevent和eventlet。

方案A：使用eventlet（Flask-SocketIO官方推荐方案）

1. 安装依赖：

pip install eventlet

2. 原有代码不需要修改，直接启动服务即可，socketio.run()会自动识别eventlet并启用WebSocket支持。
3. 如果同时使用Celery出现兼容问题，需要在Celery启动入口的最顶部添加猴子补丁配置：

import eventlet
eventlet.monkey_patch()

同时Celery启动命令指定使用eventlet池：

celery -A 你的Celery实例名 worker --pool=eventlet -l info

方案B：使用ASGI服务器运行（不想用异步协程库可选）

1. 安装ASGI服务器依赖：

pip install uvicorn

2. 修改app.py的启动逻辑：

from app import create_app, socketio

app = create_app()

if __name__ == '__main__':
    import uvicorn
    uvicorn.run("app:app", host="127.0.0.1", port=5000, workers=1)

注意ASGI模式下如果要开多worker，必须配置共享消息队列，否则不同worker的连接无法互通。

第二步：修复跨进程消息推送问题 #

需要配置消息中间件做跨进程的消息同步，推荐使用Redis，也可以替换为RabbitMQ。

1. 安装Redis依赖：

pip install redis

2. 修改__init__.py中的SocketIO初始化配置，添加消息队列参数：

socketio.init_app(app, cors_allowed_origins='*', message_queue='redis://localhost:6379/0')

3. 如果是Celery任务内部直接调用emit推送消息，不要直接导入Flask应用内的socketio实例，需要单独初始化一个绑定相同消息队列的SocketIO实例：

from flask_socketio import SocketIO
# 仅初始化绑定消息队列，不需要绑定Flask应用
socketio = SocketIO(message_queue='redis://localhost:6379/0')

# Celery任务内直接调用即可
@celery.task
def your_task():
    socketio.emit('test_message', {'msg':'Test'}, namespace='/test')

验证 #

完成配置后启动服务，服务端不会再抛出WSGI相关报错，浏览器控制台WebSocket连接失败提示消失，调用/send_message接口后即可收到test_message事件的日志。

内容的提问来源于stack exchange，提问作者Dr. Funkenstein