本文目录导读：
我来介绍几种Python中避免缓存数据重复存储的方法：
使用字典/LRU缓存
from functools import lru_cache
import time
# 方法1：使用lru_cache装饰器
@lru_cache(maxsize=128)
def expensive_function(n):
"""模拟耗时计算"""
time.sleep(1)
return n * 2
# 多次调用相同参数只会执行一次
print(expensive_function(5)) # 首次计算
print(expensive_function(5)) # 直接从缓存读取
print(expensive_function(5)) # 直接从缓存读取
print(f"缓存信息: {expensive_function.cache_info()}")
自定义缓存类
class Cache:
def __init__(self):
self._data = {}
def get_or_set(self, key, value_func):
"""如果key存在返回缓存值，否则执行函数并缓存"""
if key not in self._data:
self._data[key] = value_func()
return self._data[key]
def clear(self):
self._data.clear()
# 使用示例
cache = Cache()
def process_data(data_id):
"""模拟数据处理"""
return f"Processed data {data_id}"
# 避免重复存储
result1 = cache.get_or_set("user_1", lambda: process_data("user_1"))
result2 = cache.get_or_set("user_1", lambda: process_data("user_1")) # 不会重复执行
print(f"结果相同: {result1 == result2}") # True
使用集合检查重复
class DataDeduplicator:
def __init__(self):
self.processed_ids = set()
def process_if_new(self, data_id, data):
"""只处理未处理过的数据"""
if data_id in self.processed_ids:
print(f"数据 {data_id} 已处理，跳过")
return False
# 处理新数据
self.processed_ids.add(data_id)
print(f"处理新数据: {data_id}")
# 实际存储数据...
return True
# 使用示例
dedup = DataDeduplicator()
dedup.process_if_new("001", {"name": "Alice"}) # 处理
dedup.process_if_new("001", {"name": "Alice"}) # 跳过
dedup.process_if_new("002", {"name": "Bob"}) # 处理
使用字典记录哈希值
import hashlib
import json
class ContentAddressableCache:
def __init__(self):
self.content_hash_map = {} # hash -> storage_key
self.storage = {} # storage_key -> data
def store(self, data):
"""基于内容去重存储"""
# 计算内容的哈希值
content_str = json.dumps(data, sort_keys=True)
content_hash = hashlib.sha256(content_str.encode()).hexdigest()
if content_hash in self.content_hash_map:
print(f"数据已存在，使用已有存储位置")
return self.content_hash_map[content_hash]
# 新数据，生成存储key并保存
storage_key = f"data_{len(self.storage)}"
self.storage[storage_key] = data
self.content_hash_map[content_hash] = storage_key
print(f"存储新数据: {storage_key}")
return storage_key
# 使用示例
cache = ContentAddressableCache()
data1 = {"name": "Alice", "age": 30}
data2 = {"age": 30, "name": "Alice"} # 相同内容，不同顺序
key1 = cache.store(data1) # 存储新数据
key2 = cache.store(data2) # 命中缓存
print(f"相同数据，相同存储key: {key1 == key2}") # True
带过期时间的缓存
import time
from datetime import datetime, timedelta
class TimeBasedCache:
def __init__(self, ttl_seconds=300):
self.cache = {} # key -> (value, expiry_time)
self.ttl = ttl_seconds
def get(self, key):
"""获取缓存值，检查是否过期"""
if key in self.cache:
value, expiry = self.cache[key]
if datetime.now() < expiry:
return value
else:
# 过期删除
del self.cache[key]
return None
def set(self, key, value):
"""设置缓存，自动添加过期时间"""
expiry = datetime.now() + timedelta(seconds=self.ttl)
self.cache[key] = (value, expiry)
def get_or_compute(self, key, compute_func):
"""缓存模式：获取或计算"""
cached = self.get(key)
if cached is not None:
return cached
# 计算并缓存
value = compute_func()
self.set(key, value)
return value
# 使用示例
cache = TimeBasedCache(ttl_seconds=10)
def compute_data():
print("执行实际计算...")
return {"result": 42}
result1 = cache.get_or_compute("key1", compute_data) # 计算
result2 = cache.get_or_compute("key1", compute_data) # 从缓存获取
print(f"重复获取: {result1 == result2}") # True
使用数据库唯一约束
import sqlite3
class DatabaseCache:
def __init__(self, db_path=':memory:'):
self.conn = sqlite3.connect(db_path)
self.cursor = self.conn.cursor()
# 创建表，带唯一约束
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value TEXT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.commit()
def store(self, key, value):
"""使用INSERT OR IGNORE避免重复"""
self.cursor.execute(
"INSERT OR IGNORE INTO cache (key, value) VALUES (?, ?)",
(key, value)
)
self.conn.commit()
return self.cursor.rowcount > 0 # 返回是否是新插入
def get(self, key):
self.cursor.execute("SELECT value FROM cache WHERE key=?", (key,))
result = self.cursor.fetchone()
return result[0] if result else None
# 使用示例
db_cache = DatabaseCache()
db_cache.store("user:1", "Alice")
db_cache.store("user:1", "Bob") # 因为唯一约束，不会更新
print(db_cache.get("user:1")) # 输出: Alice
最佳实践建议
-
选择合适的缓存策略
- 小规模数据：内存字典
- 大规模数据：Redis/Memcached
- 需要持久化：数据库
-
缓存失效策略
- TTL（过期时间）
- LRU（最近最少使用）
- LFU（最不经常使用）
-
监控缓存命中率
class MonitoredCache: def __init__(self): self.hits = 0 self.misses = 0 self.cache = {} def get(self, key): if key in self.cache: self.hits += 1 return self.cache[key] self.misses += 1 return None def hit_rate(self): total = self.hits + self.misses return self.hits / total if total > 0 else 0
选择哪种方法取决于你的具体需求：数据大小、访问频率、是否需要持久化等因素。
