线程安全

副标题 / 摘要 单例常见但容易出错，尤其在并发场景。本文给出线程安全的实现思路与注意事项。 目标读者 使用单例的后端工程师 关注并发正确性的开发者 负责核心基础组件的团队 背景 / 动机 单例常用于配置、连接池或缓存管理。 错误实现会导致多实例或竞态问题。 核心概念 单例：全局唯一实例 线程安全：并发访问不破坏一致性 延迟初始化：按需创建实例 实践指南 / 步骤 优先使用语言内建的单例机制 需要懒加载时使用锁或原子操作 避免在构造函数中做重逻辑 对并发访问写压力测试 可运行示例 import threading class Singleton: _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance if __name__ == "__main__": objs = [] for _ in range(3): objs.append(Singleton()) print(len(set(id(x) for x in objs))) # 1 解释与原理 双重检查锁避免了每次创建都加锁的开销，同时保证并发下只生成一个实例。 关键是“锁内再检查”，避免竞态。 常见问题与注意事项 单例会带来全局状态污染吗？ 会，因此应谨慎使用。 ...

副标题 / 摘要 竞态条件是并发中最隐蔽的 bug 类型之一，源于对共享状态的非原子操作。本文给出可运行示例与规避方法。 目标读者 需要理解线程安全的工程师 做并发开发与性能优化的开发者 负责可靠性与稳定性的技术负责人 背景 / 动机 竞态条件会导致偶发错误：你很难复现，但它确实存在。 理解其成因，才能正确使用锁、原子操作与无锁结构。 核心概念 共享状态：多线程同时访问的变量 非原子操作：读-改-写不是一个不可分割的步骤 临界区：必须串行访问的代码区域 实践指南 / 步骤 识别共享状态 确定临界区 使用锁或原子操作保护 最小化临界区范围 用竞态检测工具验证 可运行示例 import threading counter = 0 def inc(): global counter for _ in range(100000): counter += 1 if __name__ == "__main__": t1 = threading.Thread(target=inc) t2 = threading.Thread(target=inc) t1.start() t2.start() t1.join() t2.join() print(counter) 在部分解释器/实现中可能输出小于 200000，这就是竞态。 解释与原理 counter += 1 并不是原子操作，它包含读取、加一、写回三个步骤。 两个线程交错执行时会丢失更新。 ...

副标题 / 摘要 死锁发生在多个线程互相等待对方持有的资源。本文解释成因、给出示例，并总结规避方法。 目标读者 需要理解并发风险的工程师 负责系统可靠性的开发者 做并发设计评审的技术负责人 背景 / 动机 死锁通常在压力或线上才出现，一旦发生会导致服务完全卡住。 理解死锁条件并提前规避，是并发系统的必修课。 核心概念 互斥：资源一次只能被一个线程占用 占有且等待：拿着资源还要等另一个资源 不可抢占：资源不能被强制夺取 循环等待：形成等待环 实践指南 / 步骤 统一锁顺序（按固定顺序获取） 减少锁数量 设置超时与回退 用锁分层减少交叉等待 监控阻塞与持锁时间 可运行示例 import threading import time lock_a = threading.Lock() lock_b = threading.Lock() def t1(): with lock_a: time.sleep(0.1) with lock_b: pass def t2(): with lock_b: time.sleep(0.1) with lock_a: pass if __name__ == "__main__": threading.Thread(target=t1).start() threading.Thread(target=t2).start() time.sleep(1) print("可能已死锁") 解释与原理 当 t1 拿着 A 等 B，而 t2 拿着 B 等 A，就形成循环等待。 满足四个必要条件时死锁发生。 ...