Python脚本如何实现缓存事务一致性

本文目录导读：

1. 使用Redis事务 + Lua脚本
2. 两阶段提交协议（2PC）
3. 基于分布式锁的实现
4. 使用WAL（Write-Ahead Logging）
5. JSON schema验证示例
6. 使用建议

在Python中实现缓存事务一致性是一个复杂的工程挑战,以下是几种常见的实现方案：

使用Redis事务 + Lua脚本

import redis
import json
class RedisCacheWithTransaction:
    def __init__(self, host='localhost', port=6379):
        self.redis = redis.Redis(host=host, port=port, decode_responses=True)
    def execute_atomic_operation(self, cache_key, db_func):
        """
        使用Lua脚本实现原子操作
        """
        lua_script = """
        -- 检查缓存状态
        local cache_value = redis.call('GET', KEYS[1])
        -- 执行业务逻辑（这里只是示例）
        local new_value = ARGV[1]
        -- 更新缓存
        redis.call('SET', KEYS[1], new_value)
        return new_value
        """
        # 加载脚本
        script = self.redis.register_script(lua_script)
        # 执行原子操作
        return script(keys=[cache_key], args=[db_func()])

两阶段提交协议（2PC）

import time
from typing import Dict, Any
import pickle
class TwoPhaseCommitCache:
    def __init__(self, cache_client):
        self.cache = cache_client
        self.transactions: Dict[str, list] = {}
    def prepare(self, transaction_id: str, operations: list) -> bool:
        """
        第一阶段：准备阶段
        """
        try:
            # 临时存储所有待写入的数据
            prepared_data = []
            for op in operations:
                key = op['key']
                value = op['value']
                # 创建备份
                old_value = self.cache.get(key)
                prepared_data.append((key, old_value))
                # 设置锁定标记
                lock_key = f"lock:{key}:{transaction_id}"
                self.cache.set(lock_key, 'prepared', ex=60)  # 60秒超时
                # 临时存储新值
                temp_key = f"temp:{key}:{transaction_id}"
                self.cache.set(temp_key, pickle.dumps(value))
                # 保存备份
                if old_value:
                    backup_key = f"backup:{key}:{transaction_id}"
                    self.cache.set(backup_key, old_value)
            self.transactions[transaction_id] = prepared_data
            return True
        except Exception as e:
            print(f"准备阶段失败: {e}")
            self.rollback(transaction_id)
            return False
    def commit(self, transaction_id: str) -> bool:
        """
        第二阶段：提交阶段
        """
        try:
            # 将所有临时数据正式写入缓存
            temp_pattern = f"temp:{transaction_id}:*"
            temp_keys = self.cache.keys(pattern=temp_pattern)
            for temp_key in temp_keys:
                # 解析原始key
                original_key = temp_key.split(':', 1)[1].split(':', 1)[1]
                actual_key = original_key.split(':', 2)[2]  # 修正这里
                # 获取临时数据
                temp_value = self.cache.get(temp_key)
                if temp_value:
                    # 写入正式缓存
                    self.cache.set(actual_key, temp_value)
                    # 清理临时数据
                    self.cache.delete(temp_key)
                    # 移除锁
                    lock_key = f"lock:{actual_key}:{transaction_id}"
                    self.cache.delete(lock_key)
                    # 删除备份
                    backup_key = f"backup:{actual_key}:{transaction_id}"
                    self.cache.delete(backup_key)
            # 清理事务记录
            if transaction_id in self.transactions:
                del self.transactions[transaction_id]
            return True
        except Exception as e:
            print(f"提交阶段失败: {e}")
            self.rollback(transaction_id)
            return False
    def rollback(self, transaction_id: str):
        """
        回滚操作
        """
        try:
            # 恢复所有备份数据
            if transaction_id in self.transactions:
                for key, old_value in self.transactions[transaction_id]:
                    if old_value:
                        self.cache.set(key, old_value)
                    else:
                        self.cache.delete(key)
                # 清理所有临时数据
                temp_pattern = f"temp:{transaction_id}:*"
                temp_keys = self.cache.keys(pattern=temp_pattern)
                for key in temp_keys:
                    self.cache.delete(key)
                # 清理锁
                lock_pattern = f"lock:*:{transaction_id}"
                lock_keys = self.cache.keys(pattern=lock_pattern)
                for key in lock_keys:
                    self.cache.delete(key)
                # 清理备份
                backup_pattern = f"backup:*:{transaction_id}"
                backup_keys = self.cache.keys(pattern=backup_pattern)
                for key in backup_keys:
                    self.cache.delete(key)
                del self.transactions[transaction_id]
        except Exception as e:
            print(f"回滚失败: {e}")

基于分布式锁的实现

import threading
import time
from contextlib import contextmanager
class DistributedTransactionCache:
    def __init__(self, cache_client, lock_timeout=30):
        self.cache = cache_client
        self.lock_timeout = lock_timeout
        self.local_buffers = threading.local()
    def acquire_lock(self, key: str) -> bool:
        """
        获取分布式锁
        """
        lock_key = f"dlock:{key}"
        # 使用SET NX EX实现分布式锁
        result = self.cache.set(lock_key, 'locked', 
                                nx=True, 
                                ex=self.lock_timeout)
        return result is True
    def release_lock(self, key: str):
        """
        释放锁
        """
        lock_key = f"dlock:{key}"
        # 使用Lua脚本确保原子性释放锁
        lua_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        script = self.cache.register_script(lua_script)
        script(keys=[lock_key], args=['locked'])
    @contextmanager
    def transaction(self):
        """
        事务上下文管理器
        """
        # 初始化事务缓冲区
        self.local_buffers.buffer = []
        self.local_buffers.locks = []
        try:
            yield self
            # 自动提交
            self._commit()
        except Exception as e:
            # 自动回滚
            self._rollback()
            raise e
    def set_with_transaction(self, key: str, value):
        """
        在事务中设置缓存值
        """
        # 获取锁
        if self.acquire_lock(key):
            self.local_buffers.locks.append(key)
            # 保存旧值用于回滚
            old_value = self.cache.get(key)
            self.local_buffers.buffer.append({
                'key': key,
                'old_value': old_value,
                'new_value': value
            })
    def _commit(self):
        """
        提交事务
        """
        try:
            for op in self.local_buffers.buffer:
                # 最终写入缓存
                self.cache.set(op['key'], op['new_value'])
        finally:
            # 释放所有锁
            for lock_key in self.local_buffers.locks:
                self.release_lock(lock_key)
            self.local_buffers.buffer = []
            self.local_buffers.locks = []
    def _rollback(self):
        """
        回滚事务
        """
        try:
            for op in reversed(self.local_buffers.buffer):
                # 恢复旧值
                if op['old_value'] is not None:
                    self.cache.set(op['key'], op['old_value'])
                else:
                    self.cache.delete(op['key'])
        finally:
            # 释放所有锁
            for lock_key in self.local_buffers.locks:
                self.release_lock(lock_key)
            self.local_buffers.buffer = []
            self.local_buffers.locks = []

使用WAL（Write-Ahead Logging）

import sqlite3
import json
from datetime import datetime
class WALCacheTransaction:
    def __init__(self, cache_client, wal_db='cache_wal.db'):
        self.cache = cache_client
        self.wal_conn = sqlite3.connect(wal_db)
        self._init_wal_table()
    def _init_wal_table(self):
        """初始化WAL表"""
        cursor = self.wal_conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS wal_logs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                transaction_id TEXT,
                operation_type TEXT,
                cache_key TEXT,
                old_value TEXT,
                new_value TEXT,
                status TEXT,
                created_at TIMESTAMP
            )
        ''')
        self.wal_conn.commit()
    def begin_transaction(self, transaction_id: str):
        """
        开始事务
        """
        # 在WAL中记录事务开始
        cursor = self.wal_conn.cursor()
        cursor.execute('''
            INSERT INTO wal_logs 
            (transaction_id, operation_type, status, created_at)
            VALUES (?, 'BEGIN', 'STARTED', ?)
        ''', (transaction_id, datetime.now()))
        self.wal_conn.commit()
    def log_operation(self, transaction_id: str, operation_type: str,
                     cache_key: str, new_value, old_value=None):
        """
        记录操作到WAL
        """
        cursor = self.wal_conn.cursor()
        cursor.execute('''
            INSERT INTO wal_logs 
            (transaction_id, operation_type, cache_key, 
             old_value, new_value, status, created_at)
            VALUES (?, ?, ?, ?, ?, 'PENDING', ?)
        ''', (transaction_id, operation_type, cache_key,
              json.dumps(old_value) if old_value else None,
              json.dumps(new_value) if new_value else None,
              datetime.now()))
        self.wal_conn.commit()
    def commit_transaction(self, transaction_id: str):
        """
        提交事务
        """
        try:
            # 获取所有待处理的操作
            cursor = self.wal_conn.cursor()
            cursor.execute('''
                SELECT id, operation_type, cache_key, new_value
                FROM wal_logs
                WHERE transaction_id = ? AND status = 'PENDING'
                ORDER BY id
            ''', (transaction_id,))
            operations = cursor.fetchall()
            # 执行实际操作
            for op in operations:
                op_id, op_type, cache_key, new_value = op
                if op_type == 'SET':
                    self.cache.set(cache_key, json.loads(new_value))
                elif op_type == 'DELETE':
                    self.cache.delete(cache_key)
                # 更新WAL状态
                cursor.execute('''
                    UPDATE wal_logs 
                    SET status = 'COMMITTED'
                    WHERE id = ?
                ''', (op_id,))
            # 标记事务完成
            cursor.execute('''
                UPDATE wal_logs
                SET status = 'COMPLETED'
                WHERE transaction_id = ? AND operation_type = 'BEGIN'
            ''', (transaction_id,))
            self.wal_conn.commit()
        except Exception as e:
            self.wal_conn.rollback()
            raise e
    def rollback_transaction(self, transaction_id: str):
        """
        回滚事务
        """
        cursor = self.wal_conn.cursor()
        # 获取所有操作（逆序执行恢复）
        cursor.execute('''
            SELECT id, operation_type, cache_key, old_value
            FROM wal_logs
            WHERE transaction_id = ? AND status = 'PENDING'
            ORDER BY id DESC
        ''', (transaction_id,))
        operations = cursor.fetchall()
        # 执行回滚
        for op in operations:
            op_id, op_type, cache_key, old_value = op
            if old_value:
                self.cache.set(cache_key, json.loads(old_value))
            else:
                self.cache.delete(cache_key)
            # 更新WAL状态
            cursor.execute('''
                UPDATE wal_logs 
                SET status = 'ROLLED_BACK'
                WHERE id = ?
            ''', (op_id,))
        # 标记事务回滚
        cursor.execute('''
            UPDATE wal_logs
            SET status = 'ROLLED_BACK'
            WHERE transaction_id = ? AND operation_type = 'BEGIN'
        ''', (transaction_id,))
        self.wal_conn.commit()
    def recover(self):
        """
        恢复未完成的事务
        """
        cursor = self.wal_conn.cursor()
        # 查找所有未完成的事务
        cursor.execute('''
            SELECT DISTINCT transaction_id
            FROM wal_logs
            WHERE status IN ('STARTED', 'PENDING')
            AND operation_type = 'BEGIN'
        ''')
        incomplete_transactions = cursor.fetchall()
        for (transaction_id,) in incomplete_transactions:
            # 尝试恢复
            try:
                self.rollback_transaction(transaction_id)
                print(f"事务 {transaction_id} 已回滚")
            except Exception as e:
                print(f"恢复事务 {transaction_id} 时出错: {e}")

JSON schema验证示例

from jsonschema import validate, ValidationError
class TransactionWithValidation:
    def __init__(self, cache_client):
        self.cache = cache_client
        self.transaction_schema = {
            "type": "object",
            "properties": {
                "transactions": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "operation": {
                                "type": "string",
                                "enum": ["SET", "DELETE", "UPDATE"]
                            },
                            "key": {"type": "string"},
                            "value": {},
                            "expected_old_value": {}
                        },
                        "required": ["operation", "key"]
                    }
                }
            },
            "required": ["transactions"]
        }
    def validate_consistency(self, data: dict) -> bool:
        """
        验证事务一致性
        """
        try:
            validate(instance=data, schema=self.transaction_schema)
            # 额外的一致性检查
            for transaction in data['transactions']:
                key = transaction['key']
                operation = transaction['operation']
                # 检查乐观锁条件
                if 'expected_old_value' in transaction:
                    current_value = self.cache.get(key)
                    if current_value != transaction['expected_old_value']:
                        return False
                # 检查操作合法性
                if operation == 'UPDATE' and not self.cache.exists(key):
                    return False
            return True
        except ValidationError as e:
            print(f"验证失败: {e}")
            return False

使用建议

1. 选择合适的方案：

   • 简单场景：使用Redis事务+Lua脚本
   • 强一致性需求：使用两阶段提交
   • 需要回滚能力：使用WAL模式
   • 高并发场景：使用分布式锁

2. 性能考虑：

   • 减少锁的持有时间
   • 使用乐观锁适合读多写少场景
   • 批量操作使用pipeline

3. 错误处理：

   • 实现超时机制
   • 提供重试策略
   • 记录详细的错误日志

4. 监控和告警：

   • 监控事务成功率
   • 监控锁等待时间
   • 设置合理的告警阈值

这些方案可以根据具体业务场景选择或组合使用,以达到最佳的一致性保证和性能平衡。