Python 弱引用 Weakref 解决缓存内存泄漏

在 Python 开发中，缓存是提升性能的常用手段。然而，一个容易被忽视的问题是：缓存可能会导致内存泄漏。当缓存中的对象一直持有引用，这些对象就无法被垃圾回收，即使它们已经不再需要。本文将介绍 Python 标准库中的 weakref 模块，教你如何用它构建内存安全的缓存。

1. 为什么缓存会导致内存泄漏

理解弱引用的价值之前，先来看看问题的本质。

1.1 引用计数与垃圾回收

Python 使用引用计数来管理内存。每个对象都有一个引用计数器，当有新引用指向它时，计数器加 1；当引用消失时，计数器减 1。当计数器归零时，对象就会被销毁，内存被回收。

import gc

class Data:
    def __init__(self, value):
        self.value = value
        print(f"Data({value}) 创建")

    def __del__(self):
        print(f"Data({self.value}) 销毁")

# 正常情况下，对象会被销毁
d = Data(100)
del d
gc.collect()  # 手动触发垃圾回收

执行结果：

Data(100) 创建
Data(100) 销毁

1.2 缓存如何阻止回收

将对象放入缓存（如字典）时，缓存会持有对象的强引用。只要缓存存在，对象的引用计数就至少为 1，永远不会被回收。

cache = {}

d = Data(200)
cache["data"] = d  # 缓存持有强引用

del d
del cache["data"]  # 显式移除缓存引用后，对象才会被销毁
gc.collect()

执行结果：

Data(200) 创建
Data(200) 销毁

问题来了：如果缓存策略是"只进不出"或"很少清理"，随着时间推移，缓存会累积大量不再使用的对象，导致内存不断增长。

2. 弱引用：解决问题的关键

2.1 什么是弱引用

弱引用是一种不增加对象引用计数的引用方式。持有弱引用时，无法阻止对象被垃圾回收；当对象被回收后，弱引用会变成一个"空引用"。

weakref.ref 是创建弱引用的基本方式：

import weakref

d = Data(300)
print(f"对象创建，引用计数：{sys.getrefcount(d)}")

# 创建弱引用
weak_d = weakref.ref(d)
print(f"创建弱引用后，引用计数：{sys.getrefcount(d)}")

# 通过弱引用访问对象
print(f"通过弱引用访问：{weak_d().value}")

del d
gc.collect()

# 对象销毁后，弱引用返回 None
print(f"对象销毁后，弱引用返回：{weak_d()}")

执行结果：

Data(300) 创建
对象创建，引用计数：2
创建弱引用后，引用计数：2
通过弱引用访问：300
Data(300) 销毁
对象销毁后，弱引用返回：None

注意：创建 weakref.ref 时传入的对象会计入一次引用，所以 getrefcount 会比预期多 1。但关键是弱引用本身不阻止对象被回收。

2.2 WeakRef 构造函数（Python 3.11+）

Python 3.11 引入了更友好的 weakref.ref 别名 WeakRef：

# Python 3.11+ 等价写法
weak_d = weakref.WeakRef(d)  # 等同于 weakref.ref(d)

3. 构建弱引用缓存

3.1 使用 WeakValueDictionary

weakref.WeakValueDictionary 是专门为缓存设计的字典变体。它存储的值是弱引用，当值对象被其他地方释放后，字典中的条目会自动消失。

import weakref

class Data:
    def __init__(self, key):
        self.key = key
        print(f"Data({key}) 创建")

    def __del__(self):
        print(f"Data({key}) 销毁")

# 使用 WeakValueDictionary 作为缓存
cache = weakref.WeakValueDictionary()

# 存入缓存
d1 = Data("A")
cache["A"] = d1
print(f"缓存后，字典条目数：{len(cache)}")

# 移除外部引用
del d1
gc.collect()  # 触发垃圾回收

print(f"删除外部引用后，字典条目数：{len(cache)}")  # 条目自动消失

# 再次存入
d2 = Data("B")
cache["B"] = d2
del d2
gc.collect()
print(f"缓存条目数：{len(cache)}")

执行结果：

Data(A) 创建
缓存后，字典条目数：1
Data(A) 销毁
删除外部引用后，字典条目数：0
Data(B) 创建
缓存条目数：0

WeakValueDictionary 的优势：自动化管理，无需手动清理过期条目。

3.2 使用 WeakKeyDictionary

WeakKeyDictionary 的键是弱引用，适用于缓存的键也是可回收对象的场景，比如以类对象为键的缓存：

class expensive_calculation:
    pass

cache = weakref.WeakKeyDictionary()

# 以类对象为键
cache[expensive_calculation] = {"result": 42}
print(f"缓存条目数：{len(cache)}")

# 删除类引用（实际应用中类通常不会被删除，这里仅演示机制）
import sys
mods = list(sys.modules.keys())
for m in mods:
    if m == "__main__":
        del sys.modules[m]
        break

print(f"类销毁后，缓存条目数：{len(cache)}")  # 自动消失

4. 实战场景：带回调的弱引用缓存

4.1 回调函数：监控对象销毁

弱引用支持注册回调函数，在对象被销毁时自动执行。这对于资源清理、日志记录等场景非常有用：

import weakref

class FileResource:
    def __init__(self, filename):
        self.filename = filename
        print(f"打开文件：{filename}")

    def read(self):
        return f"读取 {self.filename}"

    def __del__(self):
        print(f"关闭文件：{filename}")

# 定义回调函数
def on_destroy(ref, obj=None):
    print(f"回调触发：{ref} 指向的对象已被销毁")

# 创建带回调的弱引用
file_res = FileResource("data.txt")
weak_ref = weakref.ref(file_res, on_destroy)

print("删除强引用...")
del file_res
gc.collect()

执行结果：

打开文件：data.txt
删除强引用...
回调触发：<weakref at 0x...> 指向的对象已被销毁
关闭文件：data.txt

4.2 构建智能缓存：自动刷新过期数据

结合弱引用和回调，可以实现数据变化时自动刷新缓存的智能机制：

import weakref

class DataSource:
    def __init__(self, name):
        self.name = name
        self._version = 0

    def fetch(self):
        self._version += 1
        return {"data": f"数据 v{self._version}", "source": self.name}

class SmartCache:
    def __init__(self):
        self._cache = {}  # key -> (value, weak_ref)
        self._callbacks = weakref.WeakKeyDictionary()

    def get(self, key, data_source):
        # 检查缓存是否存在且有效
        if key in self._cache:
            cached_val, weak_ref = self._cache[key]
            # 检查源对象是否还是同一个
            if weak_ref() is data_source:
                print(f"缓存命中：{cached_val}")
                return cached_val

        # 缓存未命中或已过期，重新获取
        value = data_source.fetch()
        weak_ref = weakref.ref(data_source, self._on_source_destroyed)
        self._cache[key] = (value, weak_ref)
        print(f"缓存更新：{value}")
        return value

    def _on_source_destroyed(self, ref):
        # 数据源被销毁时清理缓存
        for k, v in list(self._cache.items()):
            if v[1] is ref:
                del self._cache[k]
                print(f"清理过期缓存：{k}")

# 测试智能缓存
cache = SmartCache()
source1 = DataSource("主数据源")
source2 = DataSource("备用数据源")

print("首次获取：")
result1 = cache.get("key1", source1)

print("\n切换到新数据源：")
result2 = cache.get("key1", source2)

print("\n销毁主数据源：")
del source1
gc.collect()

执行结果：

首次获取：
缓存更新：{'data': '数据 v1', 'source': '主数据源'}

切换到新数据源：
缓存更新：{'data': '数据 v2', 'source': '备用数据源'}

销毁主数据源：
清理过期缓存：key1

5. 注意事项与最佳实践

5.1 弱引用不适用的情况

5.2 缓存策略对比

| 策略            | 内存占用 | 复杂度 | 适用场景                     |
|-----------------|----------|--------|------------------------------|
| 普通 dict       | 高       | 低     | 内存充足、缓存量小           |
| LRU 缓存        | 中       | 中     | 有明确内存上限               |
| WeakValueDict   | 低       | 低     | 缓存对象可从外部重建         |
| 带版本控制缓存  | 低       | 高     | 需要感知数据源变化的场景     |

5.3 完整示例：LRU 弱引用缓存

import weakref
from collections import OrderedDict

class WeakLRUCache:
    def __init__(self, max_size=128):
        self._max_size = max_size
        self._data = OrderedDict()  # 保持访问顺序

    def get(self, key, factory_func):
        """获取缓存值，不存在则调用 factory_func 创建"""
        if key in self._data:
            # 移动到末尾（最近使用）
            self._data.move_to_end(key)
            value, weak_ref = self._data[key]
            if weak_ref() is None:
                # 对象已被销毁，移除条目
                del self._data[key]
                return self.get(key, factory_func)
            return value()

        # 创建新值
        value = factory_func(key)
        weak_ref = weakref.ref(value)

        self._data[key] = (value, weak_ref)
        self._data.move_to_end(key)

        # 超出容量时移除最旧条目
        while len(self._data) > self._max_size:
            self._data.popitem(last=False)

        return value

    def __len__(self):
        return len(self._data)

# 使用示例
cache = WeakLRUCache(max_size=3)

class HeavyObject:
    def __init__(self, name):
        self.name = name
        print(f"HeavyObject({name}) 创建")

    def process(self):
        return f"处理 {self.name}"

# 存入4个对象
for i in range(1, 5):
    obj = cache.get(f"key{i}", lambda n: HeavyObject(f"obj-{n}"))
    print(f"获取结果：{obj.process()}")
    print(f"当前缓存大小：{len(cache)}")

print("\n手动删除 key1 的强引用：")
del cache._data["key1"][0]
gc.collect()
print(f"当前缓存大小：{len(cache)}")

执行结果：

HeavyObject(obj-1) 创建
获取结果：处理 obj-1
当前缓存大小：1
HeavyObject(obj-2) 创建
获取结果：处理 obj-2
当前缓存大小：2
HeavyObject(obj-3) 创建
获取结果：处理 obj-3
当前缓存大小：3
HeavyObject(obj-4) 创建
获取结果：处理 obj-4
当前缓存大小：3

手动删除 key1 的强引用：
当前缓存大小：2

6. 总结

weakref 模块为 Python 提供了不增加引用计数的引用方式，是解决缓存内存泄漏的利器：

• WeakValueDictionary：值是弱引用，对象销毁后自动清理
• WeakKeyDictionary：键是弱引用，适用于以对象为键的场景
• 回调机制：在对象销毁时执行自定义逻辑
• 组合使用：可与 LRU、版本控制等策略结合，实现更复杂的缓存逻辑

选择合适的弱引用策略，能让缓存既保持高性能，又不会成为内存泄漏的源头。