Python 异步调试神器：必须掌握的 3 个核心工具

在 Python 中调试异步代码感觉就像解决一个拼图，其中各个部分不断移动, 非阻塞特性对性能非常有用，但会带来争用条件、死锁和未处理的异常等挑战。多年来，我一直依赖三个关键工具来使调试异步 Python 代码变得易于管理, 在本文中，我将分享这些工具如何帮助你有效调试并节省大量时间的实际示例.

1. 用于跟踪异步 bug

如你曾经调试过 Python 代码，则可能使用过 Python 的内置调试器,虽然它不是为异步工作流量身定制的，但它仍然是检查变量和单步调试代码的可靠工具.

一个难以发现的错误

这是一个简单的异步程序，但有一个容易错过的错误：

import asyncioasyncdeffetch_data():  print("Fetching data...")  await asyncio.sleep(1)  print("Data fetched.")asyncdefmain():      fetch_data()  # Forgot `await`!  print("Task completed.")asyncio.run(main())

问题是什么？程序完成且没有错误，但不会运行，发生这种情况是因为 was called but not awaited，因此它没有执行。

pdb调试方式

使用pdb捕获问题

import pdbasyncdefmain():      pdb.set_trace()  # Add breakpoint      fetch_data()  # Missing `await`  print("Task completed.")asyncio.run(main())

运行脚本，当它暂停时：

1. 1. 使用命令单步执行代码
2. 2. 键入以检查其值，你将看到它是一个协程对象，而不是结果

发现问题后，通过添加 ：await

asyncdefmain():  await fetch_data()  print("Task completed.")asyncio.run(main())

pdb非常适合捕获像这样的简单错误，但对于更复杂的问题，我们需要专门的工具。

2.用于实时事件循环调试aiomonitor

当异步错误感觉像是隐藏在阴影中时 ， 任务以无法重现的方式冻结或重叠 ， 可以挽救局面，它允许实时检查事件循环，显示活动任务及其状态.

调用卡主的任务

下面是一个程序，其中两个任务同时运行，但一个任务偶尔会冻结：

import asyncioasyncdefworker(name):  for i inrange(3):  print(f"{name}: {i}")  await asyncio.sleep(1)asyncdefmain():      task1 = asyncio.create_task(worker("Task1"))      task2 = asyncio.create_task(worker("Task2"))  await asyncio.gather(task1, task2)asyncio.run(main())

当其中一名work停止打印时，就该引入 aiomonitor

调试方式

安装aiomonitor

pip install aiomonitor

修改代码以包含监视器：

asyncdefmain():  with aiomonitor.start_monitor():          task1 = asyncio.create_task(worker("Task1"))          task2 = asyncio.create_task(worker("Task2"))  await asyncio.gather(task1, task2)asyncio.run(main())

运行程序并使用 Telnet 连接到监视器：

telnet localhost 50101

在 REPL 中，键入 to view all running tasks，用于检查特定任务并确定为什么可能卡住或等待。

aiomonitor当你需要实时了解事件循环和正在运行的任务时，它会大放异彩。

3. 用于防止bug

调试只是成功的一半，处理异步 bug 的最好方法是防止它们，这就是它的用武之地，它是一个专为异步 Python 代码构建的测试库，可以轻松模拟协程和编写测试用例。

测试争用条件

这是一个典型的争用条件错误：

import asynciocounter = 0asyncdefincrement():  global counter    temp = counter  await asyncio.sleep(0.1)  # Simulate delay    counter = temp + 1asyncdefmain():  await asyncio.gather(increment(), increment())asyncio.run(main())

编写asynctest

安装 asynctest

pip install asynctest

编写一个测试用例来公开争用条件：

import asynctestclassTestRaceCondition(asynctest.TestCase):  asyncdeftest_race_condition(self):  global counter        counter = 0asyncdefincrement():  global counter            temp = counter  await asyncio.sleep(0.1)            counter = temp + 1await asyncio.gather(increment(), increment())self.assertEqual(counter, 2)  # 这里会失败

解决问题

添加锁以防止重叠增量：

lock = asyncio.Lock()asyncdefincrement():  global counter  asyncwith lock:        temp = counter  await asyncio.sleep(0.1)        counter = temp + 1

再次运行测试，它将通过。

结论

调试异步 Python 代码不一定是一场噩梦。使用正确的工具：

1. 1. pdb 可帮助你跟踪一些简单的问题，例如未等待的协程。
2. 2. aioMonitor 让你实时了解正在运行的任务和事件循环。
3. 3. asynctest 确保在 bug 进入生产环境之前捕获并修复它们。

这些工具中的每一个都为我在处理异步项目时节省了无数的时间，也为我节省了相当多的麻烦，希望他们也会为你省下调试时间和解决麻烦，祝调试愉快！

原文: https://www.tk1s.com/python/3-essential-tools-you-must-learn-for-debugging-async-code-in-python