架构设计之Redisson分布式锁-组合锁联锁MultiLock(六)

一、引言

在分布式系统架构中，分布式锁是解决资源竞争、保证数据一致性的核心组件。Redisson 作为 Java 生态中最成熟的 Redis 客户端之一，提供了丰富多样的分布式锁实现，从基础的可重入锁（RLock）、公平锁（FairLock）到读写锁（ReadWriteLock），再到红锁（RedLock），几乎覆盖了所有常见的分布式锁场景。

然而，在实际业务中，我们经常会遇到这样一种需求：一个业务操作需要同时持有多个独立的分布式锁才能执行。比如，在电商下单场景中，需要同时锁定「用户账户」和「商品库存」两个资源；在转账场景中，需要同时锁定「转出账户」和「转入账户」。如果使用普通的 RLock 逐一加锁，不仅代码复杂，还容易引发死锁问题。

Redisson 提供的MultiLock（联锁/组合锁）正是为了解决这一痛点而生。MultiLock 可以将多个独立的 RLock 对象组合成一个逻辑上的「联锁」，对外提供统一的加锁/解锁接口，并保证所有子锁全部加锁成功才算整体加锁成功，任意一个子锁加锁失败则整体失败并自动释放已持有的锁。

本文作为《架构设计之Redisson分布式锁》系列第六篇，将从核心原理、源码深度解析、使用方式、实战案例、性能优化等多个维度，全面剖析 Redisson MultiLock 的设计思想和实现细节，全文约 2 万字，适合对分布式锁有一定基础、希望深入理解 MultiLock 机制的开发者阅读。

二、MultiLock 概述

2.1 什么是 MultiLock

MultiLock 是 Redisson 提供的一种组合锁实现，它允许开发者将多个 RLock 对象组合成一个逻辑上的联锁。在 Redisson 的类层次结构中，MultiLock 实现了RLock接口，这意味着它可以像普通分布式锁一样被使用，但其内部管理着多个独立的锁实例。

MultiLock 的核心类为org.redisson.RedissonMultiLock，它继承自RedissonBaseLock，实现了标准的 RLock 接口。其构造方法接收一个 RLock 列表，将这些锁聚合成一个整体。

MultiLock 的核心行为可以用一句话概括：全部成功才算成功，部分失败则全部回滚。具体来说：

• 加锁时：对所有子锁依次尝试加锁，只有当所有子锁都成功加锁后，MultiLock 才算加锁成功。如果中途某个子锁加锁失败，MultiLock 会自动释放已经成功加锁的子锁，并向调用方返回加锁失败。
• 解锁时：对所有子锁依次执行解锁操作，确保所有子锁都被释放。
• 续期时：Watchdog 机制会为每个子锁独立续期，保证锁在业务执行期间不会过期。

2.2 MultiLock 与 RedLock 的关系

很多开发者容易将 MultiLock 和 RedLock（红锁）混淆，这里做一个明确的区分：

简单来说，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为例，其加锁的核心流程如下：

1. 执行 Lua 脚本：向 Redis 发送一段 Lua 脚本，脚本逻辑为：如果 key 不存在，则使用 Hash 结构设置锁信息（field 为线程标识，value 为重入次数），并设置过期时间；如果 key 存在且 field 匹配（重入场景），则增加重入计数并刷新过期时间；否则返回剩余存活时间表示锁被其他线程持有。
2. 加锁失败处理：如果 Lua 脚本返回非空（表示锁被占用），则订阅该锁对应的 Redis Channel，通过 Pub/Sub 机制等待锁释放通知。
3. Watchdog 续期：如果加锁时未指定 leaseTime（即使用默认的 -1），Redisson 会启动 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 的核心设计思想可以归纳为三点：

1. 全量成功语义：所有子锁必须全部加锁成功，整个 MultiLock 才算加锁成功。这保证了多资源操作的原子性——要么全部锁定，要么全部不锁定。
2. 失败自动回滚：在加锁过程中，如果某个子锁加锁失败，MultiLock 会自动释放已经成功加锁的子锁，避免部分锁被持有而阻塞其他线程。
3. 统一生命周期管理：MultiLock 统一管理所有子锁的加锁、解锁、续期操作，对外暴露与普通 RLock 完全一致的接口，降低使用复杂度。

4.2 加锁流程详解

MultiLock 的加锁过程可以分解为以下几个步骤：

步骤一：参数校验与准备

在tryLock方法被调用时，MultiLock 首先对参数进行校验，计算每个子锁的等待超时时间。如果调用者指定了waitTime，MultiLock 会将其均分给每个子锁（即每个子锁的等待超时 = 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 List<RLock> locks = new ArrayList<>(); /** 构造方法：接收多个 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 List<RLock> 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(); // 已加锁的子锁列表，用于失败回滚 List<RLock> acquiredLocks = new ArrayList<>(locks.size()); // 遍历所有子锁，依次加锁 for (ListIterator<RLock> 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(Collection<RLock> locks) { List<RFuture<Void>> futures = new ArrayList<>(locks.size()); // 遍历所有已加锁的子锁，异步执行解锁 for (RLock lock : locks) { futures.add(lock.unlockAsync()); } // 等待所有解锁操作完成 for (RFuture<Void> unlockFuture : futures) { unlockFuture.awaitUninterruptibly(); } }

这里使用了异步解锁机制（unlockAsync），并通过awaitUninterruptibly等待所有解锁操作完成。即使某个解锁操作失败（例如 Redis 连接断开），也不会影响其他子锁的解锁。

5.4 unlock 方法实现

@Override public void unlock() { // 获取所有子锁（包括子类可能添加的额外锁） List<RFuture<Void>> futures = new ArrayList<>(locks.size()); for (RLock lock : locks) { futures.add(lock.unlockAsync()); } // 等待所有解锁操作完成 for (RFuture<Void> 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 RFuture<Boolean> renewExpirationAsync(long threadId) { // 为每个子锁创建续期 Future List<RFuture<Boolean>> 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 自动续期

如果不指定 leaseTime，Redisson 的 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 实例的 MultiLock

MultiLock 支持跨不同的 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); RFuture<Boolean> 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()) < 0) { return OrderResult.fail("账户余额不足"); } // 2. 检查库存 Inventory inventory = inventoryService.getInventory(productId); if (inventory.getStock() < 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) < 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) < 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(List<String> taskIds) { if (taskIds.size() > MAX_BATCH_SIZE) { throw new IllegalArgumentException("批量任务数超过上限"); } // 为每个任务创建锁 List<RLock> taskLocks = taskIds.stream() .map(id -> 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 性能对比

从性能角度看，MultiLock 的开销与子锁数量成正比。建议在实际使用中：

• 子锁数量控制在 2-5 个，过多的子锁会显著增加延迟。
• 如果子锁都在同一 Redis 实例，考虑使用 Lua 脚本批量操作替代 MultiLock。
• 如果子锁分布在不同 Redis 实例，MultiLock 是最佳选择。

8.2 可靠性对比

8.3 适用场景对比

九、注意事项与最佳实践

9.1 死锁预防

虽然 MultiLock 内部会自动回滚失败加锁，但在多线程并发场景下，仍可能出现活锁问题。以下是一些预防措施：

1. 统一加锁顺序

对所有需要加锁的资源，按照固定的规则（如资源 ID 字典序）排序后再传入 MultiLock，保证所有线程以相同顺序获取锁：

// 排序后再创建 MultiLock List<String> resourceIds = Arrays.asList("resourceB", "resourceA", "resourceC"); Collections.sort(resourceIds); // 字典序排序 List<RLock> 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 true

2. 合理配置连接池

跨实例的 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 / 3

9.3 异常处理最佳实践

public void executeWithMultiLock(List<String> resourceIds) { // 1. 排序资源 ID，避免死锁 Collections.sort(resourceIds); List&lt;RLock&gt; 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() < deadline && retries < maxRetries) { try { long remaining = deadline - System.currentTimeMillis(); if (remaining <= 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 < SEGMENT_COUNT; i++) { segmentLocks[i] = redisson.getLock("lock:segment:" + i); } } /** 对一批资源 ID 加锁，使用分段减少锁数量 */ public RLock getMultiLock(List<String> resourceIds) { // 计算每个资源属于哪个分段 Set<Integer> segments = resourceIds.stream() .map(id -> Math.abs(id.hashCode()) % SEGMENT_COUNT) .collect(Collectors.toSet()); // 只对涉及的分段加锁，大幅减少锁数量 RLock[] locks = segments.stream() .sorted() // 排序避免死锁 .map(seg -> segmentLocks[seg]) .toArray(RLock[]::new); return redisson.getMultiLock(locks); } }

问题三