RedisLockRegistry源码分析

前言

官网的英文介绍大概如下：

Starting with version 4.0, the RedisLockRegistry is available. Certain components (for example aggregator and resequencer) use a lock obtained from a LockRegistry instance to ensure that only one thread is manipulating a group at a time. The DefaultLockRegistry performs this function within a single component; you can now configure an external lock registry on these components. When used with a shared MessageGroupStore, the RedisLockRegistry can be use to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.

When a lock is released by a local thread, another local thread will generally be able to acquire the lock immediately. If a lock is released by a thread using a different registry instance, it can take up to 100ms to acquire the lock.

To avoid "hung" locks (when a server fails), the locks in this registry are expired after a default 60 seconds, but this can be configured on the registry. Locks are normally held for a much smaller time.

上述大概意思是RedisLockRegistry可以确保在分布式环境中，只有一个thread在执行，也就是实现了分布式锁，当一个本地线程释放了锁，其他本地现场会立即去抢占锁，如果锁被占用了，那么会进行重试机制，100毫秒进行重试一次。同时也避免了"hung" locks 当服务器fails的时候。同时也给锁设置了默认60秒的过期时间

如何获取锁

详细流程如上图所示，这里主要核心业务是这样，首先Lock是java.util.concurrent.locks中的锁，也就是本地锁。然后自己用RedisLock实现了Lock接口而已，但是实际上RedisLock也使用了本地锁。主要是通过redis锁＋本地锁双重锁的方式实现的一个比较好的锁。针对redis锁来说只要能获取到锁，那么就算是成功的。如果获取不到锁就等待100毫秒继续重试，如果获取到锁那么就采用本地锁锁住本地的线程。通过两种方式很好的去实现了一个完善的分布式锁机制。

下面代码主要是获取锁的一个流程，先从本地锁里面获取，如果获取到了那么和redis里面存放的RedisLock锁做对比，判断是否是同一个对象，如果不是那么就删除本地锁然后重新创建一个锁返回

@Override
public Lock obtain(Object lockKey) {
    Assert.isInstanceOf(String.class, lockKey);

    //try to find the lock within hard references
    //从本地强引用里面获取锁,
    RedisLock lock = findLock(this.hardThreadLocks.get(), lockKey);

    /*
     * If the lock is locked, check that it matches what's in the store.
     * If it doesn't, the lock must have expired.
     */
    //这里主要判断了这个锁是否是锁住的,如果不是的那么该锁已经过期了
    //如果强引用里面有这个锁,并且lock.thread!=null，说明这个锁没有被占用
    if (lock != null && lock.thread != null) {
        //从redis获取锁,若如果redis锁为空或者跟当前强引用的锁不一致,可以确定两个问题
        //1.redis里面的锁和本地的锁不是一个了
        //2.redis里面没有锁
        RedisLock lockInStore = this.redisTemplate.boundValueOps(this.registryKey + ":" + lockKey).get();
        if (lockInStore == null || !lock.equals(lockInStore)) {
            //删除强引用里面锁
            getHardThreadLocks().remove(lock);
            lock = null;
        }
    }
    //如果锁==null
    if (lock == null) {
        //try to find the lock within weak references
        //尝试线从弱引用里面去找锁
        lock = findLock(this.weakThreadLocks.get(), lockKey);
        //如果弱引用锁==null 那么新建一个锁
        if (lock == null) {
            lock = new RedisLock((String) lockKey);
            //判断是否用弱引用,如果用那么就加入到弱引用里面
            if (this.useWeakReferences) {
                getWeakThreadLocks().add(lock);
            }
        }
    }

    return lock;
}

上面获取到的是RedisLock，RedisLock是实现java原生Lock接口，并重写了lock()方法。首先从localRegistry中获取到锁，这里的锁是java开发包里面的ReentrantLock。首先把本地先锁住，然后再去远程obtainLock。每次sleep() 100毫秒直到获取到远程锁为止，代码如下所示：

@Override
public void lock() {
    //这里采用java开发包里面的ReentrantLock 进行多线程的加锁,单机多线程的情况下解决并发的问题
    Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
    localLock.lock();
    while (true) {
        try {
            while (!this.obtainLock()) {
                Thread.sleep(100); //NOSONAR
            }
            break;
        }
        catch (InterruptedException e) {
                /*
                 * This method must be uninterruptible so catch and ignore
                 * interrupts and only break out of the while loop when
                 * we get the lock.
                 */
        }
        catch (Exception e) {
            localLock.unlock();
            rethrowAsLockException(e);
        }
    }
}

核心远程锁还是在RedisLock中，这里采用了redis事务＋watch的方式，watch和事务都是redis里面自带的。使用watch时候如果key的值发生了任何变化。那么exec()将不会执行，那么如下代码返回的success就是false。从而来实现redis锁的功能

private boolean obtainLock() {
    //判断创建这个类的线程和当前是否是一个,如果是就直接获取锁
    Thread currentThread = Thread.currentThread();
    if (currentThread.equals(this.thread)) {
        this.reLock++;
        return true;
    }
    //把当前锁存到集合种
    toHardThreadStorage(this);

    /*
     * Set these now so they will be persisted if successful.
     */
    this.lockedAt = System.currentTimeMillis();
    this.threadName = currentThread.getName();

    Boolean success = false;
    try {
        success = RedisLockRegistry.this.redisTemplate.execute(new SessionCallback<Boolean>() {

            @SuppressWarnings({"unchecked", "rawtypes"})
            @Override
            public Boolean execute(RedisOperations ops) throws DataAccessException {
                String key = constructLockKey();
                //监控key如果该key被改变了 那么该事务是不能被实现的会进行回滚
                ops.watch(key); //monitor key
                //如果key存在了就停止监控，如果key已经存在了 那么肯定是被别人占用了
                if (ops.opsForValue().get(key) != null) {
                    ops.unwatch(); //key already exists, stop monitoring
                    return false;
                }

                ops.multi(); //transaction start
                //设置一个值并加上过期时间 m默认是一分钟左右的时间
                //set the value and expire
                //把锁放入到redis中
                ops.opsForValue()
                        .set(key, RedisLock.this, RedisLockRegistry.this.expireAfter, TimeUnit.MILLISECONDS);

                //exec will contain all operations result or null - if execution has been aborted due to 'watch'
                return ops.exec() != null;
            }

        });

    }
    finally {
      //如果不成功那么把当前过期时间和锁的名字设置成null
        if (!success) {
            this.lockedAt = 0;
            this.threadName = null;
            toWeakThreadStorage(this);
        }
        else {
        //如果成功把当前锁的thread名称设置成currentThread
            this.thread = currentThread;
            if (logger.isDebugEnabled()) {
                logger.debug("New lock; " + this.toString());
            }
        }

    }

    return success;
}

上面是整个加锁的流程,基本流程比较简单，看完加锁应该自己都能解锁，无非就是去除redis锁和本地的锁而已。

@Override
public void unlock() {
    //判断当前运行的线程和锁的线程做对比，如果两个线程不一样那么抛出异常
    if (!Thread.currentThread().equals(this.thread)) {
        if (this.thread == null) {
            throw new IllegalStateException("Lock is not locked; " + this.toString());
        }
        throw new IllegalStateException("Lock is owned by " + this.thread.getName() + "; " + this.toString());
    }

    try {
       //如果reLock－－小于＝0的话就删除redis里面的锁
        if (this.reLock-- <= 0) {
            try {
                this.assertLockInRedisIsUnchanged();
                RedisLockRegistry.this.redisTemplate.delete(constructLockKey());
                if (logger.isDebugEnabled()) {
                    logger.debug("Released lock; " + this.toString());
                }
            }
            finally {
                this.thread = null;
                this.reLock = 0;
                toWeakThreadStorage(this);
            }
        }
    }
    finally {
    //拿到本地锁，进行解锁
        Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
        localLock.unlock();
    }
}

tryLock在原有的加锁上面增加了一个超时机制，主要是先通过本地的超时机制

@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    //拿到本地锁
    Lock localLock = RedisLockRegistry.this.localRegistry.obtain(this.lockKey);
    //先本地锁进行tryLock
    if (!localLock.tryLock(time, unit)) {
        return false;
    }
    try {
        long expire = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(time, unit);
        boolean acquired;
        //这里添加了超时机制，跟之前的无限等待做了一个区分
        while (!(acquired = obtainLock()) && System.currentTimeMillis() < expire) { //NOSONAR
            Thread.sleep(100); //NOSONAR
        }
         //超时后没有获取到锁，那么就把本地锁进行解锁
        if (!acquired) {
            localLock.unlock();
        }
        return acquired;
    }
    catch (Exception e) {
        localLock.unlock();
        rethrowAsLockException(e);
    }
    return false;
}