本文目录导读：
在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
使用建议
-
选择合适的方案：
- 简单场景：使用Redis事务+Lua脚本
- 强一致性需求：使用两阶段提交
- 需要回滚能力：使用WAL模式
- 高并发场景：使用分布式锁
-
性能考虑：
- 减少锁的持有时间
- 使用乐观锁适合读多写少场景
- 批量操作使用pipeline
-
错误处理：
- 实现超时机制
- 提供重试策略
- 记录详细的错误日志
-
监控和告警：
- 监控事务成功率
- 监控锁等待时间
- 设置合理的告警阈值
这些方案可以根据具体业务场景选择或组合使用,以达到最佳的一致性保证和性能平衡。
