Python脚本如何优化缓存存储占用空间

wen python案例 2026-06-26 244

本文目录导读：

数据压缩（最直接有效）
合理选择缓存淘汰策略
使用更高效的数据结构
缓存分割和分片
惰性加载和缓存预热
使用专用缓存库
内存复用
监控和调优工具
实际优化案例
关键优化建议

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]