最直接的方式是用 threading.lock 包裹普通集合；python 标准库无内置线程安全集合，list/set/dict 非原子操作，需显式加锁实现 threadsafeset，避免竞态与异常。
用 threading.Lock 包裹普通集合是最直接的方式
Python 标准库没有内置线程安全的集合类型，list、set、dict 本身都不是原子操作。比如 my_set.add(x) 看似简单，实际包含“查重→插入→调整哈希表”多个步骤，多线程下可能丢失元素或触发 KeyError / RuntimeError。
最稳妥的做法是显式加锁：
import threading
<p>class ThreadSafeSet:
def <strong>init</strong>(self):
self._set = set()
self._lock = threading.Lock()</p><pre class="brush:python;toolbar:false;">def add(self, item):
with self._lock:
self._set.add(item)
def discard(self, item):
with self._lock:
self._set.discard(item)
def __contains__(self, item):
with self._lock:
return item in self._set
def __len__(self):
with self._lock:
return len(self._set)
注意：不要在锁内做耗时操作（如网络请求、文件读写），否则会严重拖慢并发性能；也不要尝试在锁内调用可能再次获取同一把锁的方法（避免死锁）。
为什么不用 queue.Queue 替代？
queue.Queue 是线程安全的，但它设计目标是生产者-消费者模型，不支持随机访问、成员判断（in）、遍历或去重语义。它没有 __contains__ 方法，也不能直接用 for x in q: 遍历——必须用 q.get_nowait() 消费，且消费后元素就没了。
如果你实际需要的是“能并发增删查、但不关心顺序”的集合行为，硬套 Queue 只会让逻辑变绕、易出错。常见误用包括：
- 用
Queue存储唯一 ID，却反复get()再put()回去模拟“查是否存在”，导致竞态和重复消费 - 为支持
in操作，额外维护一个set，又没加锁同步，反而更不安全
高并发读多写少场景可考虑 threading.RLock + 读写分离
如果读操作远多于写（比如配置白名单校验），用单把 Lock 会让所有读请求排队，降低吞吐。此时可用 threading.RLock 配合手动读写标记，但 Python 标准库不提供原生读写锁，需自行控制：
更现实的选择是：读操作仍用普通 set 快照 + Lock 保护写，例如：
def snapshot(self):
with self._lock:
return set(self._set) # 返回副本，读时不加锁
<p>def contains(self, item):</p><h1>先快照再查，适合容忍短暂延迟的场景</h1><pre class="brush:python;toolbar:false;">return item in self.snapshot()
这种方式牺牲了实时性（可能查到几毫秒前的状态），但极大提升读吞吐。关键点在于：快照必须在锁内完成，且返回的是副本（不能返回 self._set.copy() 后被外部修改）。
第三方库 concurrent.futures 不解决集合线程安全问题
有人误以为用 ThreadPoolExecutor 就自动线程安全了。其实它只管理任务调度，对数据结构毫无保护作用。下面这段代码依然会丢数据：
from concurrent.futures import ThreadPoolExecutor
shared_set = set()
def worker(x):
shared_set.add(x) # ❌ 仍然竞态！
<p>with ThreadPoolExecutor() as ex:
ex.map(worker, range(1000))</p>
真正要安全，要么把 shared_set 换成上面封装的 ThreadSafeSet，要么把 add 操作移到锁保护范围内——执行器本身不提供数据同步能力。
复杂点在于：线程安全从来不是“选个高级类就完事”，而是明确读写边界、控制临界区粒度、并接受一致性与性能之间的权衡。很多 bug 出现在“我以为这个操作是原子的”这种假设上。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
