Reddission 分布式锁原理

总结:

  • 使用无参的tryLock()方法时,redisson会自动添加一个定时任务,定时刷新锁的失效时间,如果unlock时失败,则会出现该锁一直不释放的情况, 因为定时刷新的任务一直存在。
  • 使用两个参数的tryLock(long waitTime, TimeUnit unit)方法时,比无参的多了个功能就是在waitTime内,重试获取锁,直到超时,返回失败
  • tryLock(long waitTime, long leaseTime, TimeUnit unit)传释放时间leaseTime时,在waitTime内,重试获取锁,直到超时,返回失败。但不会添加定时刷新锁的失效时间的任务

加锁过程与解锁过程

redission 加锁采用了redis 的hash结构,
在这里插入图片描述

源码进入RedissionLock无参的tryLock()

尝试获取锁的代码
在这里插入图片描述

在这里插入图片描述
将当前获取锁线程的id传进去
在这里插入图片描述

   private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    // leaseTime 就是租约时间,就是Redis key 的过期时间
    // 无参的tryLock 这里必定为-1 不会进入这个判断
        if (leaseTime != -1) {
            return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        }
        
        // commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout() 获取到的值
        //    private long lockWatchdogTimeout = 30 * 1000; 这个是获取的默认的超时时间30s
        RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                                    commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                                    TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
     //等待获取加锁的结果
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        	// 存在异常 直接返回
            if (e != null) {
                return;
            }
			// 加锁成功,下面代码实现锁超时重试,也就是看门狗的逻辑
            // lock acquired
            if (ttlRemaining) {
                scheduleExpirationRenewal(threadId);
            }
        });
        return ttlRemainingFuture;
    }

这块代码执行获取锁的流程
锁可重入的逻辑在lua脚本中

    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    // 将超时时间保存在RedissonLock的 internalLockLeaseTime 变量中,用来解决锁超时问题 watchDog机制
        internalLockLeaseTime = unit.toMillis(leaseTime);
	// 真实的获取锁的过程 
	// getName()     lock = redissonClient.getLock("order"); 中的order
	//Collections.singletonList(getName()) 获取锁的key也就是当前锁的名称  传入lua 脚本中的key 对应KEYS[1]
	// internalLockLeaseTime 超时时间 传入lua脚本的参数[1] 对应ARGV[1]
	// getLockName(threadId) 线程唯一标识   对应hash结构的field hash结构的value对应重入次数
        return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                "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]);",
                Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

上面 字符串对应的lua文件,这个是我自己写的 和代码中的逻辑是一样的

--- KEYS[1] 是锁的key
--- ARGV[1] 是超时时间
--- ARGV[2] 是锁线程名称
local key = KEYS[1] --锁的key
local releaseTime = ARGV[1] -- 锁超时时间
local threadId = ARGV[2]  -- 线程唯一标识
--- 所不存在执行的流程
if (redis.call("exists", key) == 0) then
    -- 不存在获取锁
    redis.call("hset", key, threadId, '1');
    -- 设置有效期
    redis.call("pexpire", key, releaseTime);
    return nil;
end
--- 锁存在执行的流程
if (redis.call("hexists", key, threadId) == 1) then
	-- 锁重入 当前线程对应的value增加一
    redis.call("hincrby",key,threadId,'1');
    -- 重置超时时间
    redis.call("pexpire", key, releaseTime);
    return nil;  --返回结果
end

--- 当 key 不存在时,返回 -2 。 当 key 存在但没有设置剩余生存时间时,返回 -1 。 否则,以毫秒为单位,返回 key 的剩余生存时间。
return redis.call('pttl', KEYS[1])

scheduleExpirationRenewal(long threadId)

   private void scheduleExpirationRenewal(long threadId) {
        ExpirationEntry entry = new ExpirationEntry();
        //判断map 中是否存在当前线程对象的实体,如果存在则返回实体,如果不存在则创建新的返回null
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        
        if (oldEntry != null) {
            oldEntry.addThreadId(threadId);
        } else {
            entry.addThreadId(threadId);
            // 看门狗实现,创建新的实体需要增加看门狗的逻辑
            renewExpiration();
        }
    }

renewExpiration

   private void renewExpiration() {
   // 从map中获取ExpirationEntry 如果为null则直接返回
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ee == null) {
            return;
        }
        // 如果不为空则创建一个延时任务 task 与Timer返回的TimerTask关联的句柄。
        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
                if (ent == null) {
                    return;
                }
                Long threadId = ent.getFirstThreadId();
                if (threadId == null) {
                    return;
                }
                //在重新设置超时时间
                RFuture<Boolean> future = renewExpirationAsync(threadId);
                future.onComplete((res, e) -> {
                    if (e != null) {
                        log.error("Can't update lock " + getName() + " expiration", e);
                        return;
                    }
                    
                    if (res) {
                        // reschedule itself
                        renewExpiration();
                    }
                });
            }
            //internalLockLeaseTime   就是我们之前获取到的leaseTime  不传默认30秒
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
        
        // 将延时任务放入到ExpirationEntry 中    
     ee.setTimeout(task);
    }
    // 重新设置超时时间的代码
   protected RFuture<Boolean> renewExpirationAsync(long threadId) {
        return evalWriteAsync(getName(), 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(getName()),
                internalLockLeaseTime, getLockName(threadId));
    }

源码进入RedissionLock两个参数的tryLock(long waitTime, TimeUnit unit)

在这里插入图片描述
在这里插入图片描述

 @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        // 将等待时间转换成毫秒
        long time = unit.toMillis(waitTime);
        // 获取当前时间
        long current = System.currentTimeMillis();
        // 获取当前线程id
        long threadId = Thread.currentThread().getId();
        // 源码看下图
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }
        
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
        
        current = System.currentTimeMillis();
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(waitTime, unit, threadId);
            return false;
        }

        try {
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        
            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

在这里插入图片描述

    private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                                commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
            if (e != null) {
                return;
            }

            //  获取锁成功,添加锁超时延时任务
            if (ttlRemaining == null) {
                scheduleExpirationRenewal(threadId);
            }
        });
        return ttlRemainingFuture;
    }
    
      private void scheduleExpirationRenewal(long threadId) {
        ExpirationEntry entry = new ExpirationEntry();
        // 判断是否是重入的锁
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        // 是的话,直接将线程id添加到oldEntry中
        if (oldEntry != null) {
            oldEntry.addThreadId(threadId);
        } else {
            entry.addThreadId(threadId);
            //添加锁延时任务    源码是无参的renewExpiration
            renewExpiration();
        }
    }
Logo

华为云1024程序员节送福利,参与活动赢单人4000元礼包,更有热门技术干货免费学习

更多推荐