并发代码如何稳定测试而不是靠运气

验证一:「绿」的测试不代表没有竞态

有竞态的存款 def deposit_racy(self, n): cur = self.balance # 读 time.sleep(0) # 让出一次 GIL: 制造线程切换的机会 self.balance = cur + n # 写

  在读和写之间加一个线程让出点(time.sleep(0)); 4 个线程各存 200 次 1 元, 期望 800; 连续 10 次结果: [228, 269, 240, 249, 238, 272, 200, 230, 200, 247]
  10 次里恰好等于 800 的次数: 0
  去掉让出点、自然运行(同样的「先读后写」逻辑), 10 次结果: [800, 800, 800, 800, 800, 800, 800, 800, 800, 800] ← 每次都是 800: 测试全绿, 但竞态并没有消失

4 个线程各存 200 次。自然运行(没有让出点,同样的先读后写逻辑)10 次全部是 800,测试全绿;加一个让出点后,10 次结果都是 200 多,没有一次正确。竞态一直在,只是线程恰好没在读写之间被切走。所以「跑了 100 次都通过」并不能证明代码线程安全。(这份代码在自由线程构建里更容易自然出错,见第 91 篇。)

验证二:调小切换间隔——我没能复现

  5.0 ms(默认切换间隔): 300 次里出错 0 次 | 切换间隔调到 0.001 ms: 300 次里出错 0 次
  (这份 box[0] += 1 在 GIL 版 3.14 上很难自然出错; 结果取决于解释器版本和机器, 想要稳定复现需要用别的手段)

一个常听到的技巧是 sys.setswitchinterval(1e-6) 让线程更频繁地切换,以放大竞态。我把它调到 0.001 ms,重复 300 次 box[0] += 1,在这台机器、GIL 版 3.14 上出错次数依然是 0。也就是说这一招不是万能的——结果取决于解释器版本和具体代码。别把它当成可靠的复现手段,要用下面的办法。

验证三:排演一次特定的交错

用 Event 强制顺序 def deposit(name, wait_evt, set_evt): cur = a.balance; log.append(f"{name} 读到 {cur}") if set_evt: set_evt.set() # 通知对方: 我读完了 if wait_evt: wait_evt.wait() # 等对方写完我再写 a.balance = cur + 1

  用两个 Event 强制「A 读 → B 读 → B 写 → A 写」的顺序:
     线程A 读到 0
     线程B 读到 0
     线程B 写入 1
     线程A 写入 1
  两次存款后余额 = 1 (期望 2) ← 每次都稳定复现丢失更新, 不靠运气
  加锁之后同样两次存款: 2 (期望 2)

不靠运气,直接安排「A 读 → B 读 → B 写 → A 写」这个最坏的顺序:两个线程分别在关键点上等待对方发来的 Event。这样两次存款后余额每次都是 1,可以稳定复现,写成回归测试。给同样的存款加上锁之后,同样两次存款得到 2。这种「把线程调度写进测试」的做法需要你知道有问题的交错是哪种,但换来的是确定性。

验证四:超时是并发测试的安全网

  消费者线程等一个永远不会到的消息: join(timeout=0.5) 在 0.51 s 后返回, 线程仍存活 = True → 测试可以据此明确失败, 而不是把 CI 卡死
  没有超时保护的死等, 由外部 timeout 在 1.01 s 杀掉(subprocess.run(timeout=1.0))

并发代码最坏的失败方式不是断言失败,而是永远卡住。消费者线程在等一个永远不会到的消息:如果测试用无超时的 join(),整个 CI 就挂了;用 join(timeout=0.5),0.51 s 后返回,线程仍存活,测试可以据此明确失败。整个进程级别也要有兜底(subprocess.run(timeout=…),或 pytest 的超时插件)。

验证五:轮询代替固定 sleep

  后台任务实际需要 20 ms: 轮询版测试等了 21.2 ms 就通过; 「固定 sleep(0.2)」的测试要等 200 ms
  后台任务实际需要 50 ms: 轮询版测试等了 57.2 ms 就通过; 「固定 sleep(0.2)」的测试要等 200 ms
  后台任务实际需要 100 ms: 轮询版测试等了 109.1 ms 就通过; 「固定 sleep(0.2)」的测试要等 200 ms

要等后台任务完成,常见的写法是 time.sleep(0.2) 再断言。这种写法慢(永远等 200 ms)而且不稳(任务某次多花了 250 ms 就失败)。写一个 eventually(cond, timeout) 轮询:条件一成立就通过——任务需要 20、50、100 ms 时分别只等了约 21、57、109 ms;条件永远不成立才走到超时。这样既快,又不会因为机器偶尔慢而失败。

分步演示:从「碰运气」到「确定性」