本文目录导读：
- 使用 pickle 序列化（最简单）
- 使用 JSON 格式（可读性好）
- 使用 SQLite 数据库（适合大量数据）
- 使用 Redis（生产环境推荐）
- 使用 shelve（内置库，类似字典的持久化）
- 使用 diskcache 库（功能最完整）
- 选择建议
- 最佳实践建议
在Python中持久化保存缓存数据,主要有以下几种常见方法和实现方案：
使用 pickle 序列化（最简单）
import pickle
import os
class SimpleCache:
def __init__(self, cache_file='cache.pkl'):
self.cache_file = cache_file
self.data = self.load_cache()
def load_cache(self):
if os.path.exists(self.cache_file):
with open(self.cache_file, 'rb') as f:
return pickle.load(f)
return {}
def save_cache(self):
with open(self.cache_file, 'wb') as f:
pickle.dump(self.data, f)
def set(self, key, value):
self.data[key] = value
self.save_cache()
def get(self, key):
return self.data.get(key)
# 使用示例
cache = SimpleCache()
cache.set('user_123', {'name': 'Alice', 'age': 30})
print(cache.get('user_123'))
使用 JSON 格式（可读性好）
import json
import os
class JSONCache:
def __init__(self, cache_file='cache.json'):
self.cache_file = cache_file
self.data = self.load_cache()
def load_cache(self):
if os.path.exists(self.cache_file):
with open(self.cache_file, 'r', encoding='utf-8') as f:
return json.load(f)
return {}
def save_cache(self):
with open(self.cache_file, 'w', encoding='utf-8') as f:
json.dump(self.data, f, ensure_ascii=False, indent=2)
def set(self, key, value):
self.data[key] = value
self.save_cache()
def get(self, key):
return self.data.get(key)
# 使用示例
cache = JSONCache()
cache.set('search_history', ['python', 'json', 'caching'])
print(cache.get('search_history'))
使用 SQLite 数据库（适合大量数据）
import sqlite3
import json
from datetime import datetime, timedelta
class SQLiteCache:
def __init__(self, db_file='cache.db', expire_seconds=3600):
self.db_file = db_file
self.expire_seconds = expire_seconds
self.init_db()
def init_db(self):
with sqlite3.connect(self.db_file) as conn:
conn.execute('''
CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value TEXT,
expire_time TIMESTAMP
)
''')
def set(self, key, value):
expire_time = datetime.now() + timedelta(seconds=self.expire_seconds)
with sqlite3.connect(self.db_file) as conn:
conn.execute(
'INSERT OR REPLACE INTO cache (key, value, expire_time) VALUES (?, ?, ?)',
(key, json.dumps(value), expire_time)
)
def get(self, key):
with sqlite3.connect(self.db_file) as conn:
cursor = conn.execute(
'SELECT value, expire_time FROM cache WHERE key = ?',
(key,)
)
row = cursor.fetchone()
if row:
value, expire_time = row
if datetime.now() < datetime.strptime(expire_time, '%Y-%m-%d %H:%M:%S.%f'):
return json.loads(value)
else:
self.delete(key) # 自动清理过期数据
return None
def delete(self, key):
with sqlite3.connect(self.db_file) as conn:
conn.execute('DELETE FROM cache WHERE key = ?', (key,))
# 使用示例
cache = SQLiteCache('my_cache.db', expire_seconds=3600)
cache.set('api_response', {'status': 'ok', 'data': [1, 2, 3]})
result = cache.get('api_response')
print(result)
使用 Redis（生产环境推荐）
import redis
import json
class RedisCache:
def __init__(self, host='localhost', port=6379, db=0, expire_seconds=3600):
self.client = redis.Redis(host=host, port=port, db=db)
self.expire_seconds = expire_seconds
def set(self, key, value):
self.client.setex(
key,
self.expire_seconds,
json.dumps(value)
)
def get(self, key):
value = self.client.get(key)
if value:
return json.loads(value)
return None
def delete(self, key):
self.client.delete(key)
def clear_all(self):
self.client.flushdb()
# 需要安装 redis: pip install redis
# 使用示例
cache = RedisCache()
cache.set('session:user123', {'login_time': '2024-01-01', 'permissions': ['read', 'write']})
print(cache.get('session:user123'))
使用 shelve（内置库，类似字典的持久化）
import shelve
class ShelfCache:
def __init__(self, filename='cache.shelve'):
self.filename = filename
def set(self, key, value):
with shelve.open(self.filename) as shelf:
shelf[key] = value
def get(self, key):
with shelve.open(self.filename) as shelf:
return shelf.get(key)
def delete(self, key):
with shelve.open(self.filename) as shelf:
if key in shelf:
del shelf[key]
# 使用示例
cache = ShelfCache()
cache.set('config', {'theme': 'dark', 'language': 'zh_CN'})
print(cache.get('config'))
使用 diskcache 库（功能最完整）
from diskcache import Cache
class DiskCacheWrapper:
def __init__(self, directory='./cache', size_limit=100*(1024**2)): # 100MB
self.cache = Cache(directory, size_limit=size_limit)
def set(self, key, value, expire=None):
self.cache.set(key, value, expire=expire)
def get(self, key, default=None):
return self.cache.get(key, default)
def delete(self, key):
del self.cache[key]
# 需要安装: pip install diskcache
# 使用示例
cache = DiskCacheWrapper()
cache.set('computed_result', {'value': 42, 'timestamp': '2024-01-01'}, expire=3600)
print(cache.get('computed_result'))
选择建议
| 方法 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| pickle | 小数据量、单机 | 简单、支持任何对象 | 不安全、不可读 |
| JSON | 需要人工查看数据 | 可读性好、跨语言 | 只支持基本类型 |
| SQLite | 中等数据量、结构化数据 | 支持查询、过期 | 需要SQL知识 |
| Redis | 高并发、分布式系统 | 高性能、内存级 | 需要额外服务 |
| shelve | 简单键值存储 | 内置库、API简单 | 性能一般 |
| diskcache | 需要高级功能 | 功能完整、性能好 | 第三方依赖 |
最佳实践建议
import hashlib
from functools import wraps
def cached(cache_instance, expire=3600):
"""缓存装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 生成缓存键
key = f"{func.__name__}:{hashlib.md5(str(args).encode()).hexdigest()}"
# 尝试获取缓存
result = cache_instance.get(key)
if result is not None:
print(f"Cache hit for {key}")
return result
# 执行函数
result = func(*args, **kwargs)
# 保存缓存
cache_instance.set(key, result)
print(f"Cache miss for {key}")
return result
return wrapper
return decorator
# 使用示例
cache = JSONCache('app_cache.json')
@cached(cache, expire=300) # 5分钟过期
def expensive_function(user_id):
# 模拟耗时操作
import time
time.sleep(2)
return {'user_id': user_id, 'data': 'some expensive data'}
# 第一次调用会执行函数
result1 = expensive_function(123)
# 第二次调用会从缓存获取
result2 = expensive_function(123)
根据你的具体需求（数据量、并发要求、是否需要过期时间等）选择合适的方案，对于大多数中小型项目，使用 JSON + 文件存储 或 SQLite 就足够了。
