Redis事务机制深度剖析：MULTI、EXEC与WATCH命令实战

MULTI：相当于"开始记录"，标记一个事务的开始EXEC：相当于"执行记录的所有命令"，执行事务队列中的所有命令DISCARD：相当于"撕掉记录纸"，取消事务，清空命令队列WATCH：相当于"盯住某个值"，监视一个或多个键，如果事务执行前这些键被修改，事务将失败UNWATCH：相当于"不再盯着看"，取消对所有键的监视这些命令组合使用，构成了Redis事务的完整处理流程。特性Redis事务Lua脚

sinat_27016095

1554人浏览 · 2025-10-22 15:00:00

sinat_27016095 · 2025-10-22 15:00:00 发布

在高并发系统设计中，Redis凭借其高性能特性已成为不可或缺的组件。而当我们需要保证一系列命令的原子性执行时，Redis事务机制便成为我们的得力助手。本文将带您深入探索Redis事务的内部机制，剖析MULTI、EXEC与WATCH命令的工作原理，并分享实战经验与最佳实践。

一、Redis事务基础

1. Redis事务的概念与特点

Redis事务允许我们将多个命令打包，然后一次性、按顺序地执行。这就像是我们去银行办理业务时，柜员会先记录下我们的所有需求，确认无误后再一次性完成所有操作，而不是每说一个需求就立即处理一个。

Redis事务的核心特点：

命令打包：将多个命令打包成一个整体
按序执行：命令按照放入队列的顺序依次执行
原子性：所有命令要么都执行，要么都不执行（但有例外情况）
隔离性：事务执行期间，其他客户端提交的命令不会插入到事务执行队列中

⚠️ 特别注意：Redis事务与传统数据库事务的最大区别在于，Redis不支持完全的回滚机制。如果事务中某个命令执行失败，其他命令仍会继续执行。

Redis事务与传统数据库事务的ACID特性对比：

ACID特性	传统关系型数据库	Redis事务	说明
原子性(A)	✅ 完全支持	⚠️ 部分支持	Redis事务不保证命令执行错误时的回滚
一致性©	✅ 完全支持	✅ 支持	Redis在事务执行前后数据是一致的
隔离性(I)	✅ 多种隔离级别	✅ 支持	Redis单线程模型天然支持隔离性
持久性(D)	✅ 完全支持	取决于持久化配置	根据RDB/AOF配置决定

2. Redis事务的核心命令介绍

进入Redis事务世界，我们需要掌握以下五个核心命令：

MULTI：相当于"开始记录"，标记一个事务的开始
EXEC：相当于"执行记录的所有命令"，执行事务队列中的所有命令
DISCARD：相当于"撕掉记录纸"，取消事务，清空命令队列
WATCH：相当于"盯住某个值"，监视一个或多个键，如果事务执行前这些键被修改，事务将失败
UNWATCH：相当于"不再盯着看"，取消对所有键的监视

这些命令组合使用，构成了Redis事务的完整处理流程。

二、MULTI与EXEC命令详解

从上一节的基础认识，我们进一步深入Redis事务的核心命令实现原理。

1. MULTI命令原理剖析

当客户端发送MULTI命令时，Redis服务器会将该客户端的状态从正常状态切换为事务状态，就像是在银行柜台按下了"开始办理"按钮：

CLIENT_NORMAL → CLIENT_MULTI

MULTI命令执行后的内部变化：

客户端状态标志位被设置为"事务状态"
Redis为该客户端创建一个命令队列（事务队列）
服务器返回OK，表示事务已开始

从此刻起，客户端发送的所有命令都不会立即执行，而是被追加到该客户端的命令队列中，服务器会返回QUEUED表示命令已入队。

💡 技术细节：Redis在内部为每个客户端维护了一个redisClient结构，当执行MULTI后，该结构中的flags属性会设置CLIENT_MULTI标志，并初始化mstate字段用于存储事务状态。

2. EXEC命令执行流程

当客户端发送EXEC命令时，Redis会执行以下步骤：

检查事务状态，确认是否可以执行
遍历命令队列，依次执行每个命令
将每个命令的执行结果依次返回给客户端
清空命令队列，将客户端状态恢复为非事务状态

就像银行柜员在确认所有业务无误后，开始依次处理每一项业务并给出反馈。

