不能靠多跑几次验证线程安全，必须主动制造竞争窗口、控制调度节奏并检查最终一致性；需用 threading.event 等同步原语协调，避免 time.sleep，共享对象须每次测试重新初始化，最终调用 executor.shutdown(wait=true) 再断言。
如何用 pytest 模拟多线程竞争并触发竞态条件
直接说结论：不能靠“多跑几次看是否出错”来验证线程安全，必须主动制造竞争窗口、控制线程调度节奏，并检查共享状态的最终一致性。pytest 本身不提供线程同步原语，得靠 threading.Event、time.sleep（谨慎！）或 queue.Queue 来协调。
常见错误现象：AssertionError 偶发失败、数值统计对不上、AttributeError 报对象没有某个属性（其实是被另一个线程删了）、KeyError 在字典操作中出现——这些都可能是竞态未被覆盖的信号。
- 使用
concurrent.futures.ThreadPoolExecutor启动固定数量线程，比裸写threading.Thread更易管理生命周期 - 避免在测试里用
time.sleep()控制时序——它不可靠，受系统负载影响大；改用threading.Event显式等待“就绪”和“开始”两个信号 - 所有被测共享对象（如全局 dict、类实例属性）应在每个测试函数开头重新初始化，防止测试间污染
- 最终断言前加
executor.shutdown(wait=True)，确保所有线程真正结束再检查状态
为什么 pytest.mark.parametrize 不能直接用于多线程测试用例
因为 parametrize 是 pytest 在收集阶段生成多个独立测试项，每个测试运行在单独的函数调用栈中；而多线程竞争必须发生在**同一个共享作用域内**。如果你写：
@pytest.mark.parametrize("n", [2, 4, 8])
def test_counter_race(n):
# 这里每个 n 对应一个全新 counter 实例，根本没共享！
counter = Counter()
# ... 启动 n 个线程操作它
结果就是：每个测试都在隔离环境里跑，永远测不出问题。
- 正确做法是把线程数作为参数传入，但共享对象（如
counter）必须在测试函数内部创建一次，再由所有线程共用 - 若需对比不同并发度的影响，用普通 for 循环 + 多次 assert，而不是依赖 parametrize 生成测试节点
- 注意：pytest 的
scope="function"fixture 默认每次测试新建实例，不适合做跨线程共享载体；要用scope="session"或干脆不用 fixture 管理共享对象
怎样让测试稳定复现竞态，而不是“十次跑三次崩”
关键不是增加线程数，而是延长临界区、插入可控的停顿点。Python 的 GIL 并不保证原子性，尤其对复合操作（如 counter.value += 1）——它等价于读-改-写三步，中间可被切换。
- 在被测代码的关键位置（比如递增前/后）插入
threading.Event.wait()，配合外部控制释放时机，强制制造“读完还没写”的窗口 - 用
sys.setswitchinterval(1e-6)缩小线程切换间隔（仅限 CPython，且仅影响字节码级别切换，对 I/O 阻塞无效），能提高竞争概率 - 不要只测最终值是否等于预期总数；要检查中间过程，比如用
queue.Queue收集每轮操作的原始输入和输出，验证顺序或重复写入 - 真实场景中，数据库连接池、缓存更新、日志计数器最容易暴露问题——优先拿这类逻辑写测试，别从玩具 Counter 开始
哪些 Python 内置类型看似线程安全，实则不然
list.append() 和 dict[key] = value 在 CPython 中确实是原子的（得益于 GIL），但只要涉及“先查后改”，就立刻掉出安全区。例如：
# 看似简单，实则危险
if key not in my_dict:
my_dict[key] = default_value # 两步：查 + 写，中间可能被其他线程插入
同理，counter += 1、my_list.pop()（当 list 可能为空时）、len(my_dict) == 0 后清空——全是经典坑点。
-
collections.deque的append()/popleft()是原子的，适合做线程间消息传递，但不适用于需要条件判断的场景 -
queue.Queue是真正为多线程设计的，所有操作都带锁，但它的存在意义是解耦，不是替代共享变量 - 用
threading.Lock最直白，但容易漏加、死锁或锁粒度太粗；threading.RLock允许同一线程多次 acquire，适合递归调用场景 - 别迷信
@functools.lru_cache—— 它的缓存字典本身不是线程安全的，高并发下可能损坏内部结构
最常被忽略的是：即使你用了锁，如果锁保护的范围没包住整个逻辑单元（比如只锁了写，没锁前面的判断），照样出问题。线程安全不是加了锁就万事大吉，是看“共享状态的不变量是否在整个操作序列中始终成立”。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
