本文目录导读：
在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'))
选择建议
- 简单场景：
functools.lru_cache足够 - 需要精确内存控制：自定义
SmartMemoryCache - 需要持久化存储：
diskcache - 生产环境：
cachetools或cacheout库
根据你的具体需求选择合适的方案，如果需要更精确的内存控制，可能需要结合 psutil 等库监控实际内存使用情况。