EXEC命令执行流程图：

检查事务状态 → 遍历命令队列 → 执行命令并收集结果 → 清空队列 → 恢复客户端状态

Redis 2.6.5前后版本差异：

在Redis 2.6.5之前，如果事务队列中有语法错误的命令，EXEC命令会忽略错误继续执行其他命令。而在2.6.5及之后的版本中，如果事务中有语法错误（编译错误），整个事务会被拒绝执行。

3. 事务中常见错误类型及处理

Redis事务中的错误可分为两类，它们的处理方式截然不同：

1. 语法错误（编译期错误）

这类错误在命令入队时就能被检测到，例如：

MULTI
SET key value  // 正确命令
INCR key value  // 语法错误：INCR只需要一个参数
EXEC

在这种情况下，EXEC执行时会拒绝执行整个事务，返回错误信息。

2. 运行时错误

这类错误只有在命令实际执行时才会出现，例如：

MULTI
SET key "string value"
INCR key  // 运行时错误：无法对字符串执行INCR操作
EXEC

在这种情况下，Redis仍会执行所有命令，但会在返回结果中标记出错误。

⚠️ 重要提示：Redis事务不支持回滚！即使事务中某个命令执行失败，其他命令仍会继续执行，造成"部分失败"状态。这是Redis事务与传统数据库事务的重大区别。

部分失败情况下的数据状态：

在部分命令执行失败的情况下，Redis数据库会保留已成功执行命令的结果，不会回滚。这可能导致数据处于中间状态，应用程序需要自行处理这种情况。

三、WATCH命令与乐观锁机制

在前两节中，我们讨论了Redis事务的基本机制。然而，在多客户端并发环境下，我们还需要解决"条件变更"问题。这就是WATCH命令发挥作用的地方。

1. WATCH命令实现原理

WATCH命令实现了Redis的乐观锁机制。所谓乐观锁，就像是"相信大家都不会乱动我盯着的东西，但我会在最后确认一下"的机制。

乐观锁vs悲观锁：

悲观锁：假设一定会发生冲突，提前锁定资源（像是将银行柜台锁起来，一次只允许一个人办理业务）
乐观锁：假设冲突较少，只在提交时检查冲突（像是允许多人同时填单，但提交时检查信息是否有变化）

Redis使用乐观锁机制，因为它在高并发、读多写少的场景下性能更佳。

WATCH底层实现原理：

当执行WATCH命令时，Redis会在数据库中记录被监视的键与客户端的关联关系。具体来说：

服务器为每个被WATCH的键维护一个监视列表
在键被修改时，将相关客户端的REDIS_DIRTY_CAS标志设置为1
执行EXEC时，检查该标志，若为1则拒绝执行事务

💡 技术细节：Redis使用了一个简单但高效的版本号机制。每个键都有一个内部版本标记，当键被修改时，其版本号会增加。WATCH本质上就是记住键的当前版本号，在EXEC执行前比对版本是否变化。

2. CAS（Compare And Set）在Redis中的应用

CAS是乐观锁的一种实现方式，其核心理念是"比较并设置"：只有当目前值与预期值相同时，才进行更新操作。

Redis的WATCH-MULTI-EXEC流程本质上就是一个CAS操作：

WATCH记录键的当前状态
MULTI开始组装命令
EXEC时比较键的状态是否变化，未变化则执行，已变化则放弃

WATCH监视多个键的实现机制：

WATCH可以同时监视多个键，任何一个键的变化都会导致事务失败。在内部实现上，Redis为每个客户端维护了一个被监视键的字典，记录所有被该客户端监视的键。

键修改触发事务取消的全过程：

客户端A执行WATCH key，Redis记录key的当前版本
客户端A执行MULTI，进入事务状态
客户端B修改了key的值，Redis将所有监视该key的客户端标记为"脏"
客户端A执行EXEC，Redis检查发现客户端A被标记为"脏"，拒绝执行事务
Redis返回空值(nil)给客户端A，表示事务执行失败

3. WATCH实战使用模式

标准的WATCH使用模式：

WATCH key1 key2 ... # 监视一个或多个键
value = GET key     # 获取当前值
# 根据当前值进行一些计算...
MULTI               # 开始事务
SET key new-value   # 设置新值
EXEC                # 尝试执行事务
# 检查EXEC返回值，如果是nil则表示事务失败，需要重试

