ThreadPoolExecutor中调用asyncio发Telegram消息报Event loop is closed错误

问题根源 #

asyncio.run()执行完毕后会自动关闭当前线程的事件循环。而ThreadPoolExecutor的线程是复用的，当同一个线程第二次触发异步调用时，asyncio.run()会尝试复用已关闭的事件循环，直接抛出RuntimeError: Event loop is closed。

解决方案 #

方法一：封装线程安全的异步调用工具函数

通过线程本地存储（threading.local）为每个线程单独维护事件循环，避免重复创建/关闭：

import asyncio
from threading import local

# 线程本地存储，每个线程独立存储自己的事件循环
_thread_local = local()

def run_async_thread_safe(coro):
    # 获取当前线程的事件循环，不存在则创建
    try:
        loop = _thread_local.loop
    except AttributeError:
        loop = asyncio.new_event_loop()
        _thread_local.loop = loop
        # 可选：注册线程退出时的循环关闭操作
        import atexit
        atexit.register(loop.close)
    
    # 根据循环状态处理协程
    if loop.is_running():
        return loop.create_task(coro)
    else:
        return loop.run_until_complete(coro)

在你的进度通知逻辑中，替换原调用方式：

# 原代码：asyncio.run(telegram_message(progress_msg))
# 替换为：
run_async_thread_safe(telegram_message(progress_msg))

方法二：修改Telegram工具函数的同步调用逻辑

直接在utils/telegram_utils.py中封装同步调用方法，手动管理事件循环：

# utils/telegram_utils.py
import asyncio

async def telegram_message(msg):
    # 原异步发送逻辑保持不变
    ...

def send_telegram_sync(msg):
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    
    # 检查循环是否已关闭，关闭则重新创建
    if loop.is_closed():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    
    loop.run_until_complete(telegram_message(msg))

然后在多线程函数中直接调用同步方法：

send_telegram_sync(progress_msg)

注意事项 #

• 事件循环是线程绑定的，绝对不能在不同线程之间共享同一个循环对象
• 如果你的异步逻辑需要长期运行（比如持续监听），不要用run_until_complete，而是通过create_task提交到运行中的循环

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