Python 并发与 asyncio 2.9 使用调试模式

2.9 使用调试模式

在前面的部分中，我们提到了协程在应用程序中的某个时刻始终应被 await。我们还看到了在协程和任务中运行计算密集型和阻塞代码的弊端。然而，很难判断协程在 CPU 上花费的时间是否过长，或者是否不小心忘记了在应用程序中的某个地方 await。幸运的是，asyncio 提供了调试模式来帮助我们诊断这些情况。

当我们在 debug 模式下运行时，如果协程或任务的运行时间超过 100 毫秒，我们会看到一些有用的日志消息。此外，如果我们没有 await 一个协程，会抛出一个异常，这样我们就可以看到在哪里正确地添加 await。有几种不同的方式可以运行在调试模式下。

2.9.1 使用 asyncio.run

我们用来运行协程的 asyncio.run 函数暴露了一个 debug 参数。默认情况下，这个参数设置为 False，但我们可以将其设置为 True 来启用调试模式：

asyncio.run(coroutine(), debug=True)

2.9.2 使用命令行参数

可以通过在启动 Python 应用程序时传递命令行参数来启用调试模式。为此，应用 -X dev：

python3 -X dev program.py

2.9.3 使用环境变量

我们还可以通过将 PYTHONASYNCIODEBUG 变量设置为 1 来使用环境变量启用调试模式：

PYTHONASYNCIODEBUG=1 python3 program.py

注意 在 3.9 之前的 Python 版本中，调试模式存在一个漏洞。当使用  时，只有布尔  参数才有效。命令行参数和环境变量只有在手动管理事件循环时才有效。

在调试模式下，如果协程花费的时间过长，我们会看到有信息的日志消息。让我们通过尝试在任务中运行计算密集型代码来测试这一点，看看是否会收到警告，如下所示：

import asynciofrom util import async_timed@async_timed()async def cpu_bound_work() -> int:    counter = 0    for i in range(100000000):        counter = counter + 1    return counterasync def main() -> None:    task_one = asyncio.create_task(cpu_bound_work())    await task_oneasyncio.run(main(), debug=True)

当运行这个时，我们会看到一条有用的消息，指出 task_one 花费的时间过长，因此阻塞了事件循环，使其无法运行任何其他任务：

Executing <Task finished name='Task-2' coro=<cpu_bound_work() done, defined at listing_2_9.py:5> result=100000000 created at tasks.py:382> took 4.829 seconds

这对于调试无意中进行阻塞调用的问题非常有帮助。默认设置会在协程花费超过 100 毫秒时记录警告，但这个时间可能比我们想要的长或短。要更改这个值，我们可以访问事件循环，如示例 2.24 所示，设置 slow_callback_duration。这是一个浮点值，表示我们希望慢回调持续时间的秒数。

import asyncioasync def main():    loop = asyncio.get_event_loop()    loop.slow_callback_duration = .250asyncio.run(main(), debug=True)

上面的代码将慢回调持续时间设置为 250 毫秒，这意味着如果任何协程花费超过 250 毫秒的 CPU 时间运行，我们就会打印出一条消息。

总结