You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何装饰Python实例方法并避免实例间共享闭包环境?

解决类方法装饰器闭包变量共享问题

当装饰器在类定义阶段装饰方法时,整个类的所有实例会共用装饰器闭包中的变量,导致状态(比如计数、缓存的token)被共享,这显然不符合每个实例独立维护状态的需求。以下是几种保留装饰器模式的解决方案:

方案一:将状态存储在实例属性中

利用装饰器wrapper的第一个参数self(类实例),把需要维护的状态直接存在实例的属性里,每个实例拥有独立的属性,自然实现状态隔离。

针对计数示例修改:

from functools import wraps

def decorator(f):
    @wraps(f)
    def wrapper(self, *args, **kwargs):
        # 第一次调用时初始化实例属性
        if not hasattr(self, '_closure_counter'):
            self._closure_counter = 0
        self._closure_counter += 1
        print(self._closure_counter)
        return f(self, *args, **kwargs)
    return wrapper

class ClassA:
    @decorator
    def decorated_method(self):
        pass

# 测试
inst1 = ClassA()
inst2 = ClassA()
inst1.decorated_method()  # 输出1
inst2.decorated_method()  # 输出1
inst1.decorated_method()  # 输出2

适配token缓存场景:

from functools import wraps
import time

def token_cache(f):
    @wraps(f)
    def wrapper(self, *args, **kwargs):
        now = time.time()
        # 检查实例是否有有效缓存
        if not hasattr(self, '_cached_token') or not hasattr(self, '_token_expire') or now >= self._token_expire:
            # 调用原方法获取新token(假设原方法返回(token, 有效期秒数))
            token, expire_seconds = f(self, *args, **kwargs)
            self._cached_token = token
            self._token_expire = now + expire_seconds
        return self._cached_token
    return wrapper

class TokenSupplier:
    @token_cache
    def get_token(self):
        # 模拟请求新token的逻辑
        print("请求新token...")
        current_token = f"TOKEN_{int(time.time())}"
        return current_token, 10  # token有效期10秒

# 测试
supplier1 = TokenSupplier()
supplier2 = TokenSupplier()
print(supplier1.get_token())  # 打印"请求新token...",输出对应token
print(supplier1.get_token())  # 直接输出缓存的token
print(supplier2.get_token())  # 打印"请求新token...",输出另一个token

方案二:用弱引用字典存储实例状态

如果不想修改类实例的属性,可以使用weakref.WeakKeyDictionary来存储每个实例对应的状态,它会自动回收已销毁实例的状态,避免内存泄漏。

from functools import wraps
import weakref

def decorator(f):
    # 弱引用字典:键为实例,值为对应状态
    instance_states = weakref.WeakKeyDictionary()

    @wraps(f)
    def wrapper(self, *args, **kwargs):
        if self not in instance_states:
            instance_states[self] = 0
        instance_states[self] += 1
        print(instance_states[self])
        return f(self, *args, **kwargs)
    return wrapper

class ClassA:
    @decorator
    def decorated_method(self):
        pass

# 测试
inst1 = ClassA()
inst2 = ClassA()
inst1.decorated_method()  # 输出1
inst2.decorated_method()  # 输出1
del inst1  # 实例销毁后,instance_states中对应的条目会被自动清理

方案三:使用类实现装饰器

用类来封装装饰器逻辑,本质还是通过实例属性或弱引用字典实现状态隔离,写法上更面向对象:

from functools import wraps
import weakref
import time

class TokenCacheDecorator:
    def __init__(self):
        self.instance_cache = weakref.WeakKeyDictionary()

    def __call__(self, f):
        @wraps(f)
        def wrapper(self, *args, **kwargs):
            now = time.time()
            if self not in self.instance_cache:
                # 初始化缓存结构
                self.instance_cache[self] = {'token': None, 'expire': 0}
            
            cache = self.instance_cache[self]
            if not cache['token'] or now >= cache['expire']:
                token, expire_seconds = f(self, *args, **kwargs)
                cache['token'] = token
                cache['expire'] = now + expire_seconds
            return cache['token']
        return wrapper

# 实例化装饰器
token_cache = TokenCacheDecorator()

class TokenSupplier:
    @token_cache
    def get_token(self):
        print("请求新token...")
        return f"TOKEN_{int(time.time())}", 10

总结

  • 方案一最简单直接,适合大多数场景,无需额外依赖;
  • 方案二适合不能修改实例属性的场景,避免侵入类结构;
  • 方案三更适合复杂的装饰器逻辑,便于扩展维护。

内容的提问来源于stack exchange,提问作者Gonçalo

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.23 05:35:05