如何在Flask应用中集成socketio.AsyncServer？

一、配置正确性说明 #

你的AsyncServer配置是正确的：

• async_mode='asgi'：AsyncServer专为ASGI环境设计，该配置符合其运行要求；
• cors_allowed_origins="*"：开发阶段可临时启用全域名跨域，生产环境建议替换为具体允许的域名列表。

二、集成步骤与完整代码 #

Flask 2.0及以上版本支持ASGI模式，需通过socketio.ASGIApp适配器将AsyncServer与Flask应用绑定，具体代码如下：

from flask import Flask
import socketio

# 创建Flask应用实例
app = Flask(__name__)

# 初始化AsyncServer
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins="*")

# 将AsyncServer与Flask的ASGI应用集成
asgi_app = socketio.ASGIApp(sio, app)

# 示例：定义SocketIO连接/断开事件
@sio.event
async def connect(sid, environ):
    print(f"客户端 {sid} 已连接")

@sio.event
async def disconnect(sid):
    print(f"客户端 {sid} 已断开连接")

# 注意：不能使用Flask自带的app.run()，需用ASGI服务器运行
# 运行命令示例（需先安装uvicorn）：
# pip install uvicorn
# uvicorn 你的文件名:asgi_app --reload

三、关键说明 #

1. 集成原理：socketio.ASGIApp作为适配器，负责将SocketIO的ASGI请求和Flask的ASGI请求进行路由分发；
2. 运行方式：Flask自带的app.run()是WSGI服务器，无法运行ASGI应用，必须使用ASGI兼容服务器（如uvicorn、hypercorn）启动；
3. 事件定义：所有SocketIO事件处理函数需用async def定义，符合异步运行要求。

参考文档内容翻译 #

python-socketio官方文档相关内容翻译：

AsyncServer是针对ASGI环境实现的SocketIO服务器，需搭配ASGI兼容的Web框架与服务器使用。与Flask集成时，需将Flask切换至ASGI模式（Flask 2.0+支持），再通过ASGIApp适配器绑定SocketIO服务器与Flask的ASGI应用，最终通过ASGI服务器启动整体服务。

内容的提问来源于stack exchange，提问作者user3792705