本文目录导读：
Python 缓存优化需要根据缓存类型、数据特征和访问模式选择合适的策略，以下是系统性的优化方法,按效果从高到低排序：
数据压缩（最直接有效）
对象级压缩
import zlib
import pickle
def compress_and_cache(data):
# 压缩后存储
compressed = zlib.compress(pickle.dumps(data))
cache[key] = compressed
def get_from_cache(key):
compressed = cache[key]
return pickle.loads(zlib.decompress(compressed))
使用更高效的序列化
# 相比 pickle，msgpack 更紧凑
import msgpack
data = {"key": "value", "numbers": [1,2,3]}
packed = msgpack.packb(data) # 比 pickle 小 30-50%
合理选择缓存淘汰策略
from functools import lru_cache
import sys
# 限制缓存大小
@lru_cache(maxsize=128) # 只保留最近使用的128个结果
def expensive_function(n):
return n * n
# 自定义LRU缓存（比lru_cache更节省内存）
from collections import OrderedDict
class SizeLimitedCache:
def __init__(self, max_size_mb=100):
self.cache = OrderedDict()
self.max_size = max_size_mb * 1024 * 1024
self.current_size = 0
def set(self, key, value):
# 计算大小
size = sys.getsizeof(value)
# 如果单个值太大，不缓存
if size > self.max_size:
return
# 淘汰直到有足够空间
while self.current_size + size > self.max_size:
self.cache.popitem(last=False)
self.cache[key] = value
self.current_size += size
使用更高效的数据结构
替换字典为更紧凑的结构
# 使用 __slots__ 减少对象开销
class CacheEntry:
__slots__ = ['data', 'expires_at', 'size']
def __init__(self, data, expires_at):
self.data = data
self.expires_at = expires_at
self.size = sys.getsizeof(data)
# 使用数组代替字典（适用于整数键）
import array
cache = array.array('i') # 每个元素只占4字节
# 使用 blist 替代 list（大数据量时）
from blist import blist
cache = blist([None] * 1000) # 内存占用更稳定的链表
缓存分割和分片
# 将大缓存拆分成小缓存
class ShardedCache:
def __init__(self, num_shards=16):
self.shards = [{} for _ in range(num_shards)]
def _get_shard(self, key):
return self.shards[hash(key) % len(self.shards)]
def get(self, key):
return self._get_shard(key).get(key)
def set(self, key, value):
shard = self._get_shard(key)
# 只清理单个分片，而不是整个缓存
if len(shard) > 10000:
# 清理最旧的20%
for k in list(shard.keys())[:2000]:
del shard[k]
shard[key] = value
惰性加载和缓存预热
# 延迟初始化缓存
class LazyInitCache:
def __init__(self):
self._cache = None # 使用时才创建
@property
def cache(self):
if self._cache is None:
self._cache = {}
return self._cache
def get_or_compute(self, key, compute_func):
if key not in self.cache:
self.cache[key] = compute_func()
return self.cache[key]
使用专用缓存库
diskcache（磁盘缓存）
import diskcache as dc
# 使用磁盘缓存，内存占用极小
cache = dc.Cache('/tmp/my_cache')
cache.set('key', large_data, expire=3600) # 1小时后自动过期
Redis（分布式缓存）
import redis
r = redis.Redis(host='localhost', port=6379)
# Redis 支持内存优化
r.set('key', data, ex=3600) # 设置过期时间
内存复用
# 复用已分配的内存
class MemoryPool:
def __init__(self, size=1024):
self.pool = [bytearray(size) for _ in range(10)]
self.index = 0
def get_buffer(self):
buffer = self.pool[self.index]
self.index = (self.index + 1) % len(self.pool)
return buffer[:0] # 重置但保留内存
# 对象池
from collections import deque
class ObjectPool:
def __init__(self, cls, max_size=100):
self.pool = deque(maxlen=max_size)
self.cls = cls
def acquire(self):
return self.pool.pop() if self.pool else self.cls()
def release(self, obj):
self.pool.append(obj)
监控和调优工具
import tracemalloc
# 跟踪内存使用
tracemalloc.start()
# 分析缓存大小
def analyze_cache(cache):
total_size = 0
for key, value in cache.items():
size = sys.getsizeof(value)
total_size += size
print(f"Key: {key}, Size: {size/1024:.2f}KB")
print(f"Total: {total_size/1024/1024:.2f}MB")
# 使用 memory_profiler
# pip install memory_profiler
from memory_profiler import profile
@profile
def test_cache():
cache = {}
for i in range(10000):
cache[i] = "x" * 1000
实际优化案例
# 优化前的缓存
class BasicCache:
def __init__(self):
self.cache = {}
def set(self, key, value):
self.cache[key] = value # 没有限制，内存无限增长
# 优化后的缓存
class OptimizedCache:
def __init__(self, max_memory_mb=500, ttl_seconds=3600):
self.cache = {}
self.max_memory = max_memory_mb * 1024 * 1024
self.ttl = ttl_seconds
self.current_memory = 0
self.timestamps = {}
def set(self, key, value, compress=True):
# 1. 数据压缩
if compress:
value = zlib.compress(pickle.dumps(value))
size = sys.getsizeof(value)
# 2. 如果单个值太大，不缓存
if size > self.max_memory * 0.1:
return
# 3. 淘汰策略
while self.current_memory + size > self.max_memory:
oldest_key = min(self.timestamps, key=self.timestamps.get)
self._remove(oldest_key)
# 4. 存储
self.cache[key] = value
self.timestamps[key] = time.time()
self.current_memory += size
def get(self, key):
# 5. TTL 检查
if key in self.timestamps:
if time.time() - self.timestamps[key] > self.ttl:
self._remove(key)
return None
value = self.cache.get(key)
if value:
return pickle.loads(zlib.decompress(value))
return None
def _remove(self, key):
if key in self.cache:
self.current_memory -= sys.getsizeof(self.cache[key])
del self.cache[key]
del self.timestamps[key]
关键优化建议
- 先测量后优化：使用
sys.getsizeof()和tracemalloc了解缓存实际占用 - 缓存不可变数据：使用
tuple代替list，frozenset代替set - 避免缓存大对象：如果数据大于10MB，考虑分片存储
- 设置合理的TTL：不要让数据在缓存中停留过久
- 使用引用计数：
weakref可以让对象在无引用时自动释放
选择哪种优化方法取决于你的具体场景：数据量、访问频率、修改频率、硬件资源等,最有效的方式往往是组合多种策略。
