如何避免socketio.AsyncClient实例代码运行时出现阻塞问题

问题根源 #

你遇到的阻塞问题来自两个错误的用法组合：

• asyncio.run_until_complete() 是同步阻塞接口，会等到传入的任务完全执行结束才会返回
• socketio.AsyncClient.wait() 是无限期运行的协程，作用是维持客户端连接存活、持续监听服务端消息，本身不会主动退出

两者结合后，第一个Connector实例初始化时就会卡住线程，自然无法继续创建更多实例。而去掉wait()后连接直接断开，是因为没有任务持有客户端实例的引用、也没有运行中的逻辑维持连接状态，协程执行完连接逻辑后直接销毁了客户端对象。

调整方案 #

核心调整原则是：不要在同步的__init__方法中阻塞等待长期运行的协程，将一次性的连接初始化逻辑和长期运行的后台监听逻辑拆分，仅将监听任务提交到事件循环作为后台任务运行，不等待其执行完成。

代码修改示例 #

import asyncio
import socketio

class Connector:
    def __init__(self, address):
        self._address = address
        self._asyncClient = socketio.AsyncClient()
        self._loop = asyncio.get_event_loop()
        # 注册事件回调
        self._asyncClient.on('connect',
                            handler=self.on_connect,
                            namespace='/aNamespace')
        self._asyncClient.on('connect_error',
                            handler=self.on_connect_error,
                            namespace='/aNamespace')
        self._asyncClient.on('disconnect',
                            namespace='/aNamespace',
                            handler=self.on_disconnect)
        # 提交连接初始化任务到事件循环，存到实例属性防止被GC回收
        self._conn_task = self._loop.create_task(self._init_conn())

    async def _init_conn(self):
        # 先完成连接建立（短期执行，不会无限阻塞）
        # 注意：namespaces参数需要传列表，字符串是无效参数
        await self._asyncClient.connect(self._address, namespaces=['/aNamespace'])
        # 连接建立后提交后台监听任务，无需等待执行完成
        self._loop.create_task(self._asyncClient.wait())

    # 新增销毁方法，程序退出前调用释放资源
    async def close(self):
        await self._asyncClient.disconnect()
        if not self._conn_task.done():
            self._conn_task.cancel()
            try:
                await self._conn_task
            except asyncio.CancelledError:
                pass

    # 你的原有回调方法保留即可
    def on_connect(self):
        print(f"已连接到{self._address}")
    
    def on_connect_error(self, err):
        print(f"连接{self._address}失败: {err}")
    
    def on_disconnect(self):
        print(f"与{self._address}断开连接")

上层使用示例 #

所有Connector实例初始化完成后，再统一启动事件循环即可：

async def main():
    # 可创建任意数量的Connector实例，初始化不会阻塞
    connectors = [
        Connector("http://127.0.0.1:5000"),
        Connector("http://127.0.0.1:5001")
        # 更多实例按需添加
    ]
    # 阻塞保持事件循环运行，所有后台连接和监听逻辑都会正常执行
    await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

注意事项 #

• 提交到事件循环的任务必须赋值给实例属性存储，否则会被Python的垃圾回收机制自动销毁，导致连接意外断开
• 如果你的上层应用本身就有持续运行的asyncio事件循环（比如FastAPI、aiohttp服务），不需要额外加await asyncio.Event().wait()，直接在服务启动逻辑中创建Connector实例即可
• 如果需要保证所有Connector连接成功后再执行后续业务逻辑，可以给每个Connector加一个asyncio.Future标记连接状态，上层统一等待所有Future完成即可

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