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

如何为functools的lru_cache设置内存大小限制?

实现基于内存限制的LRU缓存

方法一:用第三方库 cachetools 快速实现

cachetools 是专门的缓存工具库,支持自定义元素权重规则,刚好可以用来实现按内存大小限制的LRU缓存,比自己造轮子更省心。

  1. 先安装库:
pip install cachetools
  1. 核心实现代码:
    要准确计算缓存对象的内存占用,推荐用 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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.07.11 00:22:32