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

大家好，欢迎来到IT知识分享网。

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

1. 用于跟踪异步 bug

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

一个难以发现的错误

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

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

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

pdb调试方式

使用pdb捕获问题

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

运行脚本，当它暂停时：

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

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

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

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

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

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

调用卡主的任务

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

import asyncio asyncdefworker(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

修改代码以包含监视器：

async def main(): 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 asyncio counter = 0 asyncdefincrement(): global counter temp = counter await asyncio.sleep(0.1) # Simulate delay counter = temp + 1 asyncdefmain(): await asyncio.gather(increment(), increment()) asyncio.run(main())

编写asynctest

安装 asynctest

pip install asynctest

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

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

解决问题

添加锁以防止重叠增量：

lock = asyncio.Lock() async def increment(): global counter async with lock: temp = counter await asyncio.sleep(0.1) counter = temp + 1

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

结论

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

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

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