Redis分布式锁（三）：支持锁可重入，避免锁递归调用时死锁

使用现状

Redis分布式锁的基础内容，我们已经在Redis分布式锁：基于AOP和Redis实现的简易版分布式锁这篇文章中讲过了，也在文章中示范了正常的加锁和解锁方法。

Redis分布式锁为何要支持续期，以及如何支持续期的方法，我们也已经在Redis分布式锁（二）：支持锁的续期，避免锁超时后导致多个线程获得锁

从上次升级为可续期的分布式锁后的半年时间内，这款自研的简易版分布式锁依然运行良好。

发现问题

但在最近查线上日志的时候偶然发现，有一个业务场景下，分布式锁会发生死锁。

我们经过初步排查，定位到是因为在一个已持有锁的方法中调用了另一个需要加锁执行的方法导致的。

简化后的函数如下图所示：

@Servicepublic class Lock1ServiceImpl implements Lock1Service {    @Override    @LockAnnotation(lockField = "test", lockKey = "1")    public void test() {        }}@Servicepublic class Lock2ServiceImpl implements Lock2Service {    @Autowired    private Lock1Service lock1Service;    @Override    @LockAnnotation(lockField = "test", lockKey = "1")    public void test(){        lock1Service.test();    }}

死锁的过程主要如下：

Lock2Service的test方法需要加锁（锁的key为"test:1"），然后在方法内部调用了Lock1Service的test方法，而Lock1Service的test方法也需要加同样的锁。

在执行到lock1Service.test()时，Lock2Service的test方法并没有执行结束，所以不会释放锁，而lock1Service执行test方法又必须去拿到锁，这时候就发生死锁了。

这段代码只能等lock1Service经过一段时间等待后主动放弃继续拿锁后才能继续往下进行。而lock1Service的test方法将永远也无法执行。

解决方案

问题既然已经出现了，那么接下来我们该做的就是想办法来尽量避免这个情况。

我们很快就想到了jdk中自带的ReentrantLock，我们可以按照同样的原理实现可重入的分布式锁。

实现可重入的原理也和ReentrantLock的原理类似，和ReentrantLock的不同之处在于分布式锁在第一次去获取锁的时候需要采用redis去分布式竞争。而当已持有锁且需要重入的时候，分布式锁会降级为本地锁。只有同一个线程的资源才有重入的概念。

以下是lock类的主要细节，加锁过程和解锁过程如下所示：

final Boolean tryLock(String lockValue, int waitTime) {    long startTime = System.currentTimeMillis();    long endTime = startTime + waitTime * 1000;    try {        do {            final Thread current = Thread.currentThread();            int c = this.getState();            if (c == 0) {                int lockTime = LOCK_TIME;                if (lockRedisClient.setLock(lockKey, lockValue, lockTime)) {                    lockOwnerThread = current;                    this.setState(c + 1);                    survivalClamProcessor = new SurvivalClamProcessor(lockKey, lockValue, lockTime, lockRedisClient);                    (survivalThread = threadFactoryManager.getThreadFactory().newThread(survivalClamProcessor)).start();                    log.info("线程获取重入锁成功,锁的名称为{}", lockKey);                    return Boolean.TRUE;                }            } else if (lockOwnerThread == Thread.currentThread()) {                if (c + 1 < 0) {                    throw new Error("Maximum lock count exceeded");                }                this.setState(c + 1);                log.info("线程重入锁成功,锁的名称为{},当前LockCount为{}", lockKey, state);                return Boolean.TRUE;            }            int sleepTime = SLEEP_TIME_ONCE;            if (waitTime > 0) {                log.info("线程暂时无法获得锁,当前已等待{}ms,本次将再等待{}ms,锁的名称为{}", System.currentTimeMillis() - startTime, sleepTime, lockKey);                try {                    Thread.sleep(sleepTime);                } catch (InterruptedException e) {                    log.info("线程等待过程中被中断,锁的名称为{}", lockKey, e);                }            }        } while (System.currentTimeMillis() <= endTime);        if (waitTime == 0) {            log.info("线程获得锁失败,将放弃获取锁,锁的名称为{}", lockKey);        } else {            log.info("线程获得锁失败,之前共等待{}ms,将放弃等待获取锁,锁的名称为{}", System.currentTimeMillis() - startTime, lockKey);        }        return Boolean.FALSE;    } catch (Exception e) {        return Boolean.FALSE;    }}

final void unLock(String lockValue) {    if (lockOwnerThread == Thread.currentThread()) {        int c = this.getState() - 1;        if (c == 0) {            this.setLockOwnerThread(null);            survivalClamProcessor.stop();            survivalThread.interrupt();            this.setSurvivalClamProcessor(null);            this.setSurvivalThread(null);            this.setState(c);            lockRedisClient.delLock(lockKey, lockValue);            log.info("重入锁LockCount-1,线程已成功释放锁,锁的名称为{}", lockKey);        } else {            this.setState(c);            log.info("重入锁LockCount-1,锁的名称为{}，剩余LockCount为{}", lockKey, c);        }    }}

然后为了避免使用者经常会忘记解锁或解锁不规范，所以加解锁的方法都不对外暴露，只对外暴露execute方法：

public <T> T execute(Supplier<T> supplier, int waitTime) {    String randomValue = UUID.randomUUID().toString();    Boolean holdLock = Boolean.FALSE;    try {        if (holdLock = this.tryLock(randomValue, waitTime)) {            return supplier.get();        }        return null;    } catch (Exception e) {        log.error("execute error", e);        return null;    } finally {        if (holdLock) {            this.unLock(randomValue);        }    }}

另外在aop实现的时候，因为无法从上下文中获取到同一个lock对象，故需要通过lockManager.getLock(lockField, lockKey)去获取到lock对象。如果lockPrefix和lockKey一致的话，将获得到同一个lock对象，从而实现可重入的功能。

public Lock getLock(String lockPrefix, String lockKey) {    String finalLockKey = StringUtils.isEmpty(lockPrefix) ? lockKey : (lockPrefix.concat(":").concat(lockKey));    if (lockMap.containsKey(finalLockKey)) {        return lockMap.get(finalLockKey);    } else {        Lock lock = new Lock(finalLockKey, lockRedisClient, threadFactoryManager);        Lock existLock = lockMap.putIfAbsent(finalLockKey, lock);        return Objects.nonNull(existLock) ? existLock : lock;    }}

更新说明

本次更新，除了实现了分布式锁的可重入功能之外，另外还在声明式分布式锁@LockAnnotation注解基础上，实现了通过execute方法来实现的编程式分布式锁。

此外，由于之前版本已实现可续期的功能，所以LockAnnotation上的lockTime标记为已过期，锁的过期时间统一改为30s。通过续期功能来实现需要长时间锁定的功能。

后续计划

目前实现的版本中，已可满足分布式锁的大部分场景（非公平+可自动续期+可重入的分布式锁），已可投入生产环境使用。但目前仍不支持公平锁，而且在竞争锁失败时采用了自旋+线程等待的方式实现了线程阻塞，后续可能会往这两个方向去优化。

好了，我们下一期再见，欢迎大家一起留言讨论。同时也欢迎点赞，欢迎送小星星~

.