本文目录导读：
我来详细介绍几种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（直接Redis清除）
- 复杂业务：方案3（版本号策略）
- 微服务架构：方案4（消息队列）
- 高并发：方案3 + 方案4 组合
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
