asyncio.get_event_loop() 弃用警告处理

在 Python 异步编程中，当你使用 asyncio.get_event_loop() 时遇到 "DeprecationWarning: There is no current event loop" 警告，这表明你的代码需要更新以符合最新的 asyncio 最佳实践。

问题原因

从 Python 3.10 开始，asyncio.get_event_loop() 在没有当前事件循环时会发出警告，而在 Python 3.11 及以上版本中，这将变成一个运行时错误。

原始的问题代码：

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.create_task(amain(loop=loop))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

这段代码的问题在于：在没有当前事件循环的线程中调用 asyncio.get_event_loop()，这在较新的 Python 版本中已不再被允许。

解决方案

方案一：显式创建和设置事件循环

如果 amain 函数需要一个事件循环参数，你应该显式创建和设置事件循环：

import asyncio

if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(amain(loop=loop))
    except KeyboardInterrupt:
        pass

注意

虽然这种方法避免了弃用警告，但传递事件循环参数给 amain 可能不是最佳实践。建议在函数内部获取运行中的事件循环。

方案二：修改 amain 函数（推荐）

更好的做法是修改 amain 函数，使其在内部获取当前运行的事件循环：

import asyncio

async def amain():
    # 在函数内部获取当前事件循环
    loop = asyncio.get_running_loop()
    # 其他逻辑...

if __name__ == '__main__':
    try:
        asyncio.run(amain())
    except KeyboardInterrupt:
        pass

最佳实践

使用 asyncio.get_running_loop() 而不是 asyncio.get_event_loop() 来获取当前运行的事件循环，这更加安全可靠。

方案三：WebSocket 服务器示例

如果你正在创建一个 WebSocket 服务器，可以这样实现：

import asyncio
import websockets

async def echo(websocket):
    async for message in websocket:
        await websocket.send(message)

async def main():
    async with websockets.serve(echo, "localhost", 50558):
        await asyncio.Future()  # 永远运行

if __name__ == "__main__":
    asyncio.run(main())

不同方法的比较

方案一：显式循环

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

方案二：asyncio.run

asyncio.run(main_task())  # 自动创建和管理循环

方案三：run_forever

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
server = create_server()
loop.run_until_complete(server)
loop.run_forever()

关键区别

• asyncio.run()：自动创建、运行和清理事件循环，适合大多数简单应用
• run_until_complete()：运行直到特定任务完成，然后停止
• run_forever()：无限运行事件循环，直到明确停止

重要提示

run_until_complete() 和 run_forever() 不会像 asyncio.run() 那样自动清理异步生成器。如果你使用这些方法，需要手动处理资源清理。

总结

处理 "There is no current event loop" 警告的最佳方法是：

1. 优先使用 asyncio.run() 作为程序入口点
2. 在异步函数内部使用 asyncio.get_running_loop() 获取当前事件循环
3. 如果必须显式管理事件循环，使用 asyncio.new_event_loop() 和 asyncio.set_event_loop()

这些实践确保了你的代码与现代 Python 版本的 asyncio 框架兼容，并且遵循了异步编程的最佳实践。