import threading
from typing import Union
class AtomicNum:
"""
线程安全的原子数字类
提供原子性的数值操作,确保在多线程环境下的数据一致性。
所有操作都通过内部锁保证线程安全。
Attributes:
_value: 内部存储的数值
_lock: 线程锁,保证操作的原子性
Example:
>>> atomic = AtomicNum(10)
>>> atomic.get()
10
>>> atomic.compare_and_set(10, 20)
True
>>> atomic.get()
20
"""
def __init__(self, initial_value: Union[int, float] = 0) -> None:
"""
初始化原子数字
:param initial_value: 初始值,默认为0
:raises TypeError: 如果初始值不是数字类型
"""
if not isinstance(initial_value, (int, float)):
raise TypeError(f"initial_value must be int or float, got {type(initial_value).__name__}")
self._value: Union[int, float] = initial_value
self._lock = threading.Lock()
def compare_and_set(self, expect: Union[int, float], update: Union[int, float]) -> bool:
"""
CAS(Compare-And-Swap)操作
原子性地比较当前值与期望值,如果相等则更新。
保证CAS语义:
- 返回 True 当且仅当当前值等于期望值且成功更新
- 返回 False 当且仅当当前值不等于期望值 **或** 获取锁失败
设计说明:使用 try-lock 模式(blocking=False)
- 如果锁空闲:拿锁,比较并更新
- 如果锁被其他线程占用:直接返回 False,由调用方重试
- 原理:如果锁被占用,说明值正在被修改,expect 大概率已过期,等待也无意义
这种乐观并发控制减少了无意义的等待,在高并发场景下性能更好。
单次 CAS 失败不影响正确性,调用方(如令牌桶刷新)会在下次调用重试。
:param expect: 期望的当前值
:param update: 要更新的新值
:return: 是否成功更新
:raises TypeError: 如果参数不是数字类型
Example:
>>> atomic = AtomicNum(5)
>>> atomic.compare_and_set(5, 10)
True
>>> atomic.compare_and_set(5, 15)
False
"""
if not isinstance(expect, (int, float)):
raise TypeError(f"expect must be int or float, got {type(expect).__name__}")
if not isinstance(update, (int, float)):
raise TypeError(f"update must be int or float, got {type(update).__name__}")
if self._lock.acquire(blocking=False):
try:
if self._value == expect:
self._value = update
return True
return False
finally:
self._lock.release()
return False
def get_and_set(self, new_value: Union[int, float]) -> Union[int, float]:
"""
原子性地设置新值并返回旧值
:param new_value: 要设置的新值
:return: 设置前的旧值
:raises TypeError: 如果参数不是数字类型
Example:
>>> atomic = AtomicNum(20)
>>> atomic.get_and_set(30)
20
>>> atomic.get()
30
"""
if not isinstance(new_value, (int, float)):
raise TypeError(f"new_value must be int or float, got {type(new_value).__name__}")
with self._lock:
result = self._value
self._value = new_value
return result
def get(self) -> Union[int, float]:
"""
获取当前值
:return: 当前存储的值
Example:
>>> atomic = AtomicNum(10)
>>> atomic.get()
10
"""
with self._lock:
return self._value
def set(self, new_value: Union[int, float]) -> None:
"""
设置新值
:param new_value: 要设置的新值
:raises TypeError: 如果参数不是数字类型
Example:
>>> atomic = AtomicNum(15)
>>> atomic.set(25)
>>> atomic.get()
25
"""
if not isinstance(new_value, (int, float)):
raise TypeError(f"new_value must be int or float, got {type(new_value).__name__}")
with self._lock:
self._value = new_value
def add(self, delta: Union[int, float]) -> None:
"""
原子性地增加指定值
:param delta: 要增加的值(可以为负数实现减法)
:raises TypeError: 如果参数不是数字类型
Example:
>>> atomic = AtomicNum(10)
>>> atomic.add(5)
>>> atomic.get()
15
>>> atomic.add(-3)
>>> atomic.get()
12
"""
if not isinstance(delta, (int, float)):
raise TypeError(f"delta must be int or float, got {type(delta).__name__}")
with self._lock:
self._value += delta
def try_decrement(self, num: Union[int, float]) -> bool:
"""
尝试原子性地扣减指定值
只有当前值大于等于要扣减的值时才执行扣减操作。
:param num: 要扣减的值
:return: 是否成功扣减
:raises TypeError: 如果参数不是数字类型
:raises ValueError: 如果参数为负数
Example:
>>> atomic = AtomicNum(20)
>>> atomic.try_decrement(5)
True
>>> atomic.get()
15
>>> atomic.try_decrement(20)
False
>>> atomic.get()
15
"""
if not isinstance(num, (int, float)):
raise TypeError(f"num must be int or float, got {type(num).__name__}")
if num < 0:
raise ValueError(f"num must be non-negative, got {num}")
with self._lock:
if self._value >= num:
self._value = self._value - num
return True
return False
def decrement_to_zero(self, num: Union[int, float]) -> None:
"""
强制扣减指定值,最小扣减至0
如果当前值小于要扣减的值,则将值设为0。
:param num: 要扣减的值
:raises TypeError: 如果参数不是数字类型
:raises ValueError: 如果参数为负数
Example:
>>> atomic = AtomicNum(10)
>>> atomic.decrement_to_zero(3)
>>> atomic.get()
7
>>> atomic.decrement_to_zero(10)
>>> atomic.get()
0
"""
if not isinstance(num, (int, float)):
raise TypeError(f"num must be int or float, got {type(num).__name__}")
if num < 0:
raise ValueError(f"num must be non-negative, got {num}")
with self._lock:
if self._value >= num:
self._value = self._value - num
else:
self._value = 0
def add_and_limit(self, delta: Union[int, float], limit: Union[int, float]) -> None:
"""
原子性地增加指定值,但不超过上限
:param delta: 要增加的值
:param limit: 值的上限
:raises TypeError: 如果参数不是数字类型
:raises ValueError: 如果limit为负数
Example:
>>> atomic = AtomicNum(5)
>>> atomic.add_and_limit(3, 10)
>>> atomic.get()
8
>>> atomic.add_and_limit(5, 10)
>>> atomic.get()
10
"""
if not isinstance(delta, (int, float)):
raise TypeError(f"delta must be int or float, got {type(delta).__name__}")
if not isinstance(limit, (int, float)):
raise TypeError(f"limit must be int or float, got {type(limit).__name__}")
with self._lock:
self._value = min(self._value + delta, limit)
def dec_and_limit(self, decrement: Union[int, float], limit: Union[int, float]) -> None:
"""
原子性地减少指定值,但不低于下限
:param decrement: 要减少的值
:param limit: 值的下限
:raises TypeError: 如果参数不是数字类型
Example:
>>> atomic = AtomicNum(10)
>>> atomic.dec_and_limit(3, 5)
>>> atomic.get()
7
>>> atomic.dec_and_limit(5, 5)
>>> atomic.get()
5
"""
if not isinstance(decrement, (int, float)):
raise TypeError(f"decrement must be int or float, got {type(decrement).__name__}")
if not isinstance(limit, (int, float)):
raise TypeError(f"limit must be int or float, got {type(limit).__name__}")
with self._lock:
self._value = max(self._value - decrement, limit)
def multiply(self, multiplier: Union[int, float]) -> None:
"""
原子性地乘以指定倍数
:param multiplier: 乘数
:raises TypeError: 如果参数不是数字类型
Example:
>>> atomic = AtomicNum(5)
>>> atomic.multiply(2)
>>> atomic.get()
10
>>> atomic.multiply(3)
>>> atomic.get()
30
"""
if not isinstance(multiplier, (int, float)):
raise TypeError(f"multiplier must be int or float, got {type(multiplier).__name__}")
with self._lock:
self._value *= multiplier
def increment(self) -> Union[int, float]:
"""
原子性地自增1并返回新值
:return: 自增后的新值
Example:
>>> atomic = AtomicNum(5)
>>> atomic.increment()
6
"""
with self._lock:
self._value += 1
return self._value
def decrement(self) -> Union[int, float]:
"""
原子性地自减1并返回新值
:return: 自减后的新值
Example:
>>> atomic = AtomicNum(5)
>>> atomic.decrement()
4
"""
with self._lock:
self._value -= 1
return self._value
def __repr__(self) -> str:
"""
返回对象的字符串表示
:return: 包含当前值的字符串表示
"""
return f"AtomicNum({self.get()})"
def __str__(self) -> str:
"""
返回当前值的字符串表示
:return: 当前值的字符串表示
"""
return str(self.get())
def __eq__(self, other: object) -> bool:
"""
比较当前值与另一个值是否相等
:param other: 要比较的对象
:return: 是否相等
"""
if isinstance(other, AtomicNum):
return self.get() == other.get()
if isinstance(other, (int, float)):
return self.get() == other
return False
def __lt__(self, other: Union[int, float, "AtomicNum"]) -> bool:
"""
比较当前值是否小于另一个值
:param other: 要比较的对象
:return: 是否小于
"""
if isinstance(other, AtomicNum):
return self.get() < other.get()
return self.get() < other
def __le__(self, other: Union[int, float, "AtomicNum"]) -> bool:
"""
比较当前值是否小于等于另一个值
:param other: 要比较的对象
:return: 是否小于等于
"""
if isinstance(other, AtomicNum):
return self.get() <= other.get()
return self.get() <= other
def __gt__(self, other: Union[int, float, "AtomicNum"]) -> bool:
"""
比较当前值是否大于另一个值
:param other: 要比较的对象
:return: 是否大于
"""
if isinstance(other, AtomicNum):
return self.get() > other.get()
return self.get() > other
def __ge__(self, other: Union[int, float, "AtomicNum"]) -> bool:
"""
比较当前值是否大于等于另一个值
:param other: 要比较的对象
:return: 是否大于等于
"""
if isinstance(other, AtomicNum):
return self.get() >= other.get()
return self.get() >= other
def __int__(self) -> int:
"""
转换为整数
:return: 当前值的整数表示
"""
return int(self.get())
def __float__(self) -> float:
"""
转换为浮点数
:return: 当前值的浮点数表示
"""
return float(self.get())