Python脚本如何数据库新增同步加缓存

本文目录导读：

1. 基础方案：写数据库 + 更新缓存
2. 使用装饰器实现自动化缓存同步
3. 读写分离 + 延迟双删策略
4. 使用消息队列实现异步缓存同步
5. 使用Redis管道批量处理
6. 使用示例
7. 最佳实践建议

我来详细介绍几种数据库新增同步加缓存的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}")

最佳实践建议

1. 事务一致性：使用数据库事务与缓存操作配合

2. 缓存更新策略：

   • 先更新数据库,再删除缓存
   • 使用延迟双删策略
   • 考虑使用消息队列异步处理

3. 异常处理：做好回滚和重试机制

4. 监控告警：监控缓存同步的延迟和失败率

选择合适的方案取决于你的具体需求：数据一致性要求、并发量、系统复杂度等。