为attrs.frozen冻结类的__new__添加缓存实现，咨询潜在风险

问题背景 #

我有一个被attrs.frozen装饰的冻结类，程序需要创建该类的大量实例。由于这些实例是不可变的且仅用于读取操作，我希望添加缓存机制，使得使用相同参数创建的对象直接从缓存获取而非重新创建。已完成实现，但不确定方案是否存在潜在问题。

现有实现代码 #

from typing import ClassVar

from attrs import frozen

@frozen
class Cached:
    _cache: ClassVar[dict[str, "Cached"]] = {}

    name: str
    config: int

    def __new__(cls, *args, **kwds):
        key = str(args) + str(kwds)

        if (instance:=cls._cache.get(key)) is not None:
            print("cache misses")
            return instance

        print("cache hits")
        instance = super().__new__(cls)
        instance.__init__(*args, **kwds)
        cls._cache[key] = instance
        return instance

运行结果（IPython） #

In [1]: from cached_new import Cached

In [2]: c0 = Cached("cache", 0)
cache hits

In [3]: c1 = Cached("cache", 0)
cache misses

In [4]: c0 is c1
Out[4]: True

In [5]: c2 = Cached("", 1)
cache hits

In [6]: c3 = Cached("", 1)
cache misses

In [7]: c2 is c3
Out[7]: True

潜在问题分析 #

1. 缓存键冲突风险 #

当前用str(args) + str(kwds)生成缓存键，存在明显缺陷：

• 不同参数组合可能生成相同字符串，比如Cached(1, 2)和Cached("(1, 2)", "")会得到完全一致的键，导致缓存错误匹配。
• 位置参数与等价关键字参数无法被识别，比如Cached("cache", 0)和Cached(name="cache", config=0)会生成不同键，导致重复创建实例，缓存失效。

2. 内存泄漏风险 #

缓存使用ClassVar的普通字典存储实例引用，只要类存在，所有缓存实例都不会被垃圾回收。若实例数量庞大或持有大对象/外部资源，会导致内存占用持续增长，无法释放。

3. 线程安全问题 #

实现未添加同步机制，多线程环境下同时创建相同参数实例时，会出现竞态条件：可能同时生成多个重复实例，或缓存写入时出现数据不一致。

4. 子类继承问题 #

子类会共享父类的缓存字典，引发两个问题：

• 子类实例可能被误匹配到父类的缓存实例，引发类型错误。
• 不同子类的实例可能因键冲突被错误复用。

5. attrs兼容性隐患 #

手动调用instance.__init__(*args, **kwds)可能与attrs.frozen的内部初始化逻辑冲突。虽当前运行正常，但在attrs新版本或复杂场景（如带转换器、验证器的字段）下，可能触发未知问题。

优化建议 #

1. 安全生成缓存键 #

改用可哈希的参数组合作为键，避免字符串冲突：

# 将位置参数和排序后的关键字参数组合成元组，确保等价参数生成相同键
key = (args, tuple(sorted(kwds.items())))

2. 避免内存泄漏 #

使用weakref.WeakValueDictionary替代普通字典，实例无外部引用时会被自动回收：

from typing import ClassVar
import weakref

@frozen
class Cached:
    _cache: ClassVar[weakref.WeakValueDictionary] = weakref.WeakValueDictionary()
    # 其余逻辑适配该缓存结构

3. 保证线程安全 #

添加线程锁，确保缓存读写操作原子性：

from typing import ClassVar
import threading

@frozen
class Cached:
    _cache: ClassVar[dict] = {}
    _lock: ClassVar[threading.Lock] = threading.Lock()

    def __new__(cls, *args, **kwds):
        key = (args, tuple(sorted(kwds.items())))
        with cls._lock:
            if instance := cls._cache.get(key):
                print("cache misses")
                return instance
            # 实例创建与缓存写入逻辑

4. 支持子类继承 #

通过元类为每个子类创建独立缓存：

from typing import ClassVar
import weakref

class CachedMeta(type):
    def __init__(cls, name, bases, attrs):
        super().__init__(name, bases, attrs)
        cls._cache: ClassVar[weakref.WeakValueDictionary] = weakref.WeakValueDictionary()

@frozen
class Cached(metaclass=CachedMeta):
    name: str
    config: int

    def __new__(cls, *args, **kwds):
        key = (args, tuple(sorted(kwds.items())))
        if instance := cls._cache.get(key):
            print("cache misses")
            return instance
        print("cache hits")
        instance = super().__new__(cls)
        instance.__init__(*args, **kwds)
        cls._cache[key] = instance
        return instance

5. 更优雅的attrs兼容实现 #

利用functools.lru_cache简化代码，同时兼容attrs机制：

from typing import ClassVar
from functools import lru_cache
from attrs import frozen

@frozen
class Cached:
    name: str
    config: int

    def __new__(cls, *args, **kwargs):
        # 统一参数格式，确保等价参数生成相同键
        normalized_args = args
        normalized_kwargs = tuple(sorted(kwargs.items()))
        return cls._create_instance(*normalized_args, *normalized_kwargs)

    @classmethod
    @lru_cache(maxsize=None)
    def _create_instance(cls, *args):
        # 拆分位置参数和关键字参数
        pos_args = args[:len(cls.__annotations__)]
        kw_args = dict(args[len(cls.__annotations__):])
        return cls(*pos_args, **kw_args)

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