分布式锁

DeepSeek V3 中英对照 Distributed Locks

在许多情况下，针对某个上下文（甚至单个消息）的操作必须以独占方式执行。一个例子是聚合器组件，我们必须检查当前消息的消息组状态，以确定是否可以释放该组，或者只是将该消息添加到待处理队列中供后续处理。为此，Java 提供了 java.util.concurrent.locks.Lock 实现的 API。然而，当应用程序是分布式的和/或在集群中运行时，问题变得更加复杂。在这种情况下，锁定具有挑战性，需要某种共享状态及其特定方法来实现独占性要求。

Spring Integration 提供了一个 LockRegistry 抽象，并基于 ReentrantLock API 实现了内存中的 DefaultLockRegistry。LockRegistry 的 obtain(Object) 方法需要一个特定上下文的 lock key。例如，聚合器使用 correlationKey 来锁定其分组相关的操作。这样，不同的锁可以并发使用。obtain(Object) 方法返回一个 java.util.concurrent.locks.Lock 实例（具体取决于 LockRegistry 的实现），因此，其余的逻辑与标准的 Java 并发算法相同。

从 6.2 版本开始，LockRegistry 提供了一个 executeLocked() API（此接口中的 default 方法），用于在锁定状态下执行某些任务。此 API 的行为类似于众所周知的 JdbcTemplate、JmsTemplate 或 RestTemplate。以下示例演示了此 API 的用法：

LockRegistry registry = new DefaultLockRegistry();
...
registry.executeLocked("someLockKey", () -> someExclusiveResourceCall());

该方法会重新抛出任务调用中的异常，如果 Lock 被中断则抛出 InterruptedException。此外，带有 Duration 参数的变体在 lock.tryLock() 返回 false 时会抛出 java.util.concurrent.TimeoutException。

Spring Integration 为分布式锁提供了以下 LockRegistry 实现：

• JDBC

• Redis

• Zookeeper

Spring Cloud AWS 同样提供了 DynamoDbLockRegistry。

从 7.0 版本开始，引入了 DistributedLock 接口，提供了新方法 lock(Duration ttl) 和 tryLock(long time, TimeUnit unit, Duration ttl)，用于获取具有自定义生存时间（TTL）的锁。JdbcLock 和 RedisLock 都实现了 DistributedLock 接口，以支持自定义生存时间的功能。LockRegistry<L extends Lock> 现在是一个泛型接口，适用于扩展 Lock 的类型。RenewableLockRegistry 接口现在提供了新的 renewLock(Object lockKey, Duration ttl) 方法，允许您使用自定义的生存时间值来续期锁。JdbcLockRegistry 和 RedisLockRegistry 都实现了 LockRegistry 和 RenewableLockRegistry 接口，其类型参数为 DistributedLock。

以下是一个示例，展示如何从注册表获取 DistributedLock 并使用特定的生存时间值来获取锁：

DistributedLock lock = registry.obtain("foo");
Duration timeToLive = Duration.ofMillis(500);

if(lock.tryLock(100, TimeUnit.MILLISECONDS, timeToLive)){
    try {
        // do something
    } catch (Exception e) {
        // handle exception
    } finally{
        lock. unlock();
    }
}