Python脚本如何限制缓存最大占用空间

wen python案例 2026-06-26 232

本文目录导读：

使用 cachetools 库（推荐）
使用 functools.lru_cache
基于磁盘的缓存限制
自定义内存管理方案
使用第三方库 cacheout
选择建议

在Python中限制缓存最大占用空间，有几种常见的方法,我来介绍几个实用的方案：

使用 `cachetools` 库（推荐）

cachetools 提供了基于大小限制的缓存实现：

from cachetools import LRUCache
import sys
# 创建最大容量为100个元素的LRU缓存
cache = LRUCache(maxsize=100)
# 或者更精确地限制内存使用
class SizeLimitedCache:
    def __init__(self, max_size_mb=100):
        self.max_size = max_size_mb * 1024 * 1024  # 转换为字节
        self.cache = {}
        self.current_size = 0
    def add(self, key, value):
        # 计算新对象大小
        obj_size = sys.getsizeof(value)
        # 如果单个对象超过限制，拒绝存储
        if obj_size > self.max_size:
            raise ValueError("对象太大，无法缓存")
        # 腾出空间
        while self.current_size + obj_size > self.max_size and self.cache:
            self._evict_oldest()
        self.cache[key] = value
        self.current_size += obj_size
    def _evict_oldest(self):
        if self.cache:
            oldest_key = next(iter(self.cache))
            old_value = self.cache.pop(oldest_key)
            self.current_size -= sys.getsizeof(old_value)
# 使用示例
cache = SizeLimitedCache(max_size_mb=50)
cache.add("large_data", b"x" * 1000000)  # 1MB数据

使用 `functools.lru_cache`

Python内置的LRU缓存,按条目数限制：

from functools import lru_cache
@lru_cache(maxsize=100)  # 最多缓存100个结果
def expensive_function(n):
    # 模拟耗时计算
    result = n * n
    return result
# 使用
print(expensive_function(42))  # 计算结果并缓存
print(expensive_function(42))  # 直接从缓存获取
# 查看缓存信息
print(expensive_function.cache_info())
# CacheInfo(hits=1, misses=1, maxsize=100, currsize=1)

基于磁盘的缓存限制

使用 diskcache 库实现磁盘缓存空间限制：

# 安装: pip install diskcache
import diskcache as dc
# 创建磁盘缓存，限制最大大小为1GB
cache = dc.Cache('/tmp/my_cache', size_limit=1_000_000_000)  # 1GB
# 使用
cache.set('key1', 'value1')
cache.set('key2', 'value2')
# 获取缓存大小
print(f"当前缓存大小: {cache.volume()} bytes")

自定义内存管理方案

更精确的内存控制：

import sys
import time
from collections import OrderedDict
class SmartMemoryCache:
    def __init__(self, max_size_bytes=500*1024*1024):  # 默认500MB
        self.max_size = max_size_bytes
        self.cache = OrderedDict()
        self.current_size = 0
    def _estimate_size(self, obj):
        """估算对象大小"""
        if isinstance(obj, (int, float, bool)):
            return sys.getsizeof(obj)
        elif isinstance(obj, str):
            return sys.getsizeof(obj) + len(obj.encode('utf-8'))
        elif isinstance(obj, bytes):
            return sys.getsizeof(obj) + len(obj)
        elif isinstance(obj, (list, tuple, set)):
            size = sys.getsizeof(obj)
            for item in obj:
                size += self._estimate_size(item)
            return size
        elif isinstance(obj, dict):
            size = sys.getsizeof(obj)
            for k, v in obj.items():
                size += self._estimate_size(k) + self._estimate_size(v)
            return size
        else:
            return sys.getsizeof(obj)
    def set(self, key, value, ttl=None):
        """设置缓存，支持TTL"""
        obj_size = self._estimate_size(value)
        # 单个对象超出限制
        if obj_size > self.max_size:
            print(f"警告：对象大小 {obj_size} 超过缓存限制")
            return False
        # 如果键已存在，先释放空间
        if key in self.cache:
            self.current_size -= self.cache[key]['size']
        # 腾出空间
        while self.current_size + obj_size > self.max_size and self.cache:
            _, oldest = self.cache.popitem(last=False)
            self.current_size -= oldest['size']
        # 存储新对象
        expires = time.time() + ttl if ttl else None
        self.cache[key] = {
            'value': value,
            'size': obj_size,
            'expires': expires
        }
        self.current_size += obj_size
        return True
    def get(self, key):
        """获取缓存"""
        if key not in self.cache:
            return None
        item = self.cache[key]
        # 检查TTL
        if item['expires'] and time.time() > item['expires']:
            self.current_size -= item['size']
            del self.cache[key]
            return None
        # 移动到最后（LRU）
        self.cache.move_to_end(key)
        return item['value']
    def clear(self):
        """清空缓存"""
        self.cache.clear()
        self.current_size = 0
# 使用示例
cache = SmartMemoryCache(max_size_bytes=100*1024*1024)  # 100MB
cache.set("data1", {'large': 'x' * 1000000})  # ~1MB数据
cache.set("data2", "hello", ttl=60)  # 60秒后过期
result = cache.get("data1")
print(f"当前缓存大小: {cache.current_size / 1024 / 1024:.2f} MB")

使用第三方库 `cacheout`

# 安装: pip install cacheout
from cacheout import Cache
# 创建缓存，限制条目数
cache = Cache(maxsize=100)
# 或者使用LRU策略
from cacheout import LRUCache
lru_cache = LRUCache(maxsize=100, ttl=300)  # 最多100条，TTL5分钟
# 使用
lru_cache.set('key', 'value')
print(lru_cache.get('key'))