本文目录导读：
我来详细介绍几种数据库新增同步加缓存的Python实现方案：
基础方案：写数据库 + 更新缓存
import redis
import pymysql
import json
from typing import Dict, Any
class CacheSyncManager:
def __init__(self):
self.redis_client = redis.Redis(
host='localhost',
port=6379,
decode_responses=True
)
self.db = pymysql.connect(
host='localhost',
user='root',
password='password',
database='test'
)
def add_user_with_cache(self, user_data: Dict[str, Any]) -> int:
"""新增用户并同步缓存"""
cursor = self.db.cursor()
try:
# 1. 写入数据库
sql = """INSERT INTO users (name, email, age)
VALUES (%s, %s, %s)"""
cursor.execute(sql, (
user_data['name'],
user_data['email'],
user_data['age']
))
self.db.commit()
user_id = cursor.lastrowid
# 2. 更新缓存（添加新数据）
user_data['id'] = user_id
cache_key = f"user:{user_id}"
self.redis_client.setex(
cache_key,
3600, # 1小时过期
json.dumps(user_data)
)
# 3. 清除相关列表缓存（可选）
self.redis_client.delete("users:list")
return user_id
except Exception as e:
self.db.rollback()
raise e
finally:
cursor.close()
使用装饰器实现自动化缓存同步
from functools import wraps
import hashlib
def cache_sync(prefix: str, expire: int = 3600):
"""装饰器：自动处理数据库新增和缓存同步"""
def decorator(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
# 获取数据
result = func(self, *args, **kwargs)
if isinstance(result, dict) and 'id' in result:
# 自动同步缓存
cache_key = f"{prefix}:{result['id']}"
self.redis_client.setex(
cache_key,
expire,
json.dumps(result)
)
# 清除列表缓存
list_key = f"{prefix}:list"
self.redis_client.delete(list_key)
return result
return wrapper
return decorator
class UserService:
def __init__(self):
self.redis_client = redis.Redis(decode_responses=True)
self.db = pymysql.connect(
host='localhost',
user='root',
password='password',
database='test'
)
@cache_sync(prefix="user", expire=3600)
def add_user(self, user_data: Dict) -> Dict:
"""新增用户"""
cursor = self.db.cursor()
try:
sql = """INSERT INTO users (name, email) VALUES (%s, %s)"""
cursor.execute(sql, (user_data['name'], user_data['email']))
self.db.commit()
user_id = cursor.lastrowid
return {
'id': user_id,
'name': user_data['name'],
'email': user_data['email']
}
except Exception as e:
self.db.rollback()
raise e
finally:
cursor.close()
读写分离 + 延迟双删策略
import time
from threading import Lock
class DoubleDeleteCacheSync:
def __init__(self):
self.redis_client = redis.Redis(decode_responses=True)
self.write_db = pymysql.connect(
host='write_host', # 主库
user='root',
password='password',
database='test'
)
self.read_db = pymysql.connect(
host='read_host', # 从库
user='root',
password='password',
database='test'
)
self.lock = Lock()
def safe_add_with_cache(self, table: str, data: Dict) -> int:
"""安全的缓存同步（延迟双删）"""
with self.lock:
cursor = self.write_db.cursor()
try:
# 1. 写入主库
columns = ', '.join(data.keys())
values = ', '.join(['%s'] * len(data))
sql = f"INSERT INTO {table} ({columns}) VALUES ({values})"
cursor.execute(sql, list(data.values()))
self.write_db.commit()
record_id = cursor.lastrowid
# 2. 第一次删除缓存
cache_key = f"{table}:{record_id}"
self.redis_client.delete(cache_key)
# 3. 延迟删除（处理主从同步延迟）
time.sleep(0.1) # 根据实际情况调整
self.redis_client.delete(cache_key)
# 4. 更新列表缓存
list_key = f"{table}:list"
self.redis_client.delete(list_key)
return record_id
except Exception as e:
self.write_db.rollback()
raise e
finally:
cursor.close()
def get_with_cache(self, table: str, record_id: int) -> Dict:
"""从缓存读取，缓存miss则从数据库加载"""
cache_key = f"{table}:{record_id}"
# 1. 查询缓存
cached_data = self.redis_client.get(cache_key)
if cached_data:
return json.loads(cached_data)
# 2. 从从库查询
cursor = self.read_db.cursor()
try:
sql = f"SELECT * FROM {table} WHERE id = %s"
cursor.execute(sql, (record_id,))
row = cursor.fetchone()
if row:
columns = [desc[0] for desc in cursor.description]
data = dict(zip(columns, row))
# 3. 写入缓存
self.redis_client.setex(
cache_key,
3600,
json.dumps(data)
)
return data
return None
finally:
cursor.close()
使用消息队列实现异步缓存同步
import threading
import queue
from datetime import datetime
class AsyncCacheSync:
def __init__(self):
self.redis_client = redis.Redis(decode_responses=True)
self.db = pymysql.connect(
host='localhost',
user='root',
password='password',
database='test'
)
self.sync_queue = queue.Queue()
self.start_sync_worker()
def start_sync_worker(self):
"""启动异步同步工作线程"""
def worker():
while True:
try:
task = self.sync_queue.get(timeout=1)
self._process_sync_task(task)
except queue.Empty:
continue
except Exception as e:
print(f"Sync error: {e}")
thread = threading.Thread(target=worker, daemon=True)
thread.start()
def _process_sync_task(self, task: Dict):
"""处理同步任务"""
action = task.get('action')
table = task.get('table')
record_id = task.get('id')
data = task.get('data')
if action == 'add':
# 更新缓存
cache_key = f"{table}:{record_id}"
self.redis_client.setex(
cache_key,
3600,
json.dumps(data)
)
# 更新列表缓存（使用有序集合）
list_key = f"{table}:list"
self.redis_client.zadd(
list_key,
{f"{record_id}": datetime.now().timestamp()}
)
# 设置列表缓存过期
self.redis_client.expire(list_key, 7200)
def add_user_async(self, user_data: Dict) -> int:
"""异步新增用户"""
cursor = self.db.cursor()
try:
# 1. 写入数据库
sql = """INSERT INTO users (name, email, age)
VALUES (%s, %s, %s)"""
cursor.execute(sql, (
user_data['name'],
user_data['email'],
user_data['age']
))
self.db.commit()
user_id = cursor.lastrowid
# 2. 异步更新缓存
user_data['id'] = user_id
self.sync_queue.put({
'action': 'add',
'table': 'users',
'id': user_id,
'data': user_data
})
return user_id
except Exception as e:
self.db.rollback()
raise e
finally:
cursor.close()
使用Redis管道批量处理
class BatchCacheSync:
def __init__(self):
self.redis_client = redis.Redis(decode_responses=True)
self.db = pymysql.connect(
host='localhost',
user='root',
password='password',
database='test'
)
def batch_add_with_cache(self, users_data: list) -> list:
"""批量新增用户并同步缓存"""
cursor = self.db.cursor()
user_ids = []
try:
# 1. 批量写入数据库
sql = """INSERT INTO users (name, email, age)
VALUES (%s, %s, %s)"""
values = [
(u['name'], u['email'], u['age'])
for u in users_data
]
cursor.executemany(sql, values)
self.db.commit()
# 获取插入的ID
for i in range(len(users_data)):
user_id = cursor.lastrowid - len(users_data) + 1 + i
user_ids.append(user_id)
# 2. 使用Redis管道批量设置缓存
pipeline = self.redis_client.pipeline()
for i, user_id in enumerate(user_ids):
cache_key = f"user:{user_id}"
user_data = {
'id': user_id,
**users_data[i]
}
pipeline.setex(cache_key, 3600, json.dumps(user_data))
# 3. 清除列表缓存
pipeline.delete("users:list")
# 执行管道
pipeline.execute()
return user_ids
except Exception as e:
self.db.rollback()
raise e
finally:
cursor.close()
使用示例
# 使用示例
if __name__ == "__main__":
# 基础方案
manager = CacheSyncManager()
user_id = manager.add_user_with_cache({
'name': '张三',
'email': 'zhangsan@example.com',
'age': 25
})
print(f"新增用户ID: {user_id}")
# 批量新增
batch_manager = BatchCacheSync()
user_ids = batch_manager.batch_add_with_cache([
{'name': '李四', 'email': 'lisi@example.com', 'age': 30},
{'name': '王五', 'email': 'wangwu@example.com', 'age': 28}
])
print(f"批量新增用户IDs: {user_ids}")
最佳实践建议
-
事务一致性：使用数据库事务与缓存操作配合
-
缓存更新策略：
- 先更新数据库,再删除缓存
- 使用延迟双删策略
- 考虑使用消息队列异步处理
-
异常处理：做好回滚和重试机制
-
监控告警：监控缓存同步的延迟和失败率
选择合适的方案取决于你的具体需求：数据一致性要求、并发量、系统复杂度等。
版权声明：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直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
