Redis进阶- Redisson分布式锁实现原理及源码解析

举报
小工匠 发表于 2021/09/10 23:59:05 2021/09/10
【摘要】 文章目录 Pre用法Redisson分布式锁实现原理Redisson分布式锁源码分析redisson.getLock(lockKey) 的逻辑redissonLock.lock()的逻辑redis...

在这里插入图片描述


Pre

Redis进阶-细说分布式锁中我们梳理了使用Redis实现分布式锁的演进过程,并提出了目前最完善的解决方案:Redisson 实现分布式锁 。

这里我们来分析下Redisson分布式锁实现原理及源码解析


用法

使用redisson实现分布式锁的操作步骤,三部曲

  • 第一步: 获取锁 RLock redissonLock = redisson.getLock(lockKey);
  • 第二步: 加锁,实现锁续命功能 redissonLock.lock();
  • 第三步:释放锁 redissonLock.unlock();

Redisson分布式锁实现原理

熟悉了基本用法以后,我们来看下Redission实现分布式锁的原理,再理解了原理之后,后续梳理源码实现就更加得心应手了。

在这里插入图片描述


Redisson分布式锁源码分析

流程图如下

重点主要是依赖lua脚本的原子性,实现加锁和释放锁的功能

在这里插入图片描述

redisson.getLock(lockKey) 的逻辑

    @Override
    public RLock getLock(String name) {
        return new RedissonLock(connectionManager.getCommandExecutor(), name);
    }


  
 
  • 1
  • 2
  • 3
  • 4
  • 5

实例化RedissonLock,我们看下RedissonLock的构造函数

    public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
        this.commandExecutor = commandExecutor;
        this.id = commandExecutor.getConnectionManager().getId();
        this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
    }

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • super(commandExecutor, name); 父类name赋值,后续通过getName()获取

  • commandExecutor: 执行lua脚本的executor

  • id 是个UUID, 后面被用来当做 和threadId组成 value值,用作判断加锁和释放锁是否是同一个线程的校验。

  • internalLockLeaseTime : 取自 Config#lockWatchdogTimeout,默认30秒,这个参数还有另外一个作用,锁续命的执行周期 internalLockLeaseTime/3 = 10秒


redissonLock.lock()的逻辑

主要是实现加锁和锁的续命

 redissonLock.lock();

  
 
  • 1

