本文目录导读：
我来介绍几种Python压缩缓存文件的方法：
基础文件压缩方法
使用 gzip 压缩单个文件
import gzip
import shutil
import os
def compress_file(input_file, output_file=None):
"""压缩单个文件"""
if output_file is None:
output_file = input_file + '.gz'
with open(input_file, 'rb') as f_in:
with gzip.open(output_file, 'wb') as f_out:
shutil.copyfileobj(f_in, f_out)
# 可选：删除原文件
# os.remove(input_file)
return output_file
# 使用示例
compress_file('cache/data.json')
使用 zipfile 压缩多个文件
import zipfile
import os
def compress_to_zip(source_dir, output_file='cache.zip'):
"""将目录压缩为zip文件"""
with zipfile.ZipFile(output_file, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk(source_dir):
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.relpath(file_path, os.path.dirname(source_dir))
zipf.write(file_path, arcname)
return output_file
缓存数据压缩存储
使用 pickle + gzip 压缩序列化数据
import pickle
import gzip
def save_compressed_cache(data, filepath):
"""保存压缩的缓存数据"""
with gzip.open(filepath, 'wb') as f:
pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)
def load_compressed_cache(filepath):
"""加载压缩的缓存数据"""
with gzip.open(filepath, 'rb') as f:
return pickle.load(f)
# 使用示例
cache_data = {'key': 'value', 'list': [1, 2, 3]}
save_compressed_cache(cache_data, 'cache/compressed_cache.pkl.gz')
loaded_data = load_compressed_cache('cache/compressed_cache.pkl.gz')
使用 lzma 提供更好的压缩率
import lzma
import json
def compress_with_lzma(data, filepath):
"""使用 LZMA 算法压缩数据"""
compressed = lzma.compress(json.dumps(data).encode('utf-8'))
with open(filepath, 'wb') as f:
f.write(compressed)
def decompress_with_lzma(filepath):
"""解压 LZMA 压缩的数据"""
with open(filepath, 'rb') as f:
compressed = f.read()
return json.loads(lzma.decompress(compressed).decode('utf-8'))
图片缓存压缩
使用 Pillow 压缩图片
from PIL import Image
import os
def compress_image(input_path, output_path=None, quality=85):
"""压缩图片文件"""
if output_path is None:
output_path = input_path
img = Image.open(input_path)
# 转换为RGB模式（去除alpha通道）
if img.mode in ('RGBA', 'LA', 'P'):
img = img.convert('RGB')
# 保存并压缩
img.save(output_path, 'JPEG', optimize=True, quality=quality)
return output_path
# 批量压缩目录中的图片
def compress_images_in_directory(directory, quality=85):
for filename in os.listdir(directory):
if filename.lower().endswith(('.png', '.jpg', '.jpeg', '.gif')):
filepath = os.path.join(directory, filename)
compress_image(filepath, quality=quality)
缓存清理策略
自动清理过期缓存
import time
import os
import shutil
class CacheCleaner:
def __init__(self, cache_dir, max_size_mb=100, expire_hours=24):
self.cache_dir = cache_dir
self.max_size_bytes = max_size_mb * 1024 * 1024
self.expire_seconds = expire_hours * 3600
def clean_expired(self):
"""清理过期缓存"""
now = time.time()
for filename in os.listdir(self.cache_dir):
filepath = os.path.join(self.cache_dir, filename)
if os.path.isfile(filepath):
# 检查文件是否过期
if now - os.path.getmtime(filepath) > self.expire_seconds:
os.remove(filepath)
print(f"删除过期文件: {filename}")
def clean_by_size(self):
"""清理超出大小限制的缓存"""
files = []
total_size = 0
# 获取所有文件及其大小
for filename in os.listdir(self.cache_dir):
filepath = os.path.join(self.cache_dir, filename)
if os.path.isfile(filepath):
size = os.path.getsize(filepath)
files.append((filepath, size, os.path.getmtime(filepath)))
total_size += size
# 按修改时间排序（最旧的在前）
files.sort(key=lambda x: x[2])
# 删除最旧的文件，直到总大小低于限制
while total_size > self.max_size_bytes and files:
filepath, size, _ = files.pop(0)
os.remove(filepath)
total_size -= size
print(f"删除缓存文件: {os.path.basename(filepath)}")
# 使用示例
cleaner = CacheCleaner('cache/', max_size_mb=50, expire_hours=48)
cleaner.clean_expired()
cleaner.clean_by_size()
完整示例：缓存管理类
import hashlib
import json
import gzip
import os
import time
class CompressedCache:
def __init__(self, cache_dir='.cache', compress_level=9):
self.cache_dir = cache_dir
self.compress_level = compress_level
os.makedirs(cache_dir, exist_ok=True)
def _get_cache_path(self, key):
"""获取缓存文件路径"""
hash_key = hashlib.md5(key.encode()).hexdigest()
return os.path.join(self.cache_dir, f'{hash_key}.cache')
def set(self, key, value, ttl=None):
"""设置缓存"""
cache_data = {
'value': value,
'expire_time': time.time() + ttl if ttl else None,
'created': time.time()
}
cache_path = self._get_cache_path(key)
with gzip.open(cache_path, 'wb', compresslevel=self.compress_level) as f:
f.write(json.dumps(cache_data).encode())
def get(self, key):
"""获取缓存"""
cache_path = self._get_cache_path(key)
if not os.path.exists(cache_path):
return None
try:
with gzip.open(cache_path, 'rb') as f:
cache_data = json.loads(f.read().decode())
# 检查是否过期
if cache_data['expire_time'] and time.time() > cache_data['expire_time']:
os.remove(cache_path)
return None
return cache_data['value']
except Exception:
return None
def clear(self):
"""清空所有缓存"""
for filename in os.listdir(self.cache_dir):
filepath = os.path.join(self.cache_dir, filename)
if filename.endswith('.cache'):
os.remove(filepath)
def get_stats(self):
"""获取缓存统计信息"""
total_size = 0
file_count = 0
for filename in os.listdir(self.cache_dir):
filepath = os.path.join(self.cache_dir, filename)
if filename.endswith('.cache'):
total_size += os.path.getsize(filepath)
file_count += 1
return {
'file_count': file_count,
'total_size_mb': total_size / (1024 * 1024),
'cache_dir': self.cache_dir
}
# 使用示例
cache = CompressedCache()
cache.set('user_123', {'name': 'Alice', 'data': [1,2,3]}, ttl=3600)
data = cache.get('user_123')
print(cache.get_stats())
最佳实践建议
-
选择合适的压缩算法：
gzip：快速，兼容性好lzma：压缩率高，但速度较慢zstd：平衡性能和压缩率
-
设置合理的缓存策略：
- TTL（生存时间）
- 最大缓存大小
- 定期清理
-
监控缓存效率：
- 记录命中率
- 监控压缩率
- 定期评估存储成本
这些方法可以根据具体场景选择合适的压缩方案,有效减少缓存文件的存储空间。