注意事项与最佳实践：

设置合理的重试次数：事务失败后应当重试，但需要限制重试次数，避免无限循环
监视最少的键：只监视真正需要的键，减少冲突概率
尽量缩短事务时间：监视到执行的时间越短，冲突可能性越小
避免在MULTI与EXEC之间进行耗时操作：这会增加事务冲突的概率

常见陷阱：

过度监视：监视了过多的键，导致事务频繁失败
忘记检查EXEC返回值：没有处理事务可能失败的情况
在WATCH后没有立即读取值：导致判断逻辑基于过期数据

四、事务在实际项目中的应用场景

理论讲解完毕，让我们看看Redis事务在实际项目中的应用场景。

1. 库存管理与秒杀系统

秒杀系统是Redis事务的经典应用场景，其核心是防止超卖问题。

使用Redis事务实现库存扣减的示例代码：

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)
product_id = "iphone15"
inventory_key = f"inventory:{product_id}"

def purchase_with_retry(user_id, product_id, retry_times=3):
    """使用Redis事务进行商品购买"""
    for i in range(retry_times):
        # 监视库存键
        r.watch(inventory_key)
        
        # 获取当前库存
        current_inventory = int(r.get(inventory_key) or 0)
        if current_inventory <= 0:
            # 库存不足，取消监视并返回
            r.unwatch()
            return {"success": False, "message": "库存不足"}
        
        try:
            # 开始事务
            pipe = r.pipeline(transaction=True)
            pipe.multi()
            
            # 扣减库存
            pipe.decr(inventory_key)
            # 记录购买记录
            pipe.sadd(f"purchased:{product_id}", user_id)
            # 执行事务
            result = pipe.execute()
            
            # 事务成功执行
            return {"success": True, "message": "购买成功", "new_inventory": result[0]}
        except redis.exceptions.WatchError:
            # 事务执行失败，说明库存已被修改
            time.sleep(0.1)  # 短暂休眠后重试
            continue
    
    # 超过重试次数
    return {"success": False, "message": "系统繁忙，请稍后再试"}

# 初始设置库存
r.set(inventory_key, 100)

# 模拟购买
result = purchase_with_retry("user123", "iphone15")
print(result)

性能优化与限流结合方案：

预热库存：系统启动时将库存加载到Redis
分布式限流：使用Redis计数器实现请求限流
异步确认：库存扣减成功后，异步处理订单创建
批量处理：在极高并发下，可考虑使用Lua脚本代替事务

2. 账户余额与积分管理

账户资金操作是另一个需要事务保证的经典场景。

账户余额转账的实现示例：

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def transfer_money(from_user, to_user, amount, max_retries=5):
    """转账操作，从一个账户转移金额到另一个账户"""
    from_key = f"balance:{from_user}"
    to_key = f"balance:{to_user}"
    
    for _ in range(max_retries):
        # 监视两个账户
        r.watch(from_key, to_key)
        
        # 获取当前余额
        from_balance = int(r.get(from_key) or 0)
        to_balance = int(r.get(to_key) or 0)
        
        # 检查余额是否充足
        if from_balance < amount:
            r.unwatch()
            return {"success": False, "message": "余额不足"}
        
        try:
            # 开始事务
            pipe = r.pipeline(transaction=True)
            pipe.multi()
            
            # 执行转账
            pipe.decrby(from_key, amount)
            pipe.incrby(to_key, amount)
            
            # 记录交易历史
            transaction_id = int(time.time() * 1000)
            transaction_key = f"transaction:{transaction_id}"
            pipe.hmset(transaction_key, {
                "from": from_user,
                "to": to_user,
                "amount": amount,
                "timestamp": int(time.time())
            })
            
            # 执行事务
            result = pipe.execute()
            return {
                "success": True, 
                "message": "转账成功", 
                "from_balance": result[0],
                "to_balance": result[1],
                "transaction_id": transaction_id
            }
            
        except redis.exceptions.WatchError:
            # 监视的键被修改，重试
            continue
    
    return {"success": False, "message": "系统繁忙，请稍后再试"}

# 初始化账户余额
r.set("balance:user1", 1000)
r.set("balance:user2", 500)

