asyncio事件循环延迟创建出现重复实例，如何确保仅初始化一个？

我有多个需要在asyncio事件循环下运行的模块，但直到运行时才知道会加载哪些模块。因此在初始化阶段，我希望确保有且仅有一个生效的事件循环。

我尝试了以下代码，预期'A'代码块会创建并使用一个事件循环，'B'代码块会复用同一个事件循环：

try:
    loop = asyncio.get_running_loop()  # use existing event loop
    logger.info(f'*** A using existing loop {loop}')
except RuntimeError:
    loop = asyncio.new_event_loop()    # ... or create and use a new one
    asyncio.set_event_loop(loop)
    logger.info(f'*** A creating loop {loop}')

try:
    loop = asyncio.get_running_loop()  # use existing event loop
    logger.info(f'*** B using existing loop {loop}')
except RuntimeError:
    loop = asyncio.new_event_loop()    # ... or create and use a new one
    asyncio.set_event_loop(loop)
    logger.info(f'*** B creating loop {loop}')

预期输出：

INFO:builtins:*** A creating loop <ProactorEventLoop running=False closed=False debug=False>
INFO:builtins:*** B using existing loop <ProactorEventLoop running=False closed=False debug=False>

但实际结果是创建了两个事件循环：

INFO:builtins:*** A creating loop <ProactorEventLoop running=False closed=False debug=False>
INFO:builtins:*** B creating loop <ProactorEventLoop running=False closed=False debug=False>

最终这会导致RuntimeError：got Future <Future pending> attached to a different loop。

请问如何实现仅当事件循环不存在时才动态创建它？

问题根源 #

asyncio.get_running_loop()仅能获取当前正在运行的事件循环，你创建的循环只是被设为当前线程的默认循环，但并未启动（未调用loop.run_forever()或loop.run_until_complete()），所以第二次调用get_running_loop()依然会抛出RuntimeError，进而触发新循环的创建。

修正方案：使用asyncio.get_event_loop() #

在同步上下文（非异步函数内部），asyncio.get_event_loop()会返回当前线程的默认事件循环；如果不存在，会自动创建一个并设置为默认循环，无论该循环是否处于运行状态。

修正后的通用代码示例：

import asyncio
import logging

logger = logging.getLogger(__name__)

def get_or_create_loop():
    try:
        loop = asyncio.get_event_loop()
        # 额外处理循环已关闭的极端情况
        if loop.is_closed():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            logger.info(f'*** Created new loop (old one closed) {loop}')
        else:
            logger.info(f'*** Using existing loop {loop}')
        return loop
    except RuntimeError:
        # 处理极少数无法自动创建默认循环的情况
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        logger.info(f'*** Creating new loop {loop}')
        return loop

# 模块A初始化
loop_a = get_or_create_loop()
# 模块B初始化
loop_b = get_or_create_loop()

# 验证两个模块拿到的是同一个循环
assert loop_a is loop_b

补充说明 #

• 该函数在任意模块初始化时调用，都能确保同一线程内获取到同一个事件循环。
• 如果后续需要运行循环，只需调用loop.run_forever()或loop.run_until_complete(coro)即可，此时get_running_loop()也能正常获取到该循环。

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