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

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

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

asyncio.run(main())

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

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

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

调用卡主的任务

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

import asyncio

async def worker(name):  
    for i inrange(3):  
        print(f"{name}: {i}")  
        await asyncio.sleep(1)

async def main():  
    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

async def increment():  
global counter  
  temp = counter  
await asyncio.sleep(0.1)  # Simulate delay  
  counter = temp + 1

async def main():  
await asyncio.gather(increment(), increment())

asyncio.run(main())

编写asynctest

安装 asynctest

pip install asynctest

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

import asynctest

class TestRaceCondition(asynctest.TestCase):  
  async def test_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. 1. pdb 可帮助你跟踪一些简单的问题，例如未等待的协程。
2. 2. aioMonitor 让你实时了解正在运行的任务和事件循环。
3. 3. asynctest 确保在 bug 进入生产环境之前捕获并修复它们。

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

更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言，推荐大家关注一个微信公众号：Python编程学习圈。每天分享行业资讯、技术干货供大家阅读，关注即可免费领取整套Python入门到进阶的学习资料以及教程，感兴趣的小伙伴赶紧行动起来吧。