# 执行转账
result = transfer_money("user1", "user2", 200)
print(result)

避免资金丢失的安全策略：

事务日志：每笔交易都记录详细的日志
定期对账：系统定期检查账户总额是否平衡
失败重试机制：关键操作失败后的自动重试策略
分布式锁：对大额交易使用分布式锁进行额外保护

3. 排行榜与计数器更新

Redis的Sorted Set是实现排行榜的理想数据结构，而事务可以确保排行榜更新的原子性。

排行榜原子性更新实现：

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def update_leaderboard(user_id, score_increment, max_retries=3):
    """更新用户在排行榜上的分数"""
    leaderboard_key = "game:leaderboard"
    user_score_key = f"user:score:{user_id}"
    
    for _ in range(max_retries):
        # 监视用户分数和排行榜
        r.watch(user_score_key, leaderboard_key)
        
        # 获取当前分数
        current_score = int(r.get(user_score_key) or 0)
        new_score = current_score + score_increment
        
        try:
            # 开始事务
            pipe = r.pipeline(transaction=True)
            pipe.multi()
            
            # 更新用户分数
            pipe.set(user_score_key, new_score)
            
            # 更新排行榜
            pipe.zadd(leaderboard_key, {user_id: new_score})
            
            # 记录更新时间
            pipe.hset(f"user:last_update:{user_id}", "timestamp", int(time.time()))
            
            # 执行事务
            pipe.execute()
            
            return {
                "success": True,
                "user_id": user_id,
                "new_score": new_score,
                "rank": r.zrevrank(leaderboard_key, user_id) + 1  # 查询最新排名
            }
            
        except redis.exceptions.WatchError:
            # 监视的键被修改，重试
            continue
    
    return {"success": False, "message": "更新失败，请重试"}

# 初始化一些用户分数
r.set("user:score:player1", 100)
r.set("user:score:player2", 85)
r.zadd("game:leaderboard", {"player1": 100, "player2": 85})

# 更新分数
result = update_leaderboard("player1", 15)
print(result)

大规模系统的性能优化：

分片排行榜：对大型排行榜进行分片处理
定时计算：对于变更频繁的排行榜，定时批量更新而非实时更新
异步更新策略：将排名更新与分数更新分离
热点数据缓存：缓存前N名或周围排名的数据

五、Redis事务的性能分析与优化

前面几节我们已经了解了Redis事务的原理与应用，现在让我们关注性能方面的考量。

1. 事务执行的性能开销

Redis事务虽然高效，但仍有一定的性能开销需要考虑。

命令队列的内存开销分析：

每个入队命令都会占用一定内存，具体包括：

命令本身的名称占用空间
命令参数占用的空间
Redis内部表示命令的数据结构开销

对于大量命令或参数较大的命令，内存开销会显著增加。

事务执行时间的构成因素：

Redis事务执行时间主要由以下几部分组成：

命令入队时间：将命令加入队列的时间
EXEC执行开销：事务提交的处理时间
各命令实际执行时间：队列中所有命令的执行时间总和
结果收集与返回时间：整理并返回所有命令执行结果的时间

网络延迟对事务的影响：

在分布式环境中，网络延迟会显著影响事务执行：

多次网络往返（MULTI、命令入队、EXEC）会累积延迟
大量命令结果返回可能导致网络拥塞

💡 性能小贴士：在高延迟网络环境下，减少客户端与服务器的交互次数至关重要。这也是为什么Redis管道（Pipeline）技术与事务经常结合使用的原因。

2. 事务使用的性能优化策略

批量操作与管道的结合使用：

Redis管道（Pipeline）技术可以显著提升事务性能：

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

# 测试函数：不使用管道的事务
def transaction_without_pipeline():
    start = time.time()
    r.execute_command('MULTI')
    for i in range(1000):
        r.execute_command('SADD', f'test:set', f'member:{i}')
    r.execute_command('EXEC')
    end = time.time()
    return end - start

# 测试函数：使用管道的事务
def transaction_with_pipeline():
    start = time.time()
    pipe = r.pipeline(transaction=True)
    for i in range(1000):
        pipe.sadd(f'test:set', f'member:{i}')
    pipe.execute()
    end = time.time()
    return end - start

