Python asyncio调用stop()后事件循环未停止的问题排查

核心问题原因 #

1. stop()方法非线程安全
   asyncio的EventLoop.stop()方法不支持跨线程直接调用，你的代码在Timer线程中直接调用该方法，会导致Event Loop内部状态混乱，无法正确响应停止信号，即使没有报错也会出现run_forever()无法返回的情况。

2. 同步阻塞调用卡住Event Loop
   myCouroutine()中使用time.sleep(3)是同步阻塞操作，会占用Event Loop线程的全部执行时间，导致Loop无法处理任何其他任务（包括停止信号），进一步加剧了停止失败的问题。

3. 协程提交方式错误
   使用call_soon_threadsafe(asyncio.ensure_future, myCouroutine())的方式不符合asyncio的线程安全规范，asyncio.ensure_future需要在Event Loop所属线程内调用，跨线程提交协程应该使用官方推荐的方法。

修正后的代码 #

import asyncio
import time
from threading import Thread
from threading import Timer

async def myCouroutine():
    # 替换为asyncio非阻塞sleep，避免卡住Event Loop
    await asyncio.sleep(3)

dbAccessLoop = None

def stopEventLoop():
    # 通过线程安全接口触发stop操作
    dbAccessLoop.call_soon_threadsafe(dbAccessLoop.stop)

def eventLoopThreadMain():
    global dbAccessLoop
    try:
        dbAccessLoop = asyncio.new_event_loop()
        asyncio.set_event_loop(dbAccessLoop)
        dbAccessLoop.run_forever()
    finally:
        dbAccessLoop.close()

def main():
    thread = Thread(target=eventLoopThreadMain)
    thread.start()

    stopLoopTimer = Timer(3, stopEventLoop)
    stopLoopTimer.start()

    # 等待Event Loop初始化完成
    while dbAccessLoop is None:
        time.sleep(0.1)
    
    # 官方推荐的跨线程提交协程方式
    asyncio.run_coroutine_threadsafe(myCouroutine(), dbAccessLoop)

    thread.join()

if __name__ == '__main__':
    main()

关键修正点说明 #

• 用await asyncio.sleep(3)替代time.sleep(3)，保证协程不会阻塞Event Loop线程，让Loop能及时处理停止信号。
• 通过call_soon_threadsafe()调用stop()，确保对Event Loop的操作线程安全，避免内部状态异常。
• 使用asyncio.run_coroutine_threadsafe()跨线程提交协程，这是asyncio官方提供的线程安全协程提交方法，能正确将协程加入目标Event Loop的任务队列。
• 添加等待Event Loop初始化的逻辑，避免主线程在Loop未创建完成时执行操作。

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