企业官网建设流程全解析

1. 协程卡死的那一晚：调试该从哪里下手

先从一个很真实的场景说起。某天我写了一个协程下载器，任务列表也不长，几十个并发请求，跑起来挺正常。但偶尔会在某个文件下载到一半时整个程序卡住，键盘按了 Ctrl+C 也只看到Task was destroyed but it is pending，连堆栈都抓不到。更难受的是，这种问题不是必现的，重跑几次可能又好了。

那时候我才意识到，协程调试跟线程调试完全是两码事。线程卡住时可以靠py-spy dump直接拿到 Python 层栈，可协程的调用链经常是断的：你看到的是"当前正在等待的协程"，但真正卡住的前因后果——谁在等它、为什么没人唤醒它——回溯信息里根本看不到。

1.1 为什么协程的回溯总是不完整

很多人第一次接触协程调试，最先碰到的就是这个问题：抛出来的异常堆栈只有半截。比如你await一个队列get()，超时后被取消，回溯里只能看到Queue.get的等待位置，却看不到这个任务是在哪个业务逻辑里发起的。

原因在于协程的调用结构。普通的函数调用栈是"你调我、我调你"的一条线，回溯天然完整。而协程是"创建后挂在事件循环上，由调度器决定什么时候执行"，A 协程awaitB 协程时，A 的状态被冻结在cr_frame里，B 的运行跟 A 在代码层面上看完全不连续。一旦出问题，Python 默认只打印当前正在运行的那一帧，冻结的父协程不会自动展开。

所以排查协程问题，第一件事就是抛弃"看异常堆栈"的习惯。你要学会主动去"翻"冻结帧。

1.2 把火山口撬开：debug 模式与守护进程

Python 官方其实给了两把基础工具，只是很多人没当回事。

第一把是asyncio.run(debug=True)，或者设环境变量PYTHONASYNCIODEBUG=1。开了之后事件循环会输出调度告警，比如回调执行太慢、任务创建后没有及时 await、Task was destroyed but it is pending这类问题都会显式打出来。注意这个不是玄学，它会在循环调度时做额外检查，代价是性能略有下降，但排查阶段根本无所谓。

第二把是给正在跑的进程加一个"急救通道"。假设程序已经卡住了，你不想重启，也不想丢失现场，最直接的办法是挂一个信号处理器。Linux 下用signal.SIGUSR1，Windows 下可以用SIGBREAK，信号触发时把当前所有协程任务的栈打印出来。

我当时写了一个简单的 dump 脚本，核心思路是：

import asyncio import signal import sys def dump_tasks(loop): tasks = asyncio.all_tasks(loop) print(f"当前存活任务数: {len(tasks)}") for task in tasks: print(f"---- 任务名: {task.get_name()} ----") print(f"状态: {task._state}") # 获取协程链 coro = task.get_coro() frame = coro.cr_frame while frame: print(f"{frame.f_code.co_filename}:{frame.f_lineno} in {frame.f_code.co_name}") frame = frame.f_back

注意cr_frame只在 Task 处于pending状态时能拿到，完成之后这段帧信息就没了。所以要在卡死的第一时间去抓。把这段代码接到信号上之后，再遇到卡死我就直接kill -SIGUSR1 <pid>，然后把打印出来的栈链对照业务代码逐帧看，基本能定位到是哪个await没被唤醒。

1.3 一定要区分"卡住"和"慢"

调试之前还要确定一个问题：程序到底是死锁了，还是单纯跑得慢。我见过不少人拿着all_tasks()打印出来的栈折腾半天，发现其实是大批量请求把事件循环拖慢了。判断方法很简单：看 CPU 占用率。死锁通常伴随 CPU 空闲——事件循环里没有任何可执行的回调，进程空转；而"慢"的时候 CPU 往往是满的，或者忙在别的拧巴逻辑上。

这一步不用工具，看一眼就够。但很多人一跳进来就开 debug、挂信号、抓栈，反而忽略了这个前提。

2. 把隐藏的协程状态挖出来：任务快照与调度器日志

基础手法能定位到"哪个任务卡了"，但还不够。我遇到过最麻烦的问题是：任务 A 卡在Event.wait()，任务 B 卡在Queue.get()，双方看起来都无辜，但谁都知道是 A 和 B 在互相等——这种跨任务依赖，单看一个任务栈根本看不出来。

后来我的做法是"两板斧"：一板斧是任务快照，把所有任务的状态、等待对象、栈信息一次性拍下来；另一板斧是调度器日志，让每个协程从创建到销毁都留下痕迹。

2.1 拍一张"全家福"：所有任务的当前等待点

在信号处理器或者断点暂停的位置，不要只打印当前任务，要把全部任务一次捞出来。asyncio.all_tasks()返回的是当前事件循环上所有还没结束的 Task，遍历它，逐个打印等待的对象。Python 3.11 之前可以用task._fut_waiter，3.12 之后这个属性变了，但好在asyncio提供了task.get_coro()和task.get_stack()，直接遍历协程链就够了。

import asyncio def snapshot_all_tasks(loop): pending = [] for task in asyncio.all_tasks(loop): if task.done(): continue info = { "name": task.get_name(), "state": task._state, "stack": [], } coro = task.get_coro() frame = coro.cr_frame while frame: info["stack"].append( f"{frame.f_code.co_filename}:{frame.f_lineno} {frame.f_code.co_name}" ) frame = frame.f_back pending.append(info) # 按文件名+行号聚合，方便看出多个任务卡在同一处 from collections import Counter counter = Counter() for item in pending: if item["stack"]: counter[item["stack"][0]] += 1 print("按停靠点统计:") for loc, count in counter.most_common(10): print(f"{count}x {loc}") for item in pending: print("====") print("任务:", item["name"], item["state"]) print("调用链:") for line in item["stack"][:8]: print(" ", line)

这个函数我叫它"全家福"。打印完一眼就能看到：哦，原来有 5 个任务全卡在同一个文件第 42 行的await queue.get()。如果都是同一个停靠点，那问题十有八九出在"没人往这个队列里放东西"，而不是任务本身写错了。

另一个技巧是：如果你的代码用了自定义等待对象，比如自己封装的Future，那么task._fut_waiter会让你直接看到它到底在等哪个 future。这在排查"某个 Future 永远不被 set_result"的时候很管用。不过_前缀的私有属性可能在版本升级时变掉，建议代码里加个兼容封装。

2.2 给每个协程写"行车记录仪"

快照适合事后取证，但如果你想在没有崩溃的情况下搞清楚协程之间的执行顺序，就得靠日志了。我项目里的惯例是封装一个小小的任务追踪装饰器，给每一个异步入口打上生命周期日志：

import asyncio import logging import time logger = logging.getLogger("coro.trace") def trace_task(name=None): def decorator(func): async def wrapper(*args, **kwargs): task_name = name or func.__name__ task = asyncio.current_task() try: logger.info("[%s] 启动, task=%s", task_name, task.get_name()) result = await func(*args, **kwargs) logger.info("[%s] 完成, task=%s", task_name, task.get_name()) return result except asyncio.CancelledError: logger.warning("[%s] 被取消, task=%s", task_name, task.get_name()) raise except Exception: logger.exception("[%s] 异常, task=%s", task_name, task.get_name()) raise return wrapper return decorator

用的时候直接在业务入口函数上@trace_task("downloader")。这样日志里就有一条贯穿每条分支的时间线。配合 2.1 的任务快照，基本能做到"某个任务卡死时翻日志，能看到它是从哪一步走到这一步的、当时其他任务正在做什么"。

日志格式建议固定成可 grep 的纯文本。我自己常用的一套字段是：

日志多加一行、多带一个状态值，卡死的时候你就少猜一个变量。真的，这是我在好几个协程项目上踩过轮子之后总结出来的最朴素经验。

2.3 别忘了 asyncio 自带的结构化获取

Python 3.8 之后asyncio给 Task 增加了get_stack()和print_stack()，这俩直接调底层帧，比我自己遍历cr_frame更简洁。比如：

task.print_stack(limit=20)

会在 stderr 上打出等待链。不过它输出的上下文比较糙，没有all_tasks()那套能统计分布的能力，所以我自己还是偏好完整快照。但如果你是 quick check，用print_stack()比写二十行代码快多了。

3. 单测里控制时间：手写一个可暂停的协程测试基座

调试能解决"已经坏了"的问题，测试才能保证"以后不坏"。但协程测试最大的坑，是时间不可控。

你写一个测试，希望验证"5 秒超时后任务被取消"这个行为。傻瓜式做法是await asyncio.sleep(5)然后断言。CI 环境一抖动，5 秒变 8 秒，断言就挂了。更糟糕的是，这种慢测试累计多了，整个套件变得难以忍受。

核心思路就一条：把真实时间从测试代码里隔离出去，让事件循环的时钟变成我们自己可以拨动的旋钮。

3.1 为什么asyncio.run()在测试里不够用

asyncio.run()适合写小脚本，但测试里直接用会有几个麻烦：每个asyncio.run()都会 new 一个事件循环，跑完就销毁，你没法在多个测试之间共享状态；loop.time()是只读的，没法快进时间；测试里的asyncio.sleep()会真实地等这么久。

所以我的测试基座都是自己管理 loop 的。基础概念懂的人不多，其实asyncio.new_event_loop()返回之后，你可以不急着run_until_complete，先往里塞回调，塞完再跑。你要的"暂停"，其实是"不要把所有协程一口气跑完，而是每次只往前执行一步"。

asyncio里控制调度步长的工具是loop.call_soon()、loop.call_later()和loop.run_until_complete()的组合。但真正想控制睡眠时间，单纯靠这些还不够，因为asyncio.sleep(5)内部是给 loop 注册了一个call_later(5, ...)，事件循环内部的"现在"是loop.time()决定的，而这个time()原本是从系统时钟读的。

3.2 替换loop.time()是实现"时间快进"的关键

我们可以自己做一个假的 timer，把loop.time()替换成可控的虚拟时钟。这样测试代码里写await asyncio.sleep(3600)，实际一毫秒都不等，事件循环直接跳到 3600 秒之后执行回调。

import asyncio import unittest class ManualTimeLoop: """把 asyncio 内部时钟替换成可控的虚拟时钟。""" def __init__(self): self.loop = asyncio.new_event_loop() self._now = 0.0 # 替换 loop 内部的时间函数 original_policy = self.loop.get_debug if hasattr(self.loop, "get_debug") else None self.loop.time = lambda: self._now # call_later 本身依赖于 loop.time() / loop.call_at() # 子类化或者直接替换 call_at 即可保持内部一致性 self._original_call_at = self.loop.call_at def fake_call_at(when, callback, *args, context=None): # when 是由 loop.time() 决定的绝对时间，我们不需要改动， # 因为 loop.time() 已经被我们替换了，when 也是相对虚拟时钟的。 return self._original_call_at(when, callback, *args, context=context) self.loop.call_at = fake_call_at def advance(self, seconds): """往后拨虚拟时钟，立即执行所有到期回调。""" self._now += seconds # 让事件循环把到期的 delayed call 全部调度掉 self.loop.run_until_complete(asyncio.sleep(0)) def run(self, coro): return self.loop.run_until_complete(coro) def close(self): self.loop.close()

但上面这个写法有个地方容易踩坑：loop.call_later(when, ...)的参数when是相对延迟，它内部会转成call_at(loop.time() + when, ...)。因为我们把loop.time()换成了虚拟时钟，这个换算依然成立。核心逻辑就是：进来的相对秒数 + 虚拟当前时间 = 未来的虚拟绝对时间，事件循环内部的堆里还是按数字大小排序的，所以完全可行。

我实际项目里还会更进一步，把loop.call_later的延迟回调塞进一个手动 sort 的堆里，这样advance(seconds)可以直接从堆里把所有when <= new_now的回调一次性取出执行，不需要sleep(0)反复催。

import heapq class VirtualScheduler: def __init__(self, loop): self.loop = loop self._now = 0.0 self._heap = [] self._seq = 0 self._orig_cal = loop.call_later loop.call_later = self._v_call_later loop.time = lambda: self._now def _v_call_later(self, delay, callback, *args, context=None): self._seq += 1 when = self._now + delay heapq.heappush(self._heap, (when, self._seq, callback, args, context)) def advance(self, seconds): target = self._now + seconds while self._heap and self._heap[0][0] <= target: when, _, callback, args, context = heapq.heappop(self._heap) self._now = max(self._now, when) callback(*args) self._now = max(self._now, target)

这个基座写完之后，写超时测试就变成了：

def test_timeout_should_cancel_when_deadline_hit(): scheduler = VirtualScheduler(new_loop()) async def job(): try: await asyncio.sleep(10) return "normal" except asyncio.CancelledError: return "cancelled" task = scheduler.loop.create_task(job()) scheduler.advance(3) # 还没到点，任务还在跑 assert not task.done() scheduler.advance(7) # 时间到了，应该被取消 assert task.done() assert task.result() == "cancelled"

这个原理几乎每个协程框架都类似。Go 的golang.org/x/time/rate测试也讲究 fake clock；Java 的虚拟线程测试也有ScheduledExecutorService的替身。理解了"把调度器时钟换成可拨动的表盘"，你在任何语言里写异步测试都不会慌。

3.3 pytest 场景下的快速接入方案

如果你的项目用 pytest，不想全部重写基座，可以直接在 fixture 里接管事件循环，结合pytest-asyncio。关键配置是把event_loop_policy和event_loop_fixture做定制，然后在测试里用一个全局的VirtualScheduler实例维持时间线。

import pytest import asyncio @pytest.fixture def vloop(): loop = asyncio.new_event_loop() scheduler = VirtualScheduler(loop) yield scheduler loop.close() @pytest.mark.asyncio async def test_retry_logic(vloop): vloop.advance(60) # 模拟过去一分钟 # 这里就可以验证定时任务是否按时触发 ...

注意 pytest-asyncio 的event_loopfixture 默认绑定asyncio.get_event_loop_policy().new_event_loop()，如果你 override 了loopfixture，一定要把loop.set_debug(True)也带上。不然 debug 模式没开，调度警告不会出现在测试日志里，等于瞎了一只眼。

4. 超时、竞态与死锁：三类高频故障的复现手法

协程程序的崩溃率不高，但卡死率很高。我把日常踩过的坑归类成三类：超时处理不当、竞态条件触发、以及协程间的死锁。每一类都有一套复现和验证的手法。

4.1 超时不是"打断了就完事"

最典型的坑是asyncio.wait_for(..., timeout=5)：超时到了，wait_for抛出TimeoutError确实没错，但要注意——它默认会取消内部任务！如果你的业务代码finally里做了清理，那没问题；如果没做，取消信号打进去后，那个协程里正被await的资源可能处于"半开"状态。

另一个更隐蔽的坑是：当wait_for超时后，你只是普通地except掉TimeoutError，但底层协程已经在被取消的途中了。这时如果你还抱着"超时了但我可以继续用这个连接"的想法，那就是给自己埋雷。

正确做法是把超时取消当成一次显式的 CancelledError 处理，捕获后把资源的范畴状态重置干净：

async def fetch_with_timeout(url, timeout=5.0): try: return await asyncio.wait_for(http_get(url), timeout=timeout) except asyncio.TimeoutError: # 此时 http_get 内部已经被取消 # 必须确保连接池里没有残留半包 await cleanup_partial_connection() raise

测试验证超时行为，就用 3.2 的虚拟时钟。不用等真实 5 秒，直接把 clock 拨过截止时间，断言TimeoutError抛出、任务被取消、清理函数被调。这一步非常关键，因为真实时间的 CI 里超时测试一不稳定，你就分不清到底是代码坏了还是机器慢了。

4.2 竞态条件：故意让所有任务"撞"在一起

协程在同一线程里跑，并不会出现线程级的数据竞争，但会因为调度顺序产生逻辑竞态。比如两个协程同时queue.get()，你默认"先来的先得"，可调度顺序稍微一变，后者先执行了，逻辑就错了。

复现竞态的常用手法，是制造一个"起跑线"。我用asyncio.Barrier（或者自己用 Event 拼）把多个任务都卡在一个栅栏上，等所有任务都到齐了再同时放行。这样时间窗口被放大到可预测的宽度，谁先谁后一目了然。

import asyncio async def race_probe(): loop = asyncio.get_running_loop() barrier = asyncio.Barrier(3) async def worker(name): await barrier.wait() # 三个任务都到这里再继续 # 这里几乎同时往下走，适合复现竞态 await asyncio.sleep(0) # 手动让出一次调度，放大冲突窗口 print(f"{name} 继续执行") tasks = [asyncio.create_task(worker(n)) for n in ["A", "B", "C"]] await asyncio.gather(*tasks)

如果你猜"问题可能是两个任务同时修改一个共享字典"，那就可以在起跑线后面先造出脏读的条件，再断言结果。这类测试比随机 sleep 撞概率稳定得多。随机 sleep 只能复现"偶尔挂"，加了栅栏之后"必现"。

4.3 协程死锁：等待链往往跨任务

前面说的Event互等是最常见死锁。Task A 在等 Event E，Task B 在 Queue Q 上取数据，而 Q 的数据得由 A 处理完之后才放进去。从单个任务看永远等不到，但看全景就明白了。

复现死锁的最简单办法：先写一个没有死锁的版本，然后把其中一个await的顺序悄悄换掉。比如把"先锁后等"改成"先等后锁"。然后搭配 2.1 的全家福快照，在测试里断言：若干秒之后任务数量不减少、且多个任务的停靠点不变化，基本就可以判定死锁了。

但更实用的做法是写一个"看门狗"测试工具：

async def watchdog(loop, timeout=2.0): await asyncio.sleep(timeout) active = asyncio.all_tasks(loop) running = [t for t in active if not t.done()] if running: raise AssertionError(f"疑似死锁, 仍有 {len(running)} 个任务未完成: {running}")

把watchdog()和被测逻辑放进同一个gather()，如果过了 2 秒任务还没结束，就主动怼一个AssertionError。别小看这个粗暴的办法，很多协程卡死都是"悄无声息"的，断言能让你在 CI 里第一时间炸出来。

4.4 事件循环阻塞：一个最容易被忽略的罪魁祸首

有些"死锁"根本不是协程等协程，而是某个任务里写了一行同步阻塞调用，比如requests.get()或者time.sleep(1)。由于协程单线程串行，这个阻塞会卡住整个事件循环，其它协程的await全部干等。

识别方法：死锁发生时，用py-spy dump看主线程栈，如果停在_socket.socket.send之类的系统调用上，多半就是同步阻塞。测试里防这个的办法是给被测代码注入一个"最大执行时间限制"：跑一个实时监控协程，如果主任务的 CPU 时间增长但虚拟时间没推进，就可以断言"事件循环被阻塞了"。

5. 从零搭建协程模块的调试与测试清单

写到最后，我把这几年做协程模块沉淀下来的流程整理成一份可复制的清单。不算完整方法论，但每一条都是真实有用的。

5.1 调试环境的"叮咚配置"

1. 所有协程入口统一走@trace_task装饰器，日志级别定为INFO，线上可开关。
2. 开发环境和 CI 统一设PYTHONASYNCIODEBUG=1，让调度告警能在日志里出现。
3. 给进程挂一个SIGUSR1信号处理器，里面调snapshot_all_tasks()。
4. Task 创建时一定要create_task(coro, name="业务名_序号")，不然后期翻日志根本对不上号。
5. 日志字段最少包含：时间戳、任务名、事件类型、文件名行号、当前签名指标（队列长度、连接数等）。

5.2 测试的固定套路

1. 涉及延迟、超时的测试，一律使用虚拟时钟，不跑真实 sleep；真实睡眠只留给集成测试。
2. 涉及并发顺序，用Barrier或自定义栅栏拉通多个任务，放大竞态窗口。
3. 每个测试文件带一个watchdog守护，超过 2 秒无进展即断言失败。
4. pytest-asyncio的event_loopfixture 打开loop.set_debug(True)。
5. 测试命名带上故障类型词根，比如test_timeout_cancel_task、test_no_deadlock_on_queue_empty，方便回归时按类型筛查。

5.3 我自己的最小可运行模板

如果你只想要一个起步环境，复制下面这个骨架就够了：

项目结构 ├── async_app │ ├── __init__.py │ ├── tracing.py # trace_task 装饰器 │ ├── scheduler.py # 虚拟时钟实现 │ └── digest.py # 快照打印函数 └── tests ├── conftest.py └── test_timeouts.py

conftest.py的核心三行是：

@pytest.fixture(autouse=True) def debug_loop(event_loop): event_loop.set_debug(True) event_loop.slow_callback_duration = 0.05 return event_loop

把slow_callback_duration缩到 50 毫秒，事件循环里一旦有某次回调执行超过这个值，日志立刻打出告警。这在早期捕获同步阻塞非常有用。

最后说一个我个人的体会：协程调试最耗时的阶段往往不是定位问题本身，而是"确认自己看到了什么"。全家福快照一次打出来，几十个任务堆在几个点上，视觉冲击力比你翻一遍源码强太多。与其反复猜代码，不如先把这些尸体摆出来。我从第一次用snapshot_all_tasks()定位到一个诡异死锁之后，就再也没空手排查过协程卡死问题。这个习惯，保我平安度过了很多个发布前夜的紧急排查。