在Python中使用Asyncio系统(3-8)?在关闭过程中等待执行器完成

在关闭期间等待执行器

在前几节“快速入门”介绍了基本executor接口示例3-3，幸好阻塞的time.sleep()调用比asyncio.sleep()时间更短。因为这意味着执行器任务比main()协程完成得更快，因此程序正确地关闭了。

本节检查当执行器作业的完成时间比所有挂起的Task实例都长时，在关闭期间会发生什么。简单的回答是:如果不进行干预，你会看到像示例3-36中的main()代码所产生的错误。

示例 3-36 执行器需要太长时间才能完成

# quickstart.py
import time
import asyncio


async def main():
    loop = asyncio.get_running_loop()
    loop.run_in_executor(None, blocking)
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')


def blocking():
    time.sleep(1.5)  
    print(f"{time.ctime()} Hello from a thread!")


asyncio.run(main())

• (L13) 这个代码示例与示例3-3中的代码完全相同，除了阻塞函数中的休眠时间现在比异步函数中的更长。

运行这段代码会产生以下输出：

$ python quickstart.py
Fri Jan 24 16:25:08 2020 Hello!
Fri Jan 24 16:25:09 2020 Goodbye!
exception calling callback for <Future at [...snip...]>
Traceback (most recent call last):
<big nasty traceback>
RuntimeError: Event loop is closed
Fri Jan 24 16:25:09 2020 Hello from a thread!

幕后发生的情况是，run_in_executor()不创建Task实例：它只是返回一个Future。这意味着它不包含在asyncio.run()中被取消的“活跃任务”集合中，因此在asyncio.run()中调用的run_until_complete()不用等待执行器任务完成。在asyncio.run()中调用内部loop.close()会引发RuntimeError。

在写这本书的时候，Python 3.8中的loop.close()并不等待所有执行器作业完成，这就是为什么从run_in_executor()返回的Future会报出问题：当它解析时，循环已经关闭。在核心Python开发团队中有关于如何改进这一点的讨论，但在解决方案确定之前，你需要一种处理这些错误的策略。

建议：在Python 3.9中，asyncio.run()函数已得到改进，可以正确地等待执行程序关闭，但在写本文时，还没有将其反向移植到Python 3.8中。

修正这个问题的几个想法，都有不同的取舍，我们将看看其中的几个。我的这个练习的真正目标是帮助你从不同的角度考虑事件循环的生命周期，考虑在一个重要程序中可能进行互操作的所有协程、线程和子进程的生命周期管理。

第一个想法，也是最容易实现的，如例3-37所示，就是总是在协程内部等待一个执行器任务。

示例 3-37 选项A:把执行器调用封装到一个协程中

# quickstart.py
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor as Executor


async def main():
    loop = asyncio.get_running_loop()
    future = loop.run_in_executor(None, blocking)  
    try:
        print(f'{time.ctime()} Hello!')
        await asyncio.sleep(1.0)
        print(f'{time.ctime()} Goodbye!')
    finally:
        await future  


def blocking():
    time.sleep(2.0)
    print(f"{time.ctime()} Hello from a thread!")


try:
    asyncio.run(main())
except KeyboardInterrupt:
    print('Bye!')

• (L8) 这个想法的目的是修复run_in_executor()只返回一个Future实例而不是一个任务的缺点。我们不能在asyncio.run()中使用all_tasks()捕获作业，但是我们可以在future上使用await。计划的第一部分是在main()函数中创建一个future。
• (L14) 我们可以使用try/finally结构来确保在main()函数返回之前等待future函数完成。

代码可以运行，但是它对执行器函数的生命周期管理有很大的限制：这意味着你必须在创建执行器作业的每个范围内使用try/finally。我们更喜欢以创建异步任务的方式生成执行器作业，并且还让asyncio.run()内部的关机处理执行一个优雅的退出操作。

下一个想法，如例3-38中显示的，这个有点巧妙。因为我们的问题是一个执行器创建一个future而不是一个task，并且asyncio.run()中的关闭处理处理任务，所以我们的下一个计划是将future(由执行器产生)包装在一个新的task对象中。

示例 3-38 选项B：将执行器future添加到收集的任务中

# quickstart.py
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor as Executor


async def make_coro(future):  
    try:
        return await future
    except asyncio.CancelledError:
        return await future


async def main():
    loop = asyncio.get_running_loop()
    future = loop.run_in_executor(None, blocking)
    asyncio.create_task(make_coro(future))  
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')


def blocking():
    time.sleep(2.0)
    print(f"{time.ctime()} Hello from a thread!")


try:
    asyncio.run(main())
except KeyboardInterrupt:
    print('Bye!')

• (L15) 我们获取run_in_executor()调用返回的future，并把它传递给一个新的功能函数make_coro()。这里重要的一点是，我们正在使用create_task()，这意味着该任务会出现在asyncio.run()要处理关闭的all_tasks()列表中，并将在关闭过程中收到一个取消请求。
• (L6) 这个功能函数make_coro()只是简单地等待future完成，但至关重要的是，即使在CancelledError的异常处理程序中它也继续等待future完成。

这个解决方案在关闭时表现更好，建议你运行这个示例的时候，也就是在打印“Hello!”后立即按下Ctrl-C。关闭步骤还会等待make_coro()退出，这意味着它还将等待执行器作业退出。但是，这段代码非常笨拙，因为必须将每个执行器Future实例包装在make_coro()调用中。

如果我们愿意放弃asyncio.run()函数的便利性（直到Python3.9才能用），我们可以通过自定义循环处理做得更好一点，如示例3-39所示。

示例 3-39 选项C：就像露营一样，带上你自己的循环和执行器

# quickstart.py
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor as Executor


async def main():
    print(f'{time.ctime()} Hello!')
    await asyncio.sleep(1.0)
    print(f'{time.ctime()} Goodbye!')
    loop.stop()


def blocking():
    time.sleep(2.0)
    print(f"{time.ctime()} Hello from a thread!")


loop = asyncio.get_event_loop()
executor = Executor()  
loop.set_default_executor(executor)  
loop.create_task(main())
future = loop.run_in_executor(None, blocking)  
try:
    loop.run_forever()
except KeyboardInterrupt:
    print('Cancelled')
tasks = asyncio.all_tasks(loop=loop)
for t in tasks:
    t.cancel()
group = asyncio.gather(*tasks, return_exceptions=True)
loop.run_until_complete(group)
executor.shutdown(wait=True)  
loop.close()

• (L17) 这一次，我们创建自己的执行器实例。
• (L18) 我们必须将自定义执行器设置为循环的默认执行器。这意味着，只要代码调用在run_in_executor()中运行，它就会使用我们的自定义实例。
• (L20) 与前面一样，我们运行blocking函数。
• (L30) 最后，我们可以显式地等待所有执行器作业完成，然后再关闭循环。这将避免我们以前看到的“Event loop is closed”消息。我们可以这样做，因为我们可以访问执行器象；默认执行器未在asyncio API中公开，这就是为什么我们无法对它调用shutdown()，并被迫创建自己的执行器实例。

最后，我们有一个具有普遍适用性的策略：你可以在任何地方调用run_in_executor()，即使在所有异步任务完成后执行器作业仍在运行，你的程序还是会明确地关闭。

我强烈建议你亲手试试使用这个章节里显示的代码示例，尝试不同的策略来创建任务和执行器作业，及时错开它们，并尝试明确地关闭它们。我希望Python的未来版本将允许asyncio.run()函数等待执行器作业完成，但我希望本节中的讨论对你开发明确清晰的关机处理程序的思路会有一定的帮助。