# 执行测试
print(f"不使用管道的事务耗时: {transaction_without_pipeline():.6f}秒")
print(f"使用管道的事务耗时: {transaction_with_pipeline():.6f}秒")

合理控制事务中的命令数量：

过多的命令会导致事务执行时间延长，建议：

单个事务中命令数量控制在100个以内
对于大批量操作，考虑分批执行多个小事务
如果需要执行大量命令，考虑使用Lua脚本代替事务

避免长事务阻塞其他操作：

Redis是单线程模型，长事务会阻塞其他操作：

避免在事务中执行复杂度为O(N)的命令，如SORT、LREM等
不要在MULTI和EXEC之间执行耗时的客户端运算
考虑使用多个Redis实例分担负载

3. 监控与调优实践

事务执行的关键指标监控：

Redis的INFO命令可以提供一些事务相关的监控指标：

# 查看当前正在执行的事务数
redis-cli> INFO clients
# 命令执行统计
redis-cli> INFO commandstats

对于更深入的事务监控，可以考虑：

使用Redis的慢日志功能监控慢事务
对事务的执行时间进行采样和分析
使用Redis的MONITOR命令临时观察事务执行情况（生产环境慎用）

事务失败率与回滚分析：

监控WATCH命令导致的事务失败率：

计算EXEC执行失败的比例
分析哪些键经常导致冲突
考虑调整业务逻辑或数据结构减少冲突

性能瓶颈识别与解决方案：

常见性能瓶颈及解决方案：

瓶颈类型	症状	解决方案
命令过多	单个事务执行时间长	拆分为多个小事务
频繁冲突	WATCH导致事务频繁失败	减少监视的键，优化数据结构
网络延迟	事务响应慢但服务器负载低	使用管道技术，合并命令
大键问题	个别事务执行特别慢	拆分大键，使用HSCAN等渐进式命令

六、常见的事务相关踩坑经验与最佳实践

在实际项目中使用Redis事务，往往会遇到各种各样的问题。以下是我在实践中总结的一些常见踩坑及其解决方案。

1. 事务使用的常见误区

误区一：误以为Redis事务支持完整的回滚

我曾经在处理一个支付系统时，错误地认为Redis事务会像MySQL一样，在任何命令失败后自动回滚已执行的操作。实际上，Redis事务只有在EXEC执行前发现错误时才会取消整个事务，而运行期错误并不会触发回滚。

解决方案：

预先检查命令的可行性（如类型检查）
关键业务逻辑增加额外的一致性检查
实现自己的补偿机制处理部分失败情况

误区二：过度依赖事务导致性能下降

在一个高并发系统中，我们原本使用Redis事务来确保计数器的准确性，但发现系统性能急剧下降。分析后发现，过多的事务和WATCH命令导致大量的冲突和重试，显著增加了系统负载。

解决方案：

对非关键操作，考虑使用非事务性操作
使用原子命令代替事务（如INCR而不是GET+SET）
在极高并发场景，考虑使用Lua脚本替代事务

误区三：WATCH使用不当导致事务频繁失败

在一个排行榜系统中，我们监视了整个排行榜键，导致任何用户分数更新都会触发其他事务失败：

WATCH leaderboard  # 监视整个排行榜，这是错误的做法
MULTI
ZADD leaderboard user1 100
EXEC  # 如果同时有其他客户端修改了排行榜，事务会失败

解决方案：

只监视必要的特定键，而不是大范围集合
将大型集合拆分为多个小集合，减少冲突概率
使用更细粒度的锁定机制，如分用户的独立计数器

2. 分布式环境下的事务处理

主从复制与事务的一致性问题：

Redis主从复制是异步的，这会导致事务在主从节点间的可见性存在延迟：

时间轴:
t1: 客户端在主节点执行事务，更新数据
t2: 主节点返回成功
t3: 客户端读取从节点数据（此时从节点可能尚未同步最新事务）

解决方案：

对一致性要求高的读操作，直接读主节点
实现读-写-读模式，确保从主节点读取最新数据
使用WAIT命令确保复制到足够数量的从节点

集群环境下事务的局限性：

Redis集群不支持跨槽位的事务，这是很多开发者容易忽视的点：

MULTI
SET key1 value1  # 假设key1在槽1
SET key2 value2  # 假设key2在槽2
EXEC  # 这在集群模式下会失败

解决方案：

