如何在Python中使用loop.stop()终止loop.run_forever()？

你的问题核心很明确：loop.run_forever()是阻塞调用，它会持续占用线程直到loop.stop()被触发，但你把loop.stop()放在了run_forever()之后，这行代码永远没机会执行——因为run_forever()会一直运行，根本不会走到后续的stop()语句。

官方文档中"Run the event loop until stop() is called"的意思是：run_forever()会持续运行，直到在事件循环运行期间有代码调用stop()，而非等run_forever()结束后再调用。

下面是几种可行的正确实现方式：

1. 在异步任务内调用stop() #

如果你的异步任务执行完毕后就需要停止循环，可直接在任务中调用loop.stop()：

import asyncio

async def test(loop):
    print("Test")
    # 任务完成后触发循环停止
    loop.stop()

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

loop.create_task(test(loop))

loop.run_forever()
# 循环终止后记得关闭事件循环
loop.close()

2. 用call_later延迟调用stop() #

如果不需要等待任务结束，想在指定时间后自动停止循环，可以通过loop.call_later()注册延迟回调：

import asyncio

async def test():
    print("Test")
    await asyncio.sleep(1.5)
    print("Test finished")

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

loop.create_task(test())
# 2秒后自动触发stop()
loop.call_later(2, loop.stop)

loop.run_forever()
loop.close()

3. 在其他线程中调用stop() #

如果需要通过主线程以外的逻辑触发停止（比如用户输入、外部事件），可以在独立线程中调用loop.stop()：

import asyncio
import threading
import time

async def test():
    print("Test started")
    await asyncio.sleep(3)
    print("Test completed")

def trigger_stop(loop):
    # 模拟外部事件触发停止逻辑
    time.sleep(4)
    loop.stop()

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

# 启动专门线程负责触发停止
threading.Thread(target=trigger_stop, args=(loop,)).start()

loop.create_task(test())
loop.run_forever()
loop.close()

无论哪种方式，核心都是让loop.stop()在loop.run_forever()运行期间被调用，这样才能有效终止事件循环。

内容的提问来源于stack exchange，提问作者Super Kai - Kazuya Ito