Python中types.SimpleNamespace是否线程安全？原子性验证

问题 #

请问Python中的types.SimpleNamespace是否具备线程安全性？若拥有固定属性的types.SimpleNamespace对象，跨线程进行属性更新操作是否具备原子性？

以下为测试用示例代码：

import threading
import uuid
from pprint import pprint
import time
from types import SimpleNamespace

workers_count = 10
total_post_per_user = 5
storage = SimpleNamespace(
    user_count=0,
    users=set(),
    posts=[],
    latest_message=None
)

def add_post(username: str, post: str):
    storage.posts.append(post)
    storage.latest_message = post
    print(f"User {username} posted : {post}")

def worker():
    storage.user_count += 1
    user_name = f'worker_{storage.user_count}'
    storage.users.add(user_name)
    print(f"User {user_name} joined, total users {storage.user_count}")

    setattr(storage, user_name, user_name)  # 动态添加属性到SimpleNamespace

    for i in range(total_post_per_user):
        add_post(user_name, f"Message from {user_name} N {i}, message : {str(uuid.uuid4())}")
        time.sleep(.2)

workers = [
    threading.Thread(target=worker, daemon=True) for _ in range(workers_count)
]
for worker in workers:
    worker.start()
for worker in workers:
    worker.join()

print("Results")
print(f"-- User count : {storage.user_count}")
print(f"-- Users : {len(storage.users)} = {storage.users}")
print(f"-- Total posts: {len(storage.posts)} - expected : {workers_count * total_post_per_user}")
print(f"-- Last message: {storage.latest_message}")
pprint(storage)

运行结果看似线程安全，但需要确认实际情况，若并非如此请说明原因。

核心结论与分析 #

• SimpleNamespace本身无线程安全性：它只是一个轻量的属性容器，内部没有任何线程同步逻辑，完全依赖外部保障线程安全。
• 单个属性赋值的原子性来自GIL：像storage.latest_message = post这类简单赋值操作，在CPython中是原子性的——这是全局解释器锁（GIL）的特性，不是SimpleNamespace独有的。但注意，这仅针对单个赋值操作，复合操作不适用。
• 复合操作存在竞态条件：示例中的storage.user_count += 1是典型的非原子操作，它拆分为「读取当前值→计算加1→写入新值」三个步骤，多线程并发时会出现多个线程同时读取同一个值，最终导致计数结果小于预期。
• 内部数据结构本身不安全：示例中使用的set（storage.users）和list（storage.posts）都不是线程安全的数据结构，它们的add、append操作在并发场景下可能导致数据丢失、重复甚至异常。
• 测试结果的偶然性：你看到的“线程安全”结果只是因为time.sleep(.2)降低了并发冲突的概率，属于偶然情况，不能代表实际生产环境中的表现。

线程安全的改造方案 #

要在多线程环境下安全使用，必须手动添加同步机制，比如用threading.Lock保护所有对SimpleNamespace及其内部属性的访问：

import threading
import uuid
from pprint import pprint
import time
from types import SimpleNamespace

workers_count = 10
total_post_per_user = 5
storage = SimpleNamespace(
    user_count=0,
    users=set(),
    posts=[],
    latest_message=None
)
# 定义全局锁
lock = threading.Lock()

def add_post(username: str, post: str):
    with lock:
        storage.posts.append(post)
        storage.latest_message = post
    print(f"User {username} posted : {post}")

def worker():
    with lock:
        storage.user_count += 1
        user_name = f'worker_{storage.user_count}'
        storage.users.add(user_name)
        setattr(storage, user_name, user_name)
    print(f"User {user_name} joined, total users {storage.user_count}")

    for i in range(total_post_per_user):
        add_post(user_name, f"Message from {user_name} N {i}, message : {str(uuid.uuid4())}")
        time.sleep(.2)

workers = [
    threading.Thread(target=worker, daemon=True) for _ in range(workers_count)
]
for worker in workers:
    worker.start()
for worker in workers:
    worker.join()

print("Results")
print(f"-- User count : {storage.user_count}")
print(f"-- Users : {len(storage.users)} = {storage.users}")
print(f"-- Total posts: {len(storage.posts)} - expected : {workers_count * total_post_per_user}")
print(f"-- Last message: {storage.latest_message}")
pprint(storage)

内容的提问来源于stack exchange，提问作者Federico Baù