使用相同的哈希标签确保相关键在同一个槽
拆分事务，每个事务只操作同槽位的键
对于跨槽事务需求，考虑使用客户端协调或Lua脚本

Redis集群模式的替代方案：

当集群模式限制了事务使用时，可以考虑：

使用Redis哨兵模式而非集群模式
实现客户端层面的分布式事务协调
将强一致性要求的部分迁移到关系型数据库

3. 生产环境最佳实践总结

事务与Lua脚本的选择策略：

根据我的经验，选择标准如下：

特性	Redis事务	Lua脚本
易用性	✅ 较简单	⚠️ 需要学习Lua
原子性	✅ 支持	✅ 支持
回滚支持	❌ 不支持	❌ 不支持
条件判断	❌ 有限支持	✅ 完全支持
网络开销	⚠️ 多次往返	✅ 单次往返
执行效率	⚠️ 中等	✅ 高
调试难度	✅ 简单	⚠️ 较复杂

对于简单操作，优先使用事务；对于复杂逻辑或高性能需求，选择Lua脚本。

事务使用的代码规范与模式：

生产环境的事务代码应当遵循以下模式：

def safe_transaction(max_retries=3, retry_delay=0.1):
    """安全的事务执行模式"""
    for attempt in range(max_retries):
        try:
            # 1. 监视关键键
            redis.watch(key1, key2)
            
            # 2. 读取当前值
            current_values = redis.mget(key1, key2)
            
            # 3. 业务逻辑判断
            if not validate_values(current_values):
                redis.unwatch()
                return {"success": False, "reason": "业务校验失败"}
            
            # 4. 开始事务
            pipe = redis.pipeline(transaction=True)
            pipe.multi()
            
            # 5. 添加命令
            pipe.set(key1, new_value1)
            pipe.set(key2, new_value2)
            
            # 6. 执行事务
            results = pipe.execute()
            
            # 7. 成功处理
            log_success(results)
            return {"success": True, "results": results}
            
        except redis.WatchError:
            # 8. 冲突处理
            log_conflict(attempt)
            time.sleep(retry_delay)
            continue
        except Exception as e:
            # 9. 异常处理
            log_error(e)
            return {"success": False, "error": str(e)}
    
    # 10. 重试耗尽处理
    log_max_retries_reached()
    return {"success": False, "reason": "达到最大重试次数"}

错误处理与重试机制的设计：

健壮的错误处理应包括：

指数退避重试策略：每次重试增加等待时间
错误分类与针对性处理：区分暂时性错误和永久性错误
熔断机制：检测到系统异常时暂停重试
日志与监控：记录所有事务失败，设置告警阈值

七、Redis事务与其他方案的对比

在选择合适的并发控制机制时，我们需要了解各种方案的优缺点。

1. 事务 vs Lua脚本

功能与性能对比：

特性	Redis事务	Lua脚本
执行模式	命令队列批量执行	脚本整体执行
条件判断	仅WATCH提供有限判断	完整的逻辑控制
复杂计算	不支持	支持复杂计算
网络开销	多次网络往返	单次网络往返
原子性	命令级原子	脚本级原子
执行开销	较低	脚本解析有开销

性能对比示例：

假设需要原子性地为100个用户增加积分：

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

# 使用事务实现
def using_transaction():
    start = time.time()
    pipe = r.pipeline(transaction=True)
    for i in range(100):
        pipe.hincrby("user:points", f"user:{i}", 10)
    pipe.execute()
    return time.time() - start

# 使用Lua脚本实现
def using_lua():
    start = time.time()
    lua_script = """
    for i=0,99 do
        redis.call('HINCRBY', KEYS[1], 'user:'..i, ARGV[1])
    end
    return 'OK'
    """
    r.eval(lua_script, 1, "user:points", 10)
    return time.time() - start

# 清理数据
r.delete("user:points")

# 执行测试
print(f"事务执行时间: {using_transaction():.6f}秒")
r.delete("user:points")
print(f"Lua脚本执行时间: {using_lua():.6f}秒")

适用场景分析：

Redis事务适用于：

简单的原子性操作
对Lua不熟悉的开发团队
需要WATCH功能的场景

Lua脚本适用于：

需要条件判断的复杂逻辑
高性能要求场景
需要减少网络往返的操作

2. 事务 vs 分布式锁

实现机制对比：

特性	Redis事务	分布式锁
锁定模型	乐观锁（CAS）	悲观锁
阻塞行为	非阻塞，失败即返回	可阻塞等待
死锁风险	几乎不存在	存在风险，需超时释放
性能影响	冲突时性能下降	高并发时性能受限
实现复杂度	相对简单	较复杂（获取、续期、释放）

分布式锁简单实现：

import redis
import time
import uuid

class RedisLock:
    def __init__(self, redis_client, lock_name, expire_time=10):
        self.redis = redis_client
        self.lock_name = f"lock:{lock_name}"
        self.expire_time = expire_time
        self.identifier = str(uuid.uuid4())
        self.acquired = False
    
    def acquire(self, timeout=10):
        end_time = time.time() + timeout
        while time.time() < end_time:
            # 尝试获取锁
            if self.redis.set(self.lock_name, self.identifier, nx=True, ex=self.expire_time):
                self.acquired = True
                return True
            # 短暂睡眠后重试
            time.sleep(0.1)
        return False
    
    def release(self):
        if not self.acquired:
            return False
        
        # 使用Lua脚本确保只释放自己的锁
        release_script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        result = self.redis.eval(release_script, 1, self.lock_name, self.identifier)
        if result:
            self.acquired = False
            return True
        return False
    
    def __enter__(self):
        self.acquire()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

使用场景差异：

Redis事务适用于：

短时间的原子性操作
读取-计算-更新模式
乐观并发控制适合的场景（冲突较少）

分布式锁适用于：

需要独占资源的场景
复杂的多步骤操作
操作时间较长的场景
悲观并发控制适合的场景（冲突频繁）

3. 多种方案的组合使用策略

实际项目中，我们往往需要组合多种技术来解决复杂问题。

事务与Lua脚本结合使用：

Lua脚本可以作为事务的一部分执行，或者在WATCH后根据条件执行：

def combined_approach():
    # WATCH键
    redis.watch("inventory")
    
    # 获取当前库存
    current = int(redis.get("inventory") or 0)
    
    if current <= 0:
        redis.unwatch()
        return {"success": False, "reason": "库存不足"}
    
    # 复杂逻辑使用Lua脚本
    update_script = """
    local current = tonumber(redis.call('GET', KEYS[1]))
    if current <= 0 then return 0 end
    
    -- 执行复杂的库存计算
    local new_value = current - tonumber(ARGV[1])
    redis.call('SET', KEYS[1], new_value)
    
    -- 记录操作日志
    redis.call('LPUSH', KEYS[2], 'Deducted '..ARGV[1]..' at '..ARGV[2])
    
    return new_value
    """
    
    # 将Lua脚本作为事务的一部分执行
    pipe = redis.pipeline(transaction=True)
    pipe.multi()
    pipe.eval(update_script, 2, "inventory", "inventory:log", 1, int(time.time()))
    result = pipe.execute()
    
    if result[0] > 0:
        return {"success": True, "new_inventory": result[0]}
    else:
        return {"success": False, "reason": "操作失败"}

事务与分布式锁互补：

有些场景下，我们可以结合使用分布式锁和事务：

def lock_then_transaction(user_id, amount):
    """先获取锁，再使用事务处理账户余额"""
    lock_key = f"lock:user:{user_id}"
    
    # 获取分布式锁
    lock_value = str(uuid.uuid4())
    acquired = redis.set(lock_key, lock_value, nx=True, ex=10)
    
    if not acquired:
        return {"success": False, "reason": "系统繁忙，请稍后再试"}
    
    try:
        # 使用事务处理账户操作
        pipe = redis.pipeline(transaction=True)
        pipe.watch(f"balance:{user_id}")
        
        current_balance = int(redis.get(f"balance:{user_id}") or 0)
        if current_balance < amount:
            pipe.unwatch()
            return {"success": False, "reason": "余额不足"}
        
        pipe.multi()
        pipe.decrby(f"balance:{user_id}", amount)
        pipe.lpush(f"transactions:{user_id}", f"{amount}:{time.time()}")
        results = pipe.execute()
        
        return {"success": True, "new_balance": results[0]}
    finally:
        # 释放锁
        release_script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        redis.eval(release_script, 1, lock_key, lock_value)