Python脚本如何数据库更新同步清缓存

本文目录导读：

1. 使用Redis缓存（推荐）
2. 使用装饰器模式
3. 使用缓存版本号策略
4. 使用数据库触发器 + 消息队列
5. 综合使用示例
6. 关键最佳实践

我来详细介绍几种Python数据库更新后同步清除缓存的方案：

使用Redis缓存（推荐）

基础实现

import redis
import pymysql
from functools import wraps
# 连接Redis
cache = redis.Redis(host='localhost', port=6379, db=0)
class DatabaseManager:
    def __init__(self):
        self.db = pymysql.connect(
            host='localhost',
            user='user',
            password='password',
            database='test'
        )
    def update_data(self, table_name, data, condition):
        """更新数据并清除相关缓存"""
        cursor = self.db.cursor()
        try:
            # 执行数据库更新
            sql = f"UPDATE {table_name} SET {data} WHERE {condition}"
            cursor.execute(sql)
            self.db.commit()
            # 清除该表的相关缓存
            self.clear_table_cache(table_name)
            return True
        except Exception as e:
            self.db.rollback()
            raise e
        finally:
            cursor.close()
    def clear_table_cache(self, table_name):
        """清除指定表的所有缓存"""
        # 清除特定模式的所有缓存键
        pattern = f"cache:{table_name}:*"
        for key in cache.scan_iter(pattern):
            cache.delete(key)
    def get_cached_data(self, table_name, query_key):
        """获取缓存数据"""
        cache_key = f"cache:{table_name}:{query_key}"
        data = cache.get(cache_key)
        if data is None:
            # 从数据库查询
            data = self.query_database(table_name, query_key)
            # 设置缓存，过期时间3600秒
            cache.setex(cache_key, 3600, data)
        return data

使用装饰器模式

import hashlib
import json
from functools import wraps
def cache_clear_on_update(cache_instance, table_pattern):
    """装饰器：数据库更新后自动清除缓存"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 执行原函数（数据库更新操作）
            result = func(*args, **kwargs)
            # 清除相关缓存
            pattern = f"cache:{table_pattern}:*"
            for key in cache_instance.scan_iter(pattern):
                cache_instance.delete(key)
            return result
        return wrapper
    return decorator
# 使用示例
@cache_clear_on_update(cache, "users")
def update_user(user_id, new_data):
    """更新用户信息"""
    cursor = db.cursor()
    cursor.execute(
        "UPDATE users SET name=%s, email=%s WHERE id=%s",
        (new_data['name'], new_data['email'], user_id)
    )
    db.commit()
# 或更复杂的版本，支持多表
def multi_table_cache_clear(tables):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            for table in tables:
                pattern = f"cache:{table}:*"
                for key in cache.scan_iter(pattern):
                    cache.delete(key)
            return result
        return wrapper
    return decorator
@multi_table_cache_clear(['users', 'orders'])
def update_user_and_order(user_id, order_id):
    # 更新多个表
    pass

使用缓存版本号策略

class VersionedCache:
    """基于版本号的缓存管理"""
    def __init__(self, redis_client):
        self.redis = redis_client
    def get_version(self, table_name):
        """获取表版本号"""
        version = self.redis.get(f"version:{table_name}")
        return int(version) if version else 0
    def increment_version(self, table_name):
        """增加表版本号"""
        self.redis.incr(f"version:{table_name}")
    def get_cache_key(self, table_name, query_params):
        """生成带版本号的缓存键"""
        version = self.get_version(table_name)
        params_hash = hashlib.md5(
            json.dumps(query_params, sort_keys=True).encode()
        ).hexdigest()
        return f"cache:{table_name}:v{version}:{params_hash}"
    def set_cache(self, table_name, query_params, data, timeout=3600):
        """设置缓存"""
        cache_key = self.get_cache_key(table_name, query_params)
        self.redis.setex(cache_key, timeout, json.dumps(data))
    def get_cache(self, table_name, query_params):
        """获取缓存"""
        cache_key = self.get_cache_key(table_name, query_params)
        data = self.redis.get(cache_key)
        return json.loads(data) if data else None
    def on_data_update(self, table_name):
        """数据更新时调用"""
        self.increment_version(table_name)
        # 可选：清理旧版本的缓存
        self.clean_old_versions(table_name)
    def clean_old_versions(self, table_name, keep_latest=3):
        """清理旧版本缓存"""
        current_version = self.get_version(table_name)
        pattern = f"cache:{table_name}:v*"
        for key in self.redis.scan_iter(pattern):
            # 提取版本号
            try:
                parts = key.decode().split(':v')
                if len(parts) > 1:
                    version = int(parts[1].split(':')[0])
                    if version < current_version - keep_latest + 1:
                        self.redis.delete(key)
            except (ValueError, IndexError):
                continue

使用数据库触发器 + 消息队列

import pika
import pymysql
import json
class CacheSyncManager:
    def __init__(self):
        # 连接RabbitMQ
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost')
        )
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='cache_clear')
        # 连接数据库
        self.db = pymysql.connect(
            host='localhost',
            user='user',
            password='password',
            database='test'
        )
    def update_with_trigger(self, table_name, data, condition):
        """更新数据并通过消息队列通知缓存清理"""
        cursor = self.db.cursor()
        try:
            # 执行更新
            sql = f"UPDATE {table_name} SET {data} WHERE {condition}"
            cursor.execute(sql)
            self.db.commit()
            # 发送缓存清除消息
            message = json.dumps({
                'action': 'clear_cache',
                'table': table_name,
                'timestamp': time.time()
            })
            self.channel.basic_publish(
                exchange='',
                routing_key='cache_clear',
                body=message
            )
            return True
        except Exception as e:
            self.db.rollback()
            raise e
        finally:
            cursor.close()
    def start_cache_cleaner(self):
        """启动缓存清理消费者"""
        def callback(ch, method, properties, body):
            data = json.loads(body)
            if data['action'] == 'clear_cache':
                self.clear_table_cache(data['table'])
                print(f"Cleared cache for table: {data['table']}")
        self.channel.basic_consume(
            queue='cache_clear',
            on_message_callback=callback,
            auto_ack=True
        )
        self.channel.start_consuming()
# 使用示例
sync_manager = CacheSyncManager()
# 在另一个线程或进程中运行
# threading.Thread(target=sync_manager.start_cache_cleaner).start()

综合使用示例

class DatabaseCacheSync:
    def __init__(self, db_config, redis_config):
        self.db = self._connect_db(db_config)
        self.cache = redis.Redis(**redis_config)
        self.version_cache = VersionedCache(self.cache)
    def _connect_db(self, config):
        """连接数据库"""
        return pymysql.connect(**config)
    def execute_update(self, sql, params=None, tables_to_clear=None):
        """执行更新并同步清除缓存"""
        cursor = self.db.cursor()
        try:
            cursor.execute(sql, params or ())
            self.db.commit()
            # 清除缓存
            if tables_to_clear:
                for table in tables_to_clear:
                    # 方式1：直接清除所有相关缓存
                    self._clear_cache_by_pattern(table)
                    # 方式2：更新版本号
                    self.version_cache.on_data_update(table)
            return cursor.lastrowid
        except Exception as e:
            self.db.rollback()
            raise e
        finally:
            cursor.close()
    def _clear_cache_by_pattern(self, table_name):
        """通过模式匹配清除缓存"""
        pattern = f"cache:{table_name}:*"
        deleted = 0
        for key in self.cache.scan_iter(pattern):
            self.cache.delete(key)
            deleted += 1
        return deleted
    def get_or_compute(self, table_name, query_key, query_func, timeout=3600):
        """获取缓存或计算并缓存结果"""
        # 优先使用版本缓存
        cached = self.version_cache.get_cache(table_name, query_key)
        if cached:
            return cached
        # 计算新值
        result = query_func()
        # 设置缓存
        self.version_cache.set_cache(table_name, query_key, result, timeout)
        return result
# 使用示例
sync = DatabaseCacheSync(
    db_config={
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'test'
    },
    redis_config={
        'host': 'localhost',
        'port': 6379,
        'db': 0
    }
)
# 更新数据并清除缓存
sync.execute_update(
    "UPDATE users SET name=%s WHERE id=%s",
    ('New Name', 1),
    tables_to_clear=['users', 'user_stats']
)

关键最佳实践

1. 避免缓存雪崩：设置随机的过期时间
2. 缓存预热：系统启动时加载热点数据
3. 异步清理：使用消息队列异步处理缓存清理
4. 监控告警：监控缓存命中率和清理效率
5. 降级策略：缓存不可用时直接查询数据库

选择哪种方案取决于你的具体需求：

• 小项目：方案1（直接Redis清除）
• 复杂业务：方案3（版本号策略）
• 微服务架构：方案4（消息队列）
• 高并发：方案3 + 方案4 组合