asyncio.get_event_loop(): DeprecationWarning: 没有当前事件循环

Question

我正在构建一个 SMTP 服务器，并aiosmtpd使用这些示例作为构建的基础。下面是程序入口点的代码片段。

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

当我运行该程序时，我收到以下警告：

server.py:61: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()

实现这个的正确方法是什么？

Answer 1

asyncio.get_event_loop您的代码将在 Python3.10 上运行，但从 3.11 开始，当当前线程中没有运行循环时调用将出错。由于您需要循环作为 的参数amain，显然，您必须显式创建并设置它。

使用 asyncio.run 启动主任务比使用loop.run_forever 更好，除非您有特定原因这样做。[但见下文]

尝试这个：

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

2023 年 4 月 15 日添加：

asyncio.run()我在这里所做的调用 和调用loop.run_forever()（如原始问题中所示）或 之间存在区别loop.run_until_complete()。当我写这个答案时，我没有意识到 asyncio.run()总是创建一个新的事件循环。loop因此，在我上面的代码中，传递给的变量amain不会成为“运行循环”。所以我的代码避免了 DeprecationWarning/RuntimeException，但它没有将有用的循环传递到amain.

要纠正该问题，请更换该行

asyncio.run(amain(loop=loop))

和

loop.run_until_complete(amain(loop=loop))

最好进行修改amain以获取函数内部正在运行的事件循环，而不是将其传递进去。然后您可以使用 启动程序asyncio.run。但如果无法改变，那就不可能了。

请注意run_until_complete，与 不同asyncio.run， 不会清理异步生成器。标准文档中记录了这一点。