一、引言在分布式系统架构中分布式锁是解决资源竞争、保证数据一致性的核心组件。Redisson 作为 Java 生态中最成熟的 Redis 客户端之一提供了丰富多样的分布式锁实现从基础的可重入锁RLock、公平锁FairLock到读写锁ReadWriteLock再到红锁RedLock几乎覆盖了所有常见的分布式锁场景。然而在实际业务中我们经常会遇到这样一种需求一个业务操作需要同时持有多个独立的分布式锁才能执行。比如在电商下单场景中需要同时锁定「用户账户」和「商品库存」两个资源在转账场景中需要同时锁定「转出账户」和「转入账户」。如果使用普通的 RLock 逐一加锁不仅代码复杂还容易引发死锁问题。Redisson 提供的MultiLock联锁/组合锁正是为了解决这一痛点而生。MultiLock 可以将多个独立的 RLock 对象组合成一个逻辑上的「联锁」对外提供统一的加锁/解锁接口并保证所有子锁全部加锁成功才算整体加锁成功任意一个子锁加锁失败则整体失败并自动释放已持有的锁。本文作为《架构设计之Redisson分布式锁》系列第六篇将从核心原理、源码深度解析、使用方式、实战案例、性能优化等多个维度全面剖析 Redisson MultiLock 的设计思想和实现细节全文约 2 万字适合对分布式锁有一定基础、希望深入理解 MultiLock 机制的开发者阅读。二、MultiLock 概述2.1 什么是 MultiLockMultiLock 是 Redisson 提供的一种组合锁实现它允许开发者将多个 RLock 对象组合成一个逻辑上的联锁。在 Redisson 的类层次结构中MultiLock 实现了RLock接口这意味着它可以像普通分布式锁一样被使用但其内部管理着多个独立的锁实例。MultiLock 的核心类为org.redisson.RedissonMultiLock它继承自RedissonBaseLock实现了标准的 RLock 接口。其构造方法接收一个 RLock 列表将这些锁聚合成一个整体。MultiLock 的核心行为可以用一句话概括全部成功才算成功部分失败则全部回滚。具体来说加锁时对所有子锁依次尝试加锁只有当所有子锁都成功加锁后MultiLock 才算加锁成功。如果中途某个子锁加锁失败MultiLock 会自动释放已经成功加锁的子锁并向调用方返回加锁失败。解锁时对所有子锁依次执行解锁操作确保所有子锁都被释放。续期时Watchdog 机制会为每个子锁独立续期保证锁在业务执行期间不会过期。2.2 MultiLock 与 RedLock 的关系很多开发者容易将 MultiLock 和 RedLock红锁混淆这里做一个明确的区分特性RedissonMultiLockRedissonRedLock继承关系RedissonMultiLock 是父类RedissonRedLock 继承自 RedissonMultiLock锁来源任意多个 RLock可来自同一或不同 Redis 实例通常来自多个独立的 Redis 节点主从/集群核心场景业务层面的多资源锁组合解决 Redis 主从切换导致锁丢失的容错问题加锁逻辑全部成功才算成功多数节点N/21成功才算成功加锁机制顺序加锁可配置超时并行加锁多数表决简单来说RedLock 是 MultiLock 的一种特殊应用它继承了 MultiLock 的组合锁能力但引入了「多数派」的加锁成功判定逻辑。而本文讨论的 MultiLock 是更通用的组合锁要求所有子锁全部加锁成功。2.3 典型应用场景场景一电商下单——多资源锁定用户下单时需要同时锁定「用户账户余额」「商品库存」「优惠券」三个资源。如果逐一加锁可能出现「锁了账户但库存被其他人抢光」的中间状态。使用 MultiLock 可以保证三个资源同时被锁定或都不被锁定。场景二转账操作——对称资源锁定A 向 B 转账时需要同时锁定 A 和 B 的账户。如果先锁 A 再锁 B可能因为加锁顺序不一致导致死锁另一个线程先锁 B 再锁 A。MultiLock 可以统一管理锁的获取顺序配合锁排序策略有效避免死锁。场景三跨系统资源协调一个业务流程需要同时锁定 Redis 中的缓存数据、数据库中的行记录、以及消息队列中的消费位点。MultiLock 可以将这些不同来源的锁统一管理实现跨系统的原子性操作。三、Redisson 分布式锁基础回顾在深入 MultiLock 源码之前我们先快速回顾 Redisson 分布式锁的核心机制为后续理解 MultiLock 的实现打下基础。3.1 RLock 接口体系Redisson 的分布式锁体系围绕RLock接口展开该接口继承自java.util.concurrent.locks.Lock并扩展了异步、响应式等接口public interface RLock extends Lock, RLockAsync, RExpirable { // 获取锁的名称 String getName(); // 尝试加锁支持等待时间和持有时间 boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException; // 强制解锁 void forceUnlock(); // 判断是否被当前线程持有 boolean isHeldByCurrentThread(); // 判断是否被任意线程持有 boolean isLocked(); // 获取当前线程的重入次数 int getHoldCount(); }Redisson 中主要的 RLock 实现类包括RedissonLock基础可重入锁基于 Redis 的 Hash 数据结构 Lua 脚本实现。RedissonFairLock公平锁基于 Redis 的队列和 Hash 实现保证先请求的线程先获取锁。RedissonSpinLock自旋锁基于 Redis 的发布订阅 自旋重试机制。RedissonMultiLock联锁/组合锁本文重点讨论。RedissonRedLock红锁继承自 MultiLock实现多数派加锁逻辑。3.2 基础锁的加锁流程以RedissonLock为例其加锁的核心流程如下执行 Lua 脚本向 Redis 发送一段 Lua 脚本脚本逻辑为如果 key 不存在则使用 Hash 结构设置锁信息field 为线程标识value 为重入次数并设置过期时间如果 key 存在且 field 匹配重入场景则增加重入计数并刷新过期时间否则返回剩余存活时间表示锁被其他线程持有。加锁失败处理如果 Lua 脚本返回非空表示锁被占用则订阅该锁对应的 Redis Channel通过 Pub/Sub 机制等待锁释放通知。Watchdog 续期如果加锁时未指定 leaseTime即使用默认的 -1Redisson 会启动 Watchdog 定时任务每隔lockWatchdogTimeout/3默认 10 秒对锁进行续期将过期时间重置为lockWatchdogTimeout默认 30 秒。加锁的核心 Lua 脚本简化如下-- KEYS[1]: 锁的 key -- ARGV[1]: 锁的过期时间毫秒 -- ARGV[2]: 线程标识UUID:线程ID if (redis.call(exists, KEYS[1]) 0) then -- 锁不存在直接加锁 redis.call(hincrby, KEYS[1], ARGV[2], 1); redis.call(pexpire, KEYS[1], ARGV[1]); return nil; end; if (redis.call(hexists, KEYS[1], ARGV[2]) 1) then -- 锁存在且属于当前线程重入 redis.call(hincrby, KEYS[1], ARGV[2], 1); redis.call(pexpire, KEYS[1], ARGV[1]); return nil; end; -- 锁被其他线程持有返回剩余存活时间 return redis.call(pttl, KEYS[1]);3.3 解锁流程解锁同样通过 Lua 脚本保证原子性-- 检查锁是否存在且属于当前线程 if (redis.call(hexists, KEYS[1], ARGV[1]) 0) then return nil; -- 锁不属于当前线程 end; -- 递减重入计数 local counter redis.call(hincrby, KEYS[1], ARGV[1], -1); if (counter 0) then -- 重入计数大于0只续期不删除 redis.call(pexpire, KEYS[1], ARGV[2]); return 0; else -- 重入计数为0删除锁并发布解锁通知 redis.call(del, KEYS[1]); redis.call(publish, KEYS[2], ARGV[3]); return 1; end;理解这些基础机制后我们再来看 MultiLock 如何在多锁场景下协调这些操作。四、MultiLock 核心原理4.1 设计思想MultiLock 的设计思想源自分布式系统中的原子性多资源锁定需求。在单机环境下我们可以使用synchronized或ReentrantLock来保护临界区但在分布式环境下当多个资源分布在不同的 Redis 节点或不同的业务域时就需要一种机制来保证多资源锁定的原子性。MultiLock 的核心设计思想可以归纳为三点全量成功语义所有子锁必须全部加锁成功整个 MultiLock 才算加锁成功。这保证了多资源操作的原子性——要么全部锁定要么全部不锁定。失败自动回滚在加锁过程中如果某个子锁加锁失败MultiLock 会自动释放已经成功加锁的子锁避免部分锁被持有而阻塞其他线程。统一生命周期管理MultiLock 统一管理所有子锁的加锁、解锁、续期操作对外暴露与普通 RLock 完全一致的接口降低使用复杂度。4.2 加锁流程详解MultiLock 的加锁过程可以分解为以下几个步骤步骤一参数校验与准备在tryLock方法被调用时MultiLock 首先对参数进行校验计算每个子锁的等待超时时间。如果调用者指定了waitTimeMultiLock 会将其均分给每个子锁即每个子锁的等待超时 waitTime / locks.size()确保总等待时间不超过调用者预期。步骤二顺序加锁MultiLock 按照子锁列表的顺序依次对每个子锁调用tryLock方法。这是一个顺序执行的过程而非并行。这样设计的原因是为了避免「部分成功、部分失败」的复杂状态管理同时顺序加锁也便于实现失败回滚。步骤三失败回滚如果在加锁过程中某个子锁加锁失败返回 false 或抛出异常MultiLock 会立即停止后续子锁的加锁尝试并逆序释放已经成功加锁的子锁。逆序释放是为了保证解锁顺序与加锁顺序相反减少死锁风险。步骤四返回结果如果所有子锁都加锁成功MultiLock 返回 true表示联锁加锁成功如果任何子锁加锁失败则返回 false且所有已持有的子锁已被释放。下面用流程图直观展示加锁逻辑flowchart TD A[开始 tryLock] -- B[计算每个子锁的等待超时时间] B -- C[遍历子锁列表] C -- D{尝试对当前子锁加锁} D --|成功| E{是否还有剩余子锁?} D --|失败| F[逆序释放已加锁的子锁] F -- G[返回 false] E --|是| C E --|否| H[所有子锁加锁成功] H -- I[返回 true]4.3 解锁流程详解MultiLock 的解锁流程相对简单遍历所有子锁依次调用其unlock方法。即使某个子锁解锁失败也会继续尝试解锁其他子锁确保所有子锁都被释放。解锁时有一个重要的细节MultiLock 会忽略子锁解锁时的异常。这是因为解锁操作本身应该是幂等的即使某个子锁已经被释放例如已过期也不应该影响其他子锁的释放。4.4 Watchdog 续期机制当 MultiLock 未指定leaseTime即使用默认的 -1时Watchdog 机制会为每个子锁独立续期。MultiLock 内部维护了一个lockWatchdogTimeout配置默认为 30 秒。Watchdog 定时任务每隔 10 秒lockWatchdogTimeout / 3执行一次为所有子锁进行续期。关键的实现细节MultiLock 的 Watchdog 续期是独立为每个子锁执行的。这意味着每个子锁有独立的过期时间管理。如果某个子锁所在 Redis 节点出现网络问题导致续期失败Watchdog 会记录异常但不会影响其他子锁的续期。当 MultiLock 被显式解锁或业务线程结束时Watchdog 会停止续期并释放所有子锁。五、源码深度解析本章节将对 RedissonMultiLock 的核心源码进行逐行分析帮助读者深入理解其实现原理。本文基于 Redisson 3.23.x 版本源码进行分析。5.1 类结构与构造方法public class RedissonMultiLock extends RedissonBaseLock { // 子锁列表使用 List 保证顺序 final Listlt;RLockgt; locks new ArrayListlt;gt;(); /** 构造方法接收多个 RLock 对象 子锁可以来自同一个 RedissonClient也可以来自不同的 RedissonClient 甚至可以来自不同的 Redis 集群 */ public RedissonMultiLock(RLock... locks) { if (locks.length 0) { throw new IllegalArgumentException(Lock objects are not defined); } this.locks.addAll(Arrays.asList(locks)); } // 获取子锁数量 public int size() { return locks.size(); } // 获取所有子锁 protected Listlt;RLockgt; getLocks() { return locks; } }从构造方法可以看出MultiLock 的设计非常灵活子锁可以是任意 RLock 实现类RedissonLock、RedissonFairLock 等。子锁可以来自不同的 RedissonClient 实例这意味着可以跨 Redis 集群进行组合锁定。子锁通过可变参数传入最少需要一个锁但实际使用中通常至少需要两个才有意义。5.2 tryLock 核心实现tryLock是 MultiLock 最核心的方法我们逐段分析其实现Override public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { // 将等待时间转换为毫秒 long newLeaseTime -1; if (leaseTime 0) { // 如果指定了持有时间则每个子锁的持有时间相同 newLeaseTime unit.toMillis(waitTime) * 2; } // 计算当前时间用于超时判断 long time System.currentTimeMillis(); // remainTime 记录剩余等待时间 long remainTime -1; if (waitTime ! -1) { remainTime unit.toMillis(waitTime); } // 计算每个子锁的等待时间总等待时间 / 子锁数量 long lockWaitTime calcLockWaitTime(remainTime); // ... 加锁逻辑 }这里有一个关键的计算calcLockWaitTime方法将总等待时间均分给每个子锁。这样设计的好处是无论有多少个子锁总的等待时间上限是可控的不会因为子锁数量增加而导致无限等待。protected long calcLockWaitTime(long remainTime) { if (remainTime -1) { return -1; // 无限等待模式 } // 均分等待时间但每个子锁至少保留 1 毫秒 return Math.max(remainTime / locks.size(), 1); }接下来是加锁的核心循环// 记录已经成功加锁的子锁数量 int failedLocksLimit failedLocksLimit(); // 已加锁的子锁列表用于失败回滚 ListRLock acquiredLocks new ArrayList(locks.size()); // 遍历所有子锁依次加锁 for (ListIteratorRLock iterator locks.listIterator(); iterator.hasNext();) { RLock lock iterator.next(); boolean lockAcquired; try { // 如果没有指定等待时间使用 tryLock() 非阻塞尝试 if (waitTime -1 leaseTime -1) { lockAcquired lock.tryLock(); } else { // 计算当前子锁的等待超时时间 long awaitTime Math.min(lockWaitTime, remainTime); lockAcquired lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS); } } catch (InterruptedException e) { // 被中断时释放已加锁的子锁并抛出异常 unlockInner(acquiredLocks); throw e; } if (lockAcquired) { // 加锁成功加入已获取列表 acquiredLocks.add(lock); } else { // 加锁失败检查是否已达到失败上限 if (locks.size() - acquiredLocks.size() failedLocksLimit()) { break; // 失败数达到上限停止加锁 } // 失败回滚释放所有已加锁的子锁 unlockInner(acquiredLocks); // 检查是否超时 if (remainTime ! -1) { long currentTime System.currentTimeMillis(); remainTime - (currentTime - time); time currentTime; if (remainTime amp;lt; 0) { // 等待超时加锁失败 return false; } } // 重置迭代器从头开始重新加锁 // 注意这是为了处理「等待-重试」的场景 while (iterator.hasPrevious()) { iterator.previous(); } acquiredLocks.clear(); } } // 如果所有子锁都加锁成功 if (acquiredLocks.size() locks.size()) { return true; } return false;从源码中我们可以看到几个关键设计1. failedLocksLimit 方法在 RedissonMultiLock 中failedLocksLimit()方法返回 0意味着「不允许任何子锁加锁失败」。这是 MultiLock 与 RedLock 的关键区别RedLock 重写了此方法返回locks.size() - (locks.size()/2 1)即允许少数子锁加锁失败。// RedissonMultiLock 中的实现不允许失败 protected int failedLocksLimit() { return 0; } // RedissonRedLock 中的实现允许少数失败 protected int failedLocksLimit() { return locks.size() - (locks.size()/2 1); }2. 失败重试机制当某个子锁加锁失败时MultiLock 并不会立即放弃而是会先释放所有已加锁的子锁避免资源浪费。检查剩余等待时间是否充足。如果还有等待时间重置迭代器从头开始重新尝试加锁。这种「全量重试」的策略虽然简单但在高并发场景下可能导致活锁问题多个线程同时竞争多个锁每次都在不同的子锁上失败反复重试却无人成功。后续章节会讨论如何避免这一问题。3. 中断处理当线程在加锁过程中被中断时MultiLock 会立即释放已加锁的子锁并抛出 InterruptedException保证不会留下孤儿锁。5.3 unlockInner 回滚实现protected void unlockInner(CollectionRLock locks) { ListRFutureVoid futures new ArrayList(locks.size()); // 遍历所有已加锁的子锁异步执行解锁 for (RLock lock : locks) { futures.add(lock.unlockAsync()); } // 等待所有解锁操作完成 for (RFutureVoid unlockFuture : futures) { unlockFuture.awaitUninterruptibly(); } }这里使用了异步解锁机制unlockAsync并通过awaitUninterruptibly等待所有解锁操作完成。即使某个解锁操作失败例如 Redis 连接断开也不会影响其他子锁的解锁。5.4 unlock 方法实现Override public void unlock() { // 获取所有子锁包括子类可能添加的额外锁 ListRFutureVoid futures new ArrayList(locks.size()); for (RLock lock : locks) { futures.add(lock.unlockAsync()); } // 等待所有解锁操作完成 for (RFutureVoid future : futures) { future.syncUninterruptibly(); } }与unlockInner的区别在于unlock方法使用syncUninterruptibly等待解锁结果而unlockInner使用awaitUninterruptibly。两者的区别在于异常处理方式syncUninterruptibly会抛出执行异常而awaitUninterruptibly只是等待结果。5.5 续期机制源码分析MultiLock 的续期机制继承自RedissonBaseLock核心逻辑在scheduleExpirationRenewal方法中protected void scheduleExpirationRenewal(long threadId) { ExpirationEntry entry new ExpirationEntry(); ExpirationEntry oldEntry EXPIRATION_RENEWAL_MAP.putIfAbsent( getEntryName(), entry); if (oldEntry ! null) { // 已有续期任务增加引用计数重入场景 oldEntry.addThreadId(threadId); } else { // 新启动续期任务 entry.addThreadId(threadId); try { renewExpiration(); } finally { if (Thread.currentThread().isInterrupted()) { cancelExpirationRenewal(threadId); } } } }对于 MultiLock 而言续期操作会为每个子锁分别执行protected RFutureBoolean renewExpirationAsync(long threadId) { // 为每个子锁创建续期 Future ListRFutureBoolean futures new ArrayList(locks.size()); for (RLock lock : locks) { if (lock instanceof RedissonBaseLock) { futures.add(((RedissonBaseLock) lock) .renewExpirationAsync(threadId)); } } // 合并所有续期结果 return Futures.allOf(futures); }值得注意的是续期采用「尽力而为」策略即使某个子锁续期失败也不会影响其他子锁的续期更不会导致已加锁的子锁被释放。这种设计避免了因单个 Redis 节点短暂不可用而导致整个 MultiLock 失效。六、MultiLock 的使用方式6.1 基础使用最简单的 MultiLock 使用方式如下// 获取 Redisson 客户端 RedissonClient redisson Redisson.create(config); // 创建多个独立的锁 RLock lock1 redisson.getLock(lock:user:1001); RLock lock2 redisson.getLock(lock:product:2001); RLock lock3 redisson.getLock(lock:coupon:3001); // 创建 MultiLock 联锁 RLock multiLock redisson.getMultiLock(lock1, lock2, lock3); // 使用 try-finally 保证解锁 try { // 尝试加锁最多等待 10 秒持有时间 30 秒 boolean locked multiLock.tryLock(10, 30, TimeUnit.SECONDS); if (locked) { // 执行业务逻辑 doBusinessLogic(); } else { // 加锁失败处理 handleLockFailure(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { // 释放联锁自动释放所有子锁 multiLock.unlock(); }6.2 使用 Watchdog 自动续期如果不指定 leaseTimeRedisson 的 Watchdog 会自动为每个子锁续期RLock multiLock redisson.getMultiLock(lock1, lock2, lock3); // 不指定 leaseTime使用 Watchdog 自动续期 multiLock.lock(); // 或 lock(30, TimeUnit.SECONDS) 跳过 Watchdog try { // 执行耗时较长的业务逻辑 // Watchdog 会每隔 10 秒自动续期每次续期 30 秒 longRunningBusinessLogic(); } finally { multiLock.unlock(); }建议如果业务逻辑的执行时间可预期最好显式指定 leaseTime避免 Watchdog 在业务异常时无法及时释放锁。Watchdog 更适合执行时间不确定的长任务场景。6.3 跨 Redis 实例的 MultiLockMultiLock 支持跨不同的 Redis 实例甚至不同的 Redis 集群组合锁// 第一个 Redis 集群用户服务 Config config1 new Config(); config1.useClusterServers() .addNodeAddress(redis://user-cluster-1:6379, redis://user-cluster-2:6379); RedissonClient redissonUser Redisson.create(config1); // 第二个 Redis 集群订单服务 Config config2 new Config(); config2.useClusterServers() .addNodeAddress(redis://order-cluster-1:6379, redis://order-cluster-2:6379); RedissonClient redissonOrder Redisson.create(config2); // 创建跨集群的锁 RLock userLock redissonUser.getLock(lock:user:1001); RLock orderLock redissonOrder.getLock(lock:order:20240001); // 创建跨集群的 MultiLock RLock multiLock redissonUser.getMultiLock(userLock, orderLock); try { multiLock.lock(); // 跨集群的原子操作 crossClusterOperation(); } finally { multiLock.unlock(); }这种跨实例的 MultiLock 能力在微服务架构中非常实用可以实现跨服务的资源协调锁定。6.4 异步与响应式使用MultiLock 同样支持异步和响应式编程模型// 异步方式 RLock multiLock redisson.getMultiLock(lock1, lock2, lock3); RFutureBoolean future multiLock.tryLockAsync(10, 30, TimeUnit.SECONDS); future.whenComplete((locked, exception) - { if (exception ! null) { // 异常处理 return; } if (locked) { try { doBusinessLogic(); } finally { multiLock.unlockAsync(); } } }); // 响应式方式Reactive RLockReactive multiLockReactive redissonReactive .getMultiLock(lock1, lock2, lock3); multiLockReactive.lock() .then(Mono.fromRunnable(() - doBusinessLogic())) .doFinally(signalType - multiLockReactive.unlock()) .subscribe();七、实战案例7.1 电商下单场景多资源锁定在电商秒杀场景中下单操作需要同时锁定用户账户、商品库存和优惠券保证操作的原子性。以下是使用 MultiLock 的完整实现Service public class OrderService { Autowired private RedissonClient redisson; Autowired private AccountService accountService; Autowired private InventoryService inventoryService; Autowired private CouponService couponService; /** 使用 MultiLock 保证多资源锁定的原子性 */ public OrderResult placeOrder(OrderRequest request) { String userId request.getUserId(); String productId request.getProductId(); String couponId request.getCouponId(); // 创建三个独立的锁 RLock accountLock redisson.getLock(lock:account: userId); RLock inventoryLock redisson.getLock(lock:inventory: productId); RLock couponLock redisson.getLock(lock:coupon: couponId); // 组合为 MultiLock RLock multiLock redisson.getMultiLock( accountLock, inventoryLock, couponLock); try { // 尝试加锁最多等待 5 秒 boolean locked multiLock.tryLock(5, 30, TimeUnit.SECONDS); if (!locked) { return OrderResult.fail(系统繁忙请稍后重试); } // 三个资源都已锁定执行业务逻辑 // 1. 检查账户余额 Account account accountService.getAccount(userId); if (account.getBalance().compareTo(request.getAmount()) amp;lt; 0) { return OrderResult.fail(账户余额不足); } // 2. 检查库存 Inventory inventory inventoryService.getInventory(productId); if (inventory.getStock() amp;lt; request.getQuantity()) { return OrderResult.fail(商品库存不足); } // 3. 检查优惠券 if (couponId ! null) { Coupon coupon couponService.getCoupon(couponId); if (!coupon.isValid()) { return OrderResult.fail(优惠券已失效); } } // 4. 执行扣减操作 accountService.deduct(userId, request.getAmount()); inventoryService.deduct(productId, request.getQuantity()); if (couponId ! null) { couponService.useCoupon(couponId); } // 5. 创建订单 Order order createOrder(request); return OrderResult.success(order); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return OrderResult.fail(操作被中断); } finally { // 释放所有锁 multiLock.unlock(); } } }7.2 转账场景对称资源锁定与死锁避免在转账场景中死锁是最常见的问题。假设线程 T1 要锁 A 再锁 B线程 T2 要锁 B 再锁 A如果同时执行就可能死锁。使用 MultiLock 配合锁排序可以有效避免Service public class TransferService { Autowired private RedissonClient redisson; /** 转账操作A 向 B 转账 使用 MultiLock 锁排序避免死锁 */ public TransferResult transfer(String fromAccountId, String toAccountId, BigDecimal amount) { // 按照账户 ID 的字典序排序保证加锁顺序一致 String firstLockId; String secondLockId; if (fromAccountId.compareTo(toAccountId) lt; 0) { firstLockId fromAccountId; secondLockId toAccountId; } else { firstLockId toAccountId; secondLockId fromAccountId; } // 创建两个锁顺序已排序避免死锁 RLock firstLock redisson.getLock(lock:account: firstLockId); RLock secondLock redisson.getLock(lock:account: secondLockId); // 注意MultiLock 内部按传入顺序加锁这里已经排好序 RLock multiLock redisson.getMultiLock(firstLock, secondLock); try { boolean locked multiLock.tryLock(10, 30, TimeUnit.SECONDS); if (!locked) { return TransferResult.fail(转账超时请稍后重试); } // 检查转出账户余额 Account fromAccount getAccount(fromAccountId); if (fromAccount.getBalance().compareTo(amount) amp;lt; 0) { return TransferResult.fail(余额不足); } // 执行转账 deduct(fromAccountId, amount); add(toAccountId, amount); // 记录转账流水 recordTransfer(fromAccountId, toAccountId, amount); return TransferResult.success(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return TransferResult.fail(操作被中断); } finally { multiLock.unlock(); } } }关键点通过按账户 ID 字典序排序保证了所有线程总是以相同的顺序获取锁从根本上避免了死锁的发生。7.3 分布式任务调度批量资源锁定在分布式任务调度中有时需要一次性锁定多个任务槽位确保任务不会被重复执行Component public class BatchTaskExecutor { Autowired private RedissonClient redisson; private static final int MAX_BATCH_SIZE 10; /** 批量获取任务槽位并执行 */ public void executeBatchTasks(Listlt;Stringgt; taskIds) { if (taskIds.size() gt; MAX_BATCH_SIZE) { throw new IllegalArgumentException(批量任务数超过上限); } // 为每个任务创建锁 Listlt;RLockgt; taskLocks taskIds.stream() .map(id -gt; redisson.getLock(lock:task: id)) .collect(Collectors.toList()); // 组合为 MultiLock RLock multiLock redisson.getMultiLock( taskLocks.toArray(new RLock[0])); try { // 使用较短的等待时间避免长时间阻塞 boolean locked multiLock.tryLock(3, 60, TimeUnit.SECONDS); if (!locked) { log.warn(部分任务槽位被占用跳过本批次: {}, taskIds); return; } // 批量执行任务 for (String taskId : taskIds) { executeTask(taskId); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { multiLock.unlock(); } } private void executeTask(String taskId) { // 具体任务执行逻辑 log.info(执行任务: {}, taskId); } }7.4 缓存与数据库双写一致性保证在缓存与数据库双写场景中使用 MultiLock 可以保证缓存的更新和数据库的更新在同一锁保护下进行Service public class CacheConsistencyService { Autowired private RedissonClient redissonCache; // 缓存 Redis Autowired private RedissonClient redissonDb; // 分布式锁 Redis Autowired private UserRepository userRepository; /** 更新用户信息保证缓存与数据库的一致性 */ public void updateUser(User user) { String cacheKey user:cache: user.getId(); String dbLockKey user:db:lock: user.getId(); // 缓存锁和数据库锁 RLock cacheLock redissonCache.getLock(lock: cacheKey); RLock dbLock redissonDb.getLock(dbLockKey); // 使用 MultiLock 同时锁定 RLock multiLock redissonDb.getMultiLock(cacheLock, dbLock); try { multiLock.lock(30, TimeUnit.SECONDS); // 1. 先更新数据库 userRepository.save(user); // 2. 再删除缓存延迟双删策略 redissonCache.getBucket(cacheKey).delete(); // 3. 短暂延迟后再次删除缓存防止脏读 Thread.sleep(100); redissonCache.getBucket(cacheKey).delete(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { multiLock.unlock(); } } }八、与普通锁的对比分析8.1 性能对比维度普通 RLockMultiLock加锁延迟1 次 Redis 网络往返N 次 Redis 网络往返N 子锁数量解锁延迟1 次 Redis 网络往返N 次 Redis 网络往返续期开销1 次 Redis 命令/周期N 次 Redis 命令/周期Redis 连接占用1 个连接N 个连接如果子锁在不同实例可用性单节点可用即可用所有子锁节点可用才可用从性能角度看MultiLock 的开销与子锁数量成正比。建议在实际使用中子锁数量控制在 2-5 个过多的子锁会显著增加延迟。如果子锁都在同一 Redis 实例考虑使用 Lua 脚本批量操作替代 MultiLock。如果子锁分布在不同 Redis 实例MultiLock 是最佳选择。8.2 可靠性对比维度普通 RLockMultiLock死锁风险低单锁无死锁中需注意加锁顺序活锁风险无中并发竞争时可能发生部分失败处理N/A自动回滚已加锁的子锁网络分区容忍差单节点好可跨节点但需所有节点可用8.3 适用场景对比场景推荐方案原因单一资源互斥访问普通 RLock简单高效延迟低同一 Redis 实例的多资源Lua 脚本批量锁一次网络往返原子性更好跨 Redis 实例的多资源MultiLock唯一支持跨实例组合的方案多节点容错锁RedLockMultiLock 子类多数派机制容忍部分节点故障读写分离场景RReadWriteLock支持读共享、写互斥九、注意事项与最佳实践9.1 死锁预防虽然 MultiLock 内部会自动回滚失败加锁但在多线程并发场景下仍可能出现活锁问题。以下是一些预防措施1. 统一加锁顺序对所有需要加锁的资源按照固定的规则如资源 ID 字典序排序后再传入 MultiLock保证所有线程以相同顺序获取锁// 排序后再创建 MultiLock ListString resourceIds Arrays.asList(resourceB, resourceA, resourceC); Collections.sort(resourceIds); // 字典序排序 ListRLock locks resourceIds.stream() .map(id - redisson.getLock(lock: id)) .collect(Collectors.toList()); RLock multiLock redisson.getMultiLock( locks.toArray(new RLock[0]));2. 设置合理的等待超时避免使用lock()无超时等待始终使用tryLock(waitTime, leaseTime, unit)并设置合理的等待时间// 推荐设置等待超时 boolean locked multiLock.tryLock(5, 30, TimeUnit.SECONDS); // 不推荐无限等待 multiLock.lock();3. 限制子锁数量子锁数量越多加锁成功的概率越低重试次数越多。建议将子锁数量控制在 2-5 个。9.2 性能优化建议1. 同实例锁优先用 Lua 脚本如果所有子锁都在同一个 Redis 实例上可以考虑使用自定义 Lua 脚本一次性获取所有锁减少网络往返次数-- 批量加锁 Lua 脚本 local locks {} for i, key in ipairs(KEYS) do if redis.call(exists, key) 0 then redis.call(hset, key, ARGV[1], 1) redis.call(pexpire, key, ARGV[2]) table.insert(locks, key) else -- 有锁被占用释放已加锁的 for _, lockedKey in ipairs(locks) do redis.call(del, lockedKey) end return false end end return true2. 合理配置连接池跨实例的 MultiLock 会占用多个连接需要确保连接池足够大Config config new Config(); config.useSingleServer() .setAddress(redis://127.0.0.1:6379) .setConnectionPoolSize(64) // 默认 64按需调整 .setConnectionMinimumIdleSize(24); // 保证最小空闲连接3. Watchdog 超时调优根据业务特点调整 Watchdog 的超时时间Config config new Config(); config.setLockWatchdogTimeout(60000); // 60 秒默认 30 秒 // 注意续期间隔 lockWatchdogTimeout / 39.3 异常处理最佳实践public void executeWithMultiLock(ListString resourceIds) { // 1. 排序资源 ID避免死锁 Collections.sort(resourceIds); Listlt;RLockgt; locks resourceIds.stream() .map(id -gt; redisson.getLock(lock: id)) .collect(Collectors.toList()); RLock multiLock redisson.getMultiLock( locks.toArray(new RLock[0])); boolean locked false; try { // 2. 设置合理的超时时间 locked multiLock.tryLock(10, 30, TimeUnit.SECONDS); if (!locked) { // 3. 加锁失败记录日志并快速失败 log.warn(获取联锁失败资源: {}, resourceIds); throw new LockAcquisitionException(获取锁超时); } // 4. 执行业务逻辑 doBusinessLogic(); } catch (InterruptedException e) { // 5. 中断处理恢复中断状态 Thread.currentThread().interrupt(); throw new BusinessException(操作被中断, e); } catch (Exception e) { // 6. 业务异常处理 log.error(业务执行异常, e); throw e; } finally { // 7. 安全解锁防止 unlock 抛异常 if (locked) { try { multiLock.unlock(); } catch (Exception e) { log.error(解锁异常锁可能已过期, e); } } } }9.4 常见问题与解决方案问题一活锁Live Lock现象多个线程同时竞争多个锁每个线程都在不同子锁上失败反复重试但无人成功。解决方案引入随机退避Random Backoff在重试前等待一个随机时间。限制最大重试次数。使用优先级队列让高优先级线程优先获取锁。// 带随机退避的 MultiLock 重试封装 public class BackoffMultiLock { private final RLock multiLock; private final int maxRetries; private final long baseBackoffMs; public boolean tryLockWithBackoff(long waitTime, long leaseTime, TimeUnit unit) { long deadline System.currentTimeMillis() unit.toMillis(waitTime); int retries 0; while (System.currentTimeMillis() amp;lt; deadline amp;amp;amp;amp; retries amp;lt; maxRetries) { try { long remaining deadline - System.currentTimeMillis(); if (remaining amp;lt; 0) return false; boolean locked multiLock.tryLock(remaining, leaseTime, TimeUnit.MILLISECONDS); if (locked) return true; // 随机退避baseBackoffMs * (1 random) long backoff baseBackoffMs (long)(Math.random() * baseBackoffMs); Thread.sleep(Math.min(backoff, remaining)); retries; } catch (InterruptedException e) { Thread.currentThread().interrupt(); return false; } } return false; } }问题二子锁数量过多导致性能下降解决方案将多个资源合并为更粗粒度的锁如按用户 ID 分片而不是按每个资源 ID 加锁。使用分段锁Segmented Lock思想将资源分组。// 使用分段锁减少锁数量 public class SegmentedLockService { private static final int SEGMENT_COUNT 16; private final RLock[] segmentLocks new RLock[SEGMENT_COUNT]; public SegmentedLockService(RedissonClient redisson) { for (int i 0; i lt; SEGMENT_COUNT; i) { segmentLocks[i] redisson.getLock(lock:segment: i); } } /** 对一批资源 ID 加锁使用分段减少锁数量 */ public RLock getMultiLock(Listlt;Stringgt; resourceIds) { // 计算每个资源属于哪个分段 Setlt;Integergt; segments resourceIds.stream() .map(id -gt; Math.abs(id.hashCode()) % SEGMENT_COUNT) .collect(Collectors.toSet()); // 只对涉及的分段加锁大幅减少锁数量 RLock[] locks segments.stream() .sorted() // 排序避免死锁 .map(seg -gt; segmentLocks[seg]) .toArray(RLock[]::new); return redisson.getMultiLock(locks); } }问题三