如何为functools的lru_cache设置内存大小限制?
实现基于内存限制的LRU缓存
方法一:用第三方库 cachetools 快速实现
cachetools 是专门的缓存工具库,支持自定义元素权重规则,刚好可以用来实现按内存大小限制的LRU缓存,比自己造轮子更省心。
- 先安装库:
pip install cachetools
- 核心实现代码:
要准确计算缓存对象的内存占用,推荐用pympler的asizeof(比Python内置的sys.getsizeof更靠谱,能递归计算对象及其引用的所有子对象的总大小):
from cachetools import LRUCache from pympler import asizeof # 定义内存上限:100MB = 100 * 1024 * 1024 字节 MAX_MEMORY = 100 * 1024 * 1024 # 自定义权重函数,返回单个缓存键值对的总内存占用 def calc_memory_usage(key, value): return asizeof.asizeof(key) + asizeof.asizeof(value) # 初始化内存限制的LRU缓存 memory_cache = LRUCache(maxsize=MAX_MEMORY, getsizeof=calc_memory_usage) # 用装饰器形式使用缓存 @memory_cache def heavy_computation(x): # 模拟耗时计算 return x ** 3
如果不想额外依赖 pympler,可以自己写递归版的内存计算函数替代:
import sys def get_total_obj_size(obj, seen=None): """递归计算对象的总内存占用""" size = sys.getsizeof(obj) if seen is None: seen = set() obj_id = id(obj) if obj_id in seen: return 0 seen.add(obj_id) # 处理字典、实例属性、可迭代容器(排除字符串/字节串) if isinstance(obj, dict): size += sum(get_total_obj_size(v, seen) for v in obj.values()) size += sum(get_total_obj_size(k, seen) for k in obj.keys()) elif hasattr(obj, '__dict__'): size += get_total_obj_size(obj.__dict__, seen) elif hasattr(obj, '__iter__') and not isinstance(obj, (str, bytes, bytearray)): size += sum(get_total_obj_size(item, seen) for item in obj) return size # 替换上面的calc_memory_usage函数 def calc_memory_usage(key, value): return get_total_obj_size(key) + get_total_obj_size(value)
方法二:手动实现基于内存限制的LRU缓存
如果不想依赖任何第三方库,可以基于 collections.OrderedDict 手动实现LRU逻辑,同时跟踪总内存占用:
import sys from collections import OrderedDict def get_total_obj_size(obj, seen=None): """递归计算对象总内存占用,同上""" size = sys.getsizeof(obj) if seen is None: seen = set() obj_id = id(obj) if obj_id in seen: return 0 seen.add(obj_id) if isinstance(obj, dict): size += sum(get_total_obj_size(v, seen) for v in obj.values()) size += sum(get_total_obj_size(k, seen) for k in obj.keys()) elif hasattr(obj, '__dict__'): size += get_total_obj_size(obj.__dict__, seen) elif hasattr(obj, '__iter__') and not isinstance(obj, (str, bytes, bytearray)): size += sum(get_total_obj_size(item, seen) for item in obj) return size class MemoryLimitedLRU: def __init__(self, max_memory): self.max_memory = max_memory self.cache = OrderedDict() self.used_memory = 0 def __call__(self, func): def wrapper(*args, **kwargs): # 生成缓存键(注意:如果kwargs包含不可哈希类型,需要调整键的生成逻辑) cache_key = (args, frozenset(kwargs.items())) if cache_key in self.cache: # 将访问过的键移到末尾,标记为最近使用 self.cache.move_to_end(cache_key) return self.cache[cache_key] # 执行原函数得到结果 result = func(*args, **kwargs) # 计算当前键值对的内存占用 item_size = get_total_obj_size(cache_key) + get_total_obj_size(result) # 如果添加后超出内存上限,淘汰最久未使用的元素 while self.used_memory + item_size > self.max_memory and self.cache: old_key, old_val = self.cache.popitem(last=False) self.used_memory -= get_total_obj_size(old_key) + get_total_obj_size(old_val) # 将新结果加入缓存 self.cache[cache_key] = result self.used_memory += item_size return result return wrapper # 使用自定义缓存 @MemoryLimitedLRU(max_memory=100*1024*1024) def heavy_computation(x): return x ** 3
注意事项
- 内存计算的性能损耗:递归计算对象大小会带来额外开销,如果缓存的是超大对象,这个开销可能比较明显,需要根据业务场景权衡。
- 缓存键的兼容性:上面用
(args, frozenset(kwargs.items()))作为缓存键,但如果kwargs里有列表、字典这类不可哈希的类型会报错,需要根据实际情况调整键的生成逻辑(比如把不可哈希对象转为可哈希的形式)。
内容的提问来源于stack exchange,提问作者edd313
相关产品推荐
相关产品推荐

