Python中如何复用asyncio事件循环？解决aiohttp会话已关闭报错

• 每次调用asyncio.run()都会启动一个全新的事件循环，执行完传入的协程后会立即关闭当前循环
• pymyq登录后生成的aiohttp会话、设备实例都会和第一次运行的事件循环绑定，第二次调用asyncio.run()启动的新循环无法访问旧循环中的会话资源，因此抛出RuntimeError: Session is closed
• 不需要创建多个事件循环，全程复用同一个事件循环即可解决问题

Flask属于持续运行的服务，不能每次调用异步方法都重启事件循环，根据你使用的Flask版本可以选择以下两种方案：

方案1：使用异步Flask（2.0+版本，推荐） #

Flask 2.0及以上版本原生支持异步路由，可以直接复用Flask启动的全局事件循环，无需手动管理循环生命周期：

from flask import Flask
import pymyq

app = Flask(__name__)
# 全局保存登录后的设备实例，避免重复登录
myq_device = None

class MyObject:
    async def login(self):
        global myq_device
        api = await pymyq.login("你的账号", "你的密码")
        devices = await api.get_devices()
        myq_device = next(iter(devices.values()))
    
    async def set_state(self, open_state: bool):
        if not myq_device:
            await self.login()
        if open_state:
            await myq_device.open()
        else:
            await myq_device.close()

my_obj = MyObject()

# 异步路由直接调用异步方法
@app.route("/login")
async def login_api():
    await my_obj.login()
    return "登录成功"

@app.route("/open")
async def open_api():
    await my_obj.set_state(True)
    return "设备已开启"

@app.route("/close")
async def close_api():
    await my_obj.set_state(False)
    return "设备已关闭"

方案2：兼容同步Flask版本 #

如果使用的是不支持异步的旧版Flask，可以后台启动一个全局复用的事件循环，所有异步协程都提交到该循环执行：

import asyncio
import threading
from flask import Flask
import pymyq

app = Flask(__name__)
myq_device = None

# 后台启动全局事件循环，全程复用
loop = asyncio.new_event_loop()
def run_loop():
    asyncio.set_event_loop(loop)
    loop.run_forever()
threading.Thread(target=run_loop, daemon=True).start()

class MyObject:
    def login(self):
        global myq_device
        async def _login_coro():
            api = await pymyq.login("你的账号", "你的密码")
            devices = await api.get_devices()
            nonlocal myq_device
            myq_device = next(iter(devices.values()))
        # 提交协程到全局循环执行，阻塞等待结果返回
        asyncio.run_coroutine_threadsafe(_login_coro(), loop).result()
    
    def set_state(self, open_state: bool):
        if not myq_device:
            self.login()
        async def _set_coro():
            if open_state:
                await myq_device.open()
            else:
                await myq_device.close()
        asyncio.run_coroutine_threadsafe(_set_coro(), loop).result()

my_obj = MyObject()

# 同步路由正常调用方法
@app.route("/login")
def login_api():
    my_obj.login()
    return "登录成功"

@app.route("/open")
def open_api():
    my_obj.set_state(True)
    return "设备已开启"

@app.route("/close")
def close_api():
    my_obj.set_state(False)
    return "设备已关闭"

• 不要在服务类方法中频繁调用asyncio.run()，该方法设计定位是程序单次入口调用，不适合作为多次异步执行的工具方法
• 全局复用的事件循环会保留pymyq生成的会话资源，只要不手动关闭循环，会话会一直有效

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