前言

redis实现锁的机制很复杂，这里主要分析以下

SETNX

该命令来自于 SET if Not Exists 的缩写，意思是如果 key 不存在，则设置 value ，否则失败返回0

要释放锁也简单，就是通过 del 删除或者 expire 设置过期时间

1
2
3
4
> SETNX lock:168 1  // 获取锁
(integer) 1
> EXPIRE lock:168 60  // 60s 自动删除
(integer) 1

「加锁」和「设置超时」是两个命令，不是 原子操作

于是出现了

SET resource_name random_value NX PX timeout

该命可以在加锁的同时设置过期时间

释放了不是自己的锁

  1. 客户1获取锁成功设置 30 秒超时
  2. 因为执行很慢，导致锁「自动释放」
  3. 客户2申请加锁成功
  4. 客户1执行完成，执行 DEL 释放了客户2的锁

要解决这个问题需要从 random_value 下手，在释放锁的时候将自己的「唯一标识」与锁上的「标识」进行比较，匹配上则删除，伪代码：

1
2
3
4
// 比对 value 与 唯一标识
if (redis.get("lock").equals(random_value)){
   redis.del("lock"); //比对成功则删除
}

这里 GET + DEL 命令组合又会涉及到原子问题，通过 LUA 脚本实现

正确设置锁超时

锁的超时时间计算，是个很复杂的问题，包括 网络 IO、 JVM Full GC

有没有完美方案？

我们可以让获得锁的线程开启一个守护线程，用来给快要过期的锁「续航」
加锁的时候设置一个过期时间，同时客户端开启一个「守护线程」，定时去检测这个锁的失效时间
如果快要过期，但是业务逻辑还没执行完成，自动对这个锁进行续期，重新设置过期时间

Redission 分布式锁的实现，就是基于这种思想，守护线程叫「看门狗」

可重入锁

Redission 类库通过 Redis Hash来实现可重入锁

当线程拥有锁之后，往后再遇到加锁方法，直接将加锁次数加 1，然后再执行方法逻辑
退出加锁方法之后，加锁次数再减 1，当加锁次数为 0 时，锁才被真正的释放
可以看到可重入锁最大特性就是计数，计算加锁的次数

加锁逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
---- 1 代表 true
---- 0 代表 false
if (redis.call('exists', KEYS[1]) == 0) then
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return 1;
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 1;
end ;
return 0;

加锁代码首先使用 Redis exists 命令判断当前 lock 是否存在
如果不存在，则使用 hincrby 命令创建一个键为 KEYS[1] ，Hash 表中的键为 ARGV[2] ，然后加 1 ，最后设置过期时间
如果锁存在，则使用 hexists 判断当前 lock 对应的 hash 表是否存在 KEYS[1] ARGV[2] 这个键，如果存在，再使用 hincrby 加 1 ，最后再设置过期时间

解锁逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
-- 判断 hash set 可重入 key 的值是否等于 0
-- 如果为 0 代表 该可重入 key 不存在
if (redis.call('hexists', KEYS[1], ARGV[1]) == 0) then
    return nil;
end ;
-- 计算当前可重入次数
local counter = redis.call('hincrby', KEYS[1], ARGV[1], -1);
-- 小于等于 0 代表可以解锁
if (counter > 0) then
    return 0;
else
    redis.call('del', KEYS[1]);
    return 1;
end ;
return nil;

解锁逻辑…

主从架构问题

之前的分析场景都是锁在「单个」Redis 实例中，并没涉及集群

Redis 为统一分布式锁标准，搞了一个 Redlock (红锁)

Red lock

官方推荐部署 5 个以上 Redis 主节点

  1. 客户端获取当前时间 T1 (毫秒级别)
  2. 使用相同的 key 和 value 尝试从 N 个 Redis 实例上获取锁
  3. 每个请求都设置一个超时时间(毫秒级别)，改超时时间要远小于锁的有效时间，便于快取获取下一把锁
  4. 客户端获取当前时间 T2 并减去步骤 1 的 T1 计算获取锁的耗时。 当且仅当客户端在大多数实例（N/2 + 1）获取成功，且获取锁所用的总时间 T3 小于锁的有效时间，才认为加锁成功，否则加锁失败
    • 如果加锁失败，客户端应该在所有的 Redis 实例上进行解锁(有漏洞)

缺点：

  1. red lock 锁场景复杂，且占用资源
  2. 如果在极端情况(多个 redis 实例故障)可能导致锁失效风险

redission 分布式锁

失败无限重试

1
2
3
4
5
6
7
8
9
RLock lock = redisson.getLock("lock");
try {
  // 1.最常用的第一种写法
  lock.lock();
  // 执行业务逻辑
  .....
} finally {
  lock.unlock();
}

拿锁失败会不停重试，有 watch dog 自动延期机制，默认续 30s 每隔 30 / 3 = 10 秒续约到 30 秒

失败超时重试，自动虚名

1
2
// 尝试拿锁10s后停止重试,获取失败返回false，具有Watch Dog 自动延期机制， 默认续30s
boolean flag = lock.tryLock(10, TimeUnit.SECONDS);

超时自动释放锁

1
2
// 没有Watch Dog ，10s后自动释放,不需要调用 unlock 释放锁。
lock.lock(10, TimeUnit.SECONDS);

超时重试，自动解锁

1
2
3
4
5
6
7
8
9
// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁,没有 Watch dog
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
       lock.unlock();
   }
}

Watch Dog 自动延时

Redisson 提供了 watch dog 自动延时机制，提供了一个监控锁的看门狗，它的作用是在 Redisson 实例被关闭前，不断的延长锁的有效期
也就是说，如果一个拿到锁的线程一直没有完成逻辑，那么看门狗会帮助线程不断的延长锁超时时间，锁不会因为超时而被释放。

默认情况下，看门狗的续期时间是 30s，也可以通过修改 Config.lockWatchdogTimeout 来另行指定。

另外 Redisson 还提供了可以指定 leaseTime 参数的加锁方法来指定加锁的时间。
超过这个时间后锁便自动解开了，不会再延长锁的有效期。

  • watchdog 只有在未显示指定加锁超时时间(leaseTime)时才会生效
  • lockWatchdogTimeout 设定的时间不要太小 ，比如设置的是 100 毫秒，由于网络直接导致加锁完后，watchdog 去延期时，这个 key 在 redis 中已经被删除了。

Redission源码解读

在调用 lock 方法时，会链式调用 lock()->tryAcquire->tryAcquireAsync 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        RFuture<Long> ttlRemainingFuture;
        //如果指定了加锁时间，会直接去加锁
        if (leaseTime != -1) {
            ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        } else {
            //没有指定加锁时间 会先进行加锁，并且默认时间就是 LockWatchdogTimeout的时间
            //这个是异步操作 返回RFuture 类似netty中的future
            ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                    TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        }
        //这里也是类似netty Future 的addListener，在future内容执行完成后执行
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
            if (e != null) {
                return;
            }
            // lock acquired
            if (ttlRemaining == null) {
                // leaseTime不为-1时，不会自动延期
                if (leaseTime != -1) {
                    internalLockLeaseTime = unit.toMillis(leaseTime);
                } else {
                    //这里是定时执行 当前锁自动延期的动作,leaseTime为-1时，才会自动延期
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }

scheduleExpirationRenewal 会调用 renewExpiration 启用一个 timeout 定时，去执行延期动作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
private void renewExpiration() {
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ee == null) {
            return;
        }
        Timeout task = commandExecutor.getConnectionManager()
          .newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                // 省略部分代码
                ....
                RFuture<Boolean> future = renewExpirationAsync(threadId);
                future.onComplete((res, e) -> {
                    ....
                    if (res) {
                        //如果 没有报错，就再次定时延期
                        // reschedule itself
                        renewExpiration();
                    } else {
                        cancelExpirationRenewal(null);
                    }
                });
            }
            // 这里我们可以看到定时任务 是 lockWatchdogTimeout 的1/3时间去执行 renewExpirationAsync
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
        ee.setTimeout(task);
    }

scheduleExpirationRenewal 会调用 renewExpirationAsync ，执行下面 lua脚本
主要判断锁是否存在 redis 中，如果存在旧进行 pexpire 延期

1
2
3
4
5
6
7
8
9
10
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return 0;",
                Collections.singletonList(getRawName()),
                internalLockLeaseTime, getLockName(threadId));
    }
  • watch dog 在当前节点还存活且任务未完成则每 10 s 给锁续期 30s。
  • 程序释放锁操作时因为异常没有被执行，那么锁无法被释放，所以释放锁操作一定要放到 finally {} 中；
  • 要使 watchLog 机制生效 ，lock 时 不要设置 过期时间(leaseTime)。
  • watchlog 的延时时间 可以由 lockWatchdogTimeout 指定默认延时时间，但是不要设置太小。
  • watchdog 会每 lockWatchdogTimeout/3 时间，去延时。
  • 通过 lua 脚本实现延迟。