Python脚本如何避免缓存数据重复存储

wen python案例 2026-06-26 244

本文目录导读：

使用字典/LRU缓存
自定义缓存类
使用集合检查重复
使用字典记录哈希值
带过期时间的缓存
使用数据库唯一约束
最佳实践建议

我来介绍几种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

选择哪种方法取决于你的具体需求：数据大小、访问频率、是否需要持久化等因素。

本文地址： https://www.bmjup.com/post/9524.html