pytest中直接启动线程会导致测试失效,因未等待线程结束、未捕获子线程异常、未同步执行时机;须用Event控制并发、join等待、异常收集来确保线程安全测试有效。

pytest里直接启动线程会“测不准”
因为 pytest 本身是单线程执行测试函数的,t.start()一调用就返回,pytest 立刻判定测试通过,完全不管后台线程是否跑完、有没有触发竞态。你看到 assert 通过了,其实可能三个线程刚创建完实例就退出了——根本没暴露问题。
必须显式等线程结束,否则测试无效。常见错误包括:
- 只写
t.start()不跟t.join(timeout=) - 用
time.sleep(0.1)代替同步,既不可靠又难复现问题 - 没捕获子线程异常,导致失败静默消失(
sys.excepthook在子线程里不生效)
用 threading.Event 控制并发节奏
想稳定复现竞态,不能靠“运气”,得让所有线程在关键操作前统一就位,再同时放开。比如测试单例类 Logger 的 __new__ 是否被多次执行,可以这样写:
import threading
<p>def test_logger_thread_safety():
logger_instances = []
ready = threading.Event()
done = threading.Event()</p><pre class='brush:python;toolbar:false;'>def worker():
ready.wait() # 所有线程卡在这里
inst = Logger() # 真正触发单例逻辑
logger_instances.append(inst)
done.set()
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads:
t.start()
ready.set() # 同时放开全部线程
for t in threads:
t.join(timeout=3) # 必须加超时,防死锁
# 检查是否真只有一个实例
assert len(set(id(inst) for inst in logger_instances)) == 1</pre>这里 ready.wait() 和 ready.set() 是关键:它把“多个线程几乎同时进入 __new__”这个条件从概率事件变成确定行为。
立即学习“Python免费学习笔记(深入)”;
捕获线程内异常并共享状态
子线程抛异常不会让 pytest 报错,你得自己兜底。最简单办法是用一个可变容器(如 list)收集异常,或用 threading.local() 存每线程状态:
- 在
worker函数里包一层try/except,把sys.exc_info()追加到全局errors列表 - 检查
logger_instances长度是否等于线程数,少于预期说明某些线程崩溃了但没报出来 - 避免在
__new__或__init__里做耗时/易错操作(比如读文件、连数据库),否则测试更难稳定
模块级单例不用测多线程安全
如果你用的是模块级单例(比如 config.py 里直接写 instance = MyClass()),那根本不用写上面那些测试——Python 解释器保证模块只导入一次,且顶层代码执行是原子的。所有线程拿到的都是同一个对象,没有竞态窗口。
但要注意:这种写法要求初始化逻辑不依赖运行时参数;如果测试中需要重置实例(比如改环境变量后重建),就得用 importlib.reload() 配合,否则模块缓存会让测试失效。
真正容易出问题的是那些“看着像单例、实则没加锁”的 __new__ 或装饰器实现——它们在线程调度稍有交错时就会漏掉判断,生成多个实例。测的时候别只看最终结果,要盯住中间过程是否被破坏。


















