到目前为止,我们一直使用方便的 asyncio.run 来运行我们的应用程序并为我们后台创建事件循环。由于其易用性,这是创建事件循环的首选方法。然而,有时我们可能不想要 asyncio.run 提供的功能。例如,我们可能想执行自定义逻辑来停止任务,这与 asyncio.run 的行为不同,比如让任何剩余的任务完成,而不是停止它们。
此外,我们可能希望访问事件循环本身可用的方法。这些方法通常是低级别的,因此应谨慎使用。然而,如果你需要执行任务,比如直接处理套接字或计划一个任务在未来特定时间运行,你就需要访问事件循环。虽然我们不应该过度管理事件循环,但有时这是必要的。
我们可以通过使用 asyncio.new_event_loop 方法来创建一个事件循环。这将返回一个事件循环实例。有了这个,我们就可以访问事件循环提供的所有低级方法。有了事件循环,我们就可以访问一个名为 run_until_complete 的方法,它接受一个协程并运行它直到完成。当我们完成事件循环后,需要关闭它以释放其使用的任何资源。这通常应在 finally 块中进行,以防异常抛出阻止我们关闭循环。使用这些概念,我们可以创建一个循环并运行一个 asyncio 应用程序。
import asyncioasync def main(): await asyncio.sleep(1)loop = asyncio.new_event_loop()try: loop.run_until_complete(main())finally: loop.close()
此代码清单中的代码与调用 asyncio.run 的情况相似,不同之处在于这不会执行取消任何剩余任务。如果我们想要任何特殊的清理逻辑,我们可以在 finally 块中完成。
时不时地,我们可能需要访问当前正在运行的事件循环。asyncio 暴露了 asyncio.get_running_loop 函数,允许我们获取当前的事件循环。作为一个例子,让我们看看 call_soon,它将安排一个函数在事件循环的下一次迭代中运行。
import asynciodef call_later(): print("I'm being called in the future!")async def main(): loop = asyncio.get_running_loop() loop.call_soon(call_later) await delay(1)asyncio.run(main())
在上面的代码清单中,我们的 main 协程通过 asyncio.get_running_loop 获取事件循环,并告诉它运行 call_later,该函数接收一个函数并在事件循环的下一次迭代中运行它。此外,还有一个 asyncio.get_event_loop 函数,允许你访问事件循环。
这个函数在调用时如果没有运行的事件循环,可能会创建一个新的事件循环,从而导致奇怪的行为。建议使用 get_running_loop,因为它在没有运行的事件循环时会抛出异常,避免任何意外。
虽然我们不应在应用程序中频繁使用事件循环,但在某些时候,我们可能需要配置事件循环的设置或使用低级函数。我们将在下一节中通过调试模式的例子来展示如何配置事件循环。