如何为Python多线程场景下的dataclass实现属性修改锁逻辑

最优实现方案 #

你原来的实现存在两个核心问题：

• 自定义的bool类型锁完全没有线程安全保障，布尔值的判断和修改本身不是原子操作，会出现竞态条件，根本起不到互斥作用
• 锁判断逻辑写反了：你写的是lock不为False的时候才允许修改属性，和加锁的预期逻辑完全相反

要实现修改属性时自动加锁、无需主动调用校验方法，可以结合Python的@property装饰器 + 标准库threading.Lock实现，代码如下：

from dataclasses import dataclass, field
import threading

@dataclass
class Status:
    # 内部私有变量存储实际开关值，不对外暴露
    _switch: bool = False
    # 自动初始化线程锁，无需外部传入
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)

    @property
    def switch(self) -> bool:
        # 读属性自动加锁，保证拿到的是最新值，允许脏读的话可以去掉锁
        with self._lock:
            return self._switch

    @switch.setter
    def switch(self, val: bool) -> None:
        # 写属性自动加锁，无需外部调用额外方法
        with self._lock:
            self._switch = val

使用方式 #

完全和你原来无锁的类用法一致，不需要修改任何外部调用逻辑：

# 初始化
status = Status()

# 读属性
print(status.switch) # 输出 False

# 写属性，自动触发加锁逻辑，不需要调用额外的set方法
status.switch = True
print(status.switch) # 输出 True

如果后续你需要给这个类加更多需要线程安全的属性，可以统一重写__setattr__方法处理所有属性的加锁逻辑，不需要逐个写setter。

内容的提问来源于stack exchange，提问作者user14603676