看看都干了啥

  @Override
    public void lock() {
        try {
            lockInterruptibly();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }


  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

继续看 lockInterruptibly

 @Override
    public void lockInterruptibly() throws InterruptedException {
        lockInterruptibly(-1, null);
    }

  
 
  • 1
  • 2
  • 3
  • 4

继续看 lockInterruptibly(-1, null);

@Override
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
    
	    // 获取当前线程ID
        long threadId = Thread.currentThread().getId();
        
        // 尝试获取锁的剩余时间 
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired  ttl为空,说明没有线程持有该锁,直接返回 让当前线程加锁成功 
        if (ttl == null) {
            return;
        }

        RFuture<RedissonLockEntry> future = subscribe(threadId);
        commandExecutor.syncSubscription(future);
	
	
        // 死循环  
        try {
            while (true) {
            
                // 再此尝试获取锁的剩余时间 ,如果为null, 跳出循环
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message   如果ttl >=0 说明 有其他线程持有该锁
                if (ttl >= 0) {
                
                     // 获取信号量,尝试加锁,设置最大等待市场为ttl
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                
                     // 如果ttl小于0 (-1 ,-2 ) 说明已经过期,直接获取
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally {
            unsubscribe(future, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

  
 
  • 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
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

大流程已经梳理完了,我们看下 Long ttl = tryAcquire(leaseTime, unit, threadId);

 private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
        return get(tryAcquireAsync(leaseTime, unit, threadId));
    }

  
 
  • 1
  • 2
  • 3

继续看下

tryAcquireAsync(leaseTime, unit, threadId)

  
 
  • 1
 private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
 

        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        
        // 刚开始  leaseTime 传入的是 -1 ,所以走这个分支
        
        // 1)尝试加锁  待会细看 先把主要的逻辑梳理完
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        
       // 2) 注册监听事件
        ttlRemainingFuture.addListener(new FutureListener<Long>() {
            @Override
            public void operationComplete(Future<Long> future) throws Exception {
                if (!future.isSuccess()) {
                    return;
                }

                Long ttlRemaining = future.getNow();
                // lock acquired
                if (ttlRemaining == null) {
                
	                // 3)获取锁成功的话,给锁延长过期时间 
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }


  
 
  • 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
  • 30
  • 31
  • 32

继续看

 // 1)尝试加锁  待会细看 先把主要的逻辑梳理完
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);

  
 
  • 1
  • 2

看实现

    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      "redis.call('hset', 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

lua 脚本

KEYS[1] ---------> getName()
ARGV[1] ---------> internalLockLeaseTime
ARGV[2] ---------> getLockName(threadId) 实现如下

  String getLockName(long threadId) {
        return id + ":" + threadId;
    }

  
 
  • 1
  • 2
  • 3

这个id就是自开始实例化RedissonLock的id ,是个UUID

我们来解释下这段lua脚本

 // 如果 lockKey不存在 ,设置 使用hset设置 lockKey ,field为 uuid:threadId ,value为1 ,并设置过期时间

 //就是这个命令 

 //127.0.0.1:6379> hset lockkey  uuid:threadId 1
 //(integer) 1
 //127.0.0.1:6379> PEXPIRE lockkey internalLockLeaseTime

  "if (redis.call('exists', KEYS[1]) == 0) then " +
    "redis.call('hset', KEYS[1], ARGV[2], 1); " +
           "redis.call('pexpire', KEYS[1], ARGV[1]); " +
           "return nil; " +
       "end; " +
       
 // 如果 lockKey 存在和 filed 和 当前线程的uuid:threadId相同  key 加1 ,执行多少次 就加多次  设置过期时间  其实就是如下命令
 //127.0.0.1:6379> HEXISTS lockkey uuid:threadId
 //(integer) 1
 //127.0.0.1:6379> PEXPIRE lockkey  internalLockLeaseTime

   "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; " +
       
 // 最后返回 lockkey的 pttl 

   "return redis.call('pttl', KEYS[1]);"

  
 
  • 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

那继续监听时间中的 scheduleExpirationRenewal(threadId); 逻辑

 private void scheduleExpirationRenewal(final long threadId) {
        if (expirationRenewalMap.containsKey(getEntryName())) {
            return;
        }

        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        
           // 重点是run方法 
            @Override
            public void run(Timeout timeout) throws Exception {
                
                
               // 又是lua脚本  判断是否存在,存在就调用pexpire 
                RFuture<Boolean> future = commandExecutor.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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
                // 监听事件中又 调用了自己  scheduleExpirationRenewal
                future.addListener(new FutureListener<Boolean>() {
                    @Override
                    public void operationComplete(Future<Boolean> future) throws Exception {
                        expirationRenewalMap.remove(getEntryName());
                        if (!future.isSuccess()) {
                            log.error("Can't update lock " + getName() + " expiration", future.cause());
                            return;
                        }
                        
                        if (future.getNow()) {
                            // reschedule itself
                            scheduleExpirationRenewal(threadId);
                        }
                    }
                });
            }
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

        if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
            task.cancel();
        }
    }


  
 
  • 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
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

redissonLock.unlock();逻辑

  @Override
    public void unlock() {
        Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));
        if (opStatus == null) {
            throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + Thread.currentThread().getId());
        }
        if (opStatus) {
            cancelExpirationRenewal();
        }
    }

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

重点看 unlockInnerAsync(Thread.currentThread().getId())

 protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                "end;" +
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                    "return nil;" +
                "end; " +
                "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                "if (counter > 0) then " +
                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                "else " +
                    "redis.call('del', KEYS[1]); " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; "+
                "end; " +
                "return nil;",
                Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));

    }
    

  
 
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

又是lua脚本,核心就是 把value减到为0 ,删除key

KEYS[1] ---------> getName()

KEYS[2] ---------> getChannelName()
ARGV[1] ---------> LockPubSub.unlockMessage
ARGV[2] ---------> internalLockLeaseTime
ARGV[2] ---------> getLockName(threadId)


总结

至此,原理和源码我们粗略的梳理完了 ,梳理了主要的核心流程,主要是依靠lua脚本,代码写的还是非常优秀的,向开源学习!!!

文章来源: artisan.blog.csdn.net,作者:小小工匠,版权归原作者所有,如需转载,请联系作者。

原文链接:artisan.blog.csdn.net/article/details/105449188

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。