Python 并发程序为何难以测试？

2026-01-18 00:00:00 作者：舞夢輝影

Python并发程序难测试的核心在于不确定性，包括调度顺序、竞态条件难以稳定触发、事件循环管理复杂、时间相关行为不可控及错误传播路径被遮蔽等问题。

Python 并发程序难测试，核心在于**不确定性**——线程/协程的调度顺序、资源竞争时机、超时边界等无法稳定复现，导致测试结果时而通过、时而失败（flaky test）。

多个线程或协程同时读写共享状态（如全局变量、列表、字典）时，执行顺序稍有不同，就可能跳过 bug 或恰好暴露问题。测试无法控制底层调度器何时切换上下文，所以即使加了 time.sleep() 也很难让问题每次都出现。

用 threading.Lock 或 asyncio.Lock 保护临界区是修复手段，但测试本身很难“逼出”未加锁时的异常行为
工具如 pytest-xdist 并行跑测试反而可能掩盖竞态——因为每次调度路径都不同
静态检查（如 pylint 的 consider-using-with）或动态检测（如 thread-sanitizer 风格工具在 Python 中支持有限）作用有限

async/await 代码依赖事件循环运行，测试时若不显式管理循环生命周期，容易出现 “RuntimeError: no running event loop” 或任务被静默丢弃。

直接调用协程函数（如
my_coro()）只返回协程对象，不会执行；必须用 asyncio.run() 或 loop.run_until_complete()
多个测试共用一个事件循环时，未 await 完的任务可能污染后续测试；推荐每个测试用独立循环（asyncio.new_event_loop()）或使用 pytest-asyncio 自动管理
模拟异步依赖（如 HTTP 请求）需用 AsyncMock 或 respx 等支持协程的 mock 工具，普通 Mock 会报错

超时、重试、定时任务（asyncio.sleep、threading.Timer）让测试变慢且不稳定。真实等待几秒既低效又易受系统负载影响。

避免在测试中用 time.sleep(1) 等待，改用可注入的“时间提供者”（如传入 clock 参数），测试时替换为立即返回的模拟时钟
对 asyncio.sleep()，可用 asyncio.set_event_loop_policy() 替换为快进版循环（如 aiounittest.FastPolicy），或打补丁让其跳过等待
涉及系统时间的逻辑（如 JWT 过期判断），应抽象出 get_now() 接口，测试时固定返回某时刻

线程中未捕获异常会静默消失；asyncio 中未 await 的任务抛出异常可能只打印到 stderr 而不中断测试；信号、超时、取消操作的副作用也难以断言。

线程中用 threading.excepthook 捕获未处理异常，并在测试中检查是否被调用
asyncio 中确保关键任务被 asyncio.create_task() 启动后，显式 await task 或用 asyncio.wait_for(task, timeout) 控制生命周期
测试取消逻辑时，不能只检查“是否调用了 cancel()”，还要验证目标协程是否真正响应了 CancelledError 并清理资源

不复杂但容易忽略：并发测试不是写得更多，而是写得更“确定”。关键是把不确定的部分（时间、调度、外部依赖）抽出来，替换成可控的抽象，再针对逻辑本身做单元验证。