Redisson分布式锁源码篇

发布时间:2024年01月19日

☆* o(≧▽≦)o *☆嗨~我是小奥🍹
📄📄📄个人博客:小奥的博客
📄📄📄CSDN:个人CSDN
📙📙📙Github:传送门
📅📅📅面经分享(牛客主页):传送门
🍹文章作者技术和水平有限,如果文中出现错误,希望大家多多指正!
📜 如果觉得内容还不错,欢迎点赞收藏关注哟! ??

Redisson分布式锁源码篇

Redisson 是一个简单的Redis Java客户端,具有内存数据网格功能。

它不仅提供了一系列的分布式的Java常用对象,还提供了许多分布式服务,其中就包含了各种分布式锁的实现。

GitHub地址:https://github.com/redisson/redisson

Redisson分布式锁不仅是相对成熟的分布式锁方案,而且在很多企业中都会去使用的,所以了解一下底层的实现还是很有必要的。

一、使用Redisson分布式锁

Redisson分布式锁的使用非常简单,我们只需要在maven中引入依赖,然后直接调用相关API即可。

1.1 引入依赖

		<dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.13.6</version>
        </dependency>

1.2 调用API

		// 创建RedissonClient
        RedissonClient redissonClient = Redisson.create();
        // 创建分布式锁
        RLock lock = redissonClient.getLock("lock");
        // 加锁
        lock.lock();
        // 尝试获取锁,无参表示不等待立即返回
        boolean b = lock.tryLock();
        // 解锁
        lock.unlock();

二、源码解析

下面我就带大家去分析一下Redisson的可重入和可重试的源码。

2.1 可重入锁原理

(1) 原理解释

Redisson分布式锁的可重入锁的原理:

  • 使用Redis中Hash的key-value结构,存储线程标识重入次数
  • 每次获取锁的时候,如果线程标识相同,就给Hash的value加1
  • 每次释放锁的时候,并不会删除锁,而是重入次数减1
  • 当所有逻辑执行完,如果重入次数为0,则删除锁

逻辑流程图

在这里插入图片描述

(2) 源码分析

① 尝试获取锁
	// 尝试获取锁,无参表示不等待立即返回
    boolean b = lock.tryLock();

org.redisson.RedissonLock

	@Override
    public boolean tryLock() {
        return get(tryLockAsync());
    }
    @Override
    public RFuture<Boolean> tryLockAsync() {
        return tryLockAsync(Thread.currentThread().getId());
    }
    @Override
    public RFuture<Boolean> tryLockAsync(long threadId) {
        // tryAcquireOnceAsync有三个参数 
        // long waitTime 锁重试的最大等待时间
        // long leaseTime 锁自动释放时间
        // TimeUnit unit 时间单位 
        //  long threadId 线程id
        return tryAcquireOnceAsync(-1, -1, null, threadId);
    }

注意,下面的代码我们只需要关注有注释的部分,剩余的部分会在后面讲到。

    // 根据tryAcquireOnceAsync命名可知,这里获取锁只会获取一次,不会进行重试
	private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        // 判断leaseTime是否为-1(leaseTime为锁自动释放的时间,如果不传默认为-1)
        if (leaseTime != -1) {
            return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        }
        // tryLockInnerAsync是尝试获取锁的方法
        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;
    }

tryLockInnerAsync()方法是获取锁的主要实现,我们可以看到,其实都是固定的Lua脚本逻辑。

    // 尝试获取锁的逻辑(重点)
	<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        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脚本摘出来分析一下逻辑:

    -- 判断锁是否存在
    if (redis.call('exists', KEYS[1]) == 0) then 
        -- 等于0表示不存在,创建锁,设置过期时间
        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 
        -- 等于1表示是自己的锁,重试次数加1,设置过期时间
        redis.call('hincrby', KEYS[1], ARGV[2], 1); 
        redis.call('pexpire', KEYS[1], ARGV[1]); 
        return nil; 
    "end; 
    return redis.call('pttl', KEYS[1]);
② 释放锁
    // 释放锁    
	lock.unlock();

org.redisson.RedissonLock

    @Override
    public void unlock() {
        try {
            get(unlockAsync(Thread.currentThread().getId()));
        } catch (RedisException e) {
            if (e.getCause() instanceof IllegalMonitorStateException) {
                throw (IllegalMonitorStateException) e.getCause();
            } else {
                throw e;
            }
        }

同样地,我们只需要跟着有注释的代码往下看。

    @Override
    public RFuture<Void> unlockAsync(long threadId) {
        RPromise<Void> result = new RedissonPromise<Void>();
        // unlockInnerAsync 是释放锁的方法
        RFuture<Boolean> future = unlockInnerAsync(threadId);
		
        future.onComplete((opStatus, e) -> {
            cancelExpirationRenewal(threadId);

            if (e != null) {
                result.tryFailure(e);
                return;
            }

            if (opStatus == null) {
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + threadId);
                result.tryFailure(cause);
                return;
            }

            result.trySuccess(null);
        });

        return result;
    }

unlockInnerAsync()方法是释放锁的主要逻辑。我们可以看到,其实也都是固定的Lua脚本逻辑。

    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "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.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
    }

我们把这段Lua脚本摘出来分析一下逻辑:

    -- 判断锁是否存在
	if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
    	-- 等于0表示不存在,直接返回
    	return nil;
    end; 
	-- 锁重入次数减1
    local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); 
    if (counter > 0) then 
    	-- 如果锁重入次数大于0,则设置过期时间 
        redis.call('pexpire', KEYS[1], ARGV[2]);
        return 0; 
    else
    	-- 如果锁重入次数等于0,则删除锁,并发布通知
        redis.call('del', KEYS[1]);
        redis.call('publish', KEYS[2], ARGV[1]);
        return 1; 
    end; 
    return nil;

2.2 锁重试和watch dog原理

(1) 原理解释

Redisson分布式锁的锁重试和watch dog的原理:

获取锁

  • 首先尝试根据线程标识获取锁
  • 判断锁的ttl是否为null(为null表示成功获取到锁,不为null表示获取锁失败)
    • 获取锁成功,判断锁自动释放的时间是否未设置(-1)
      • 未设置,开启watch dog,进行自动续约
      • 设置则使用设置的时间
    • 获取锁失败,判断剩余等待时间是否大于0
      • 大于0,订阅并等待释放锁的信号,并进行重试逻辑
      • 小于等于0,则获取锁失败,返回

释放锁

  • 首先尝试根据线程标识释放锁
    • 释放锁成功,发送释放锁的消息,并关闭watch dog自动续约
    • 释放锁失败,记录异常,返回

在这里插入图片描述

(2) 源码分析

① 重试获取锁
    // 尝试获取锁,并设定获取锁的最大等待时长为1s
    boolean b = lock.tryLock(1, TimeUnit.SECONDS);

org.redisson.RedissonLock

    @Override
    public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        return tryLock(waitTime, -1, unit);
    }
    @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime); // 将等待时间转化为ms
        long current = System.currentTimeMillis(); // 当前时间 
        long threadId = Thread.currentThread().getId(); // 线程ID
        // tryAcquire是尝试获取锁的方法,ttl表示锁的剩余有效期 
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            // 如果ttl为null,则表示获取锁成功,返回true
            return true; 
        }
        // 否则获取锁失败,需要重试
        
        // 这里减去获取锁消耗的时间,防止获取锁时间过长超过了等待时间
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            // 如果等待时间小于0,则返回false,获取锁失败
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
        // 这里等待时间还有剩余,可以进行继续尝试获取锁
        current = System.currentTimeMillis(); // 当前时间
        // 订阅释放锁的信号(如果锁被释放,则这里会收到通知,通知的逻辑在释放锁的lua脚本中执行,publish操作)
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // await方法为等待,等待时间为time
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            // 如果等待time时间,还没有收到释放锁通知
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId); // 取消释放锁信号的订阅
                    }
                });
            }
            acquireFailed(waitTime, unit, threadId);
            return false; // 获取锁失败,返回false
        }

        // 这里已经获取到释放锁的通知
        
        try {
            // 减去等待消耗的时间
            time -= System.currentTimeMillis() - current; 
            if (time <= 0) {
                // 如果等待时间小于0,则返回false,获取锁失败
                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();
                // 以信号量机制的方式尝试获取
                // ttl为锁剩余时间,time为等待时间
                if (ttl >= 0 && ttl < time) {
                    // 如果ttl < time,只会等待ttl时间 
                    subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    // 否则等待time
                    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 Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        // 这里的get其实就是阻塞等待返回锁的剩余有效期(如果存在,不存在返回null)
        return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
    }
    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,表示这里使用异步的方式去执行获取锁的逻辑
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                 // 如果不设置默认时间,则在这里给定默认值,默认值为30s
                                 // private long lockWatchdogTimeout = 30 * 1000;
 							     commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        // 当RFuture完成以后
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
            // ttlRemaining表示剩余有效期,e表示异常
            if (e != null) {
                return;
            }

            // ttlRemaining == null,表示获取锁成功 
            if (ttlRemaining == null) {
                // 解决有效期问题,即过期时间续约
                scheduleExpirationRenewal(threadId);
            }
        });
        return ttlRemainingFuture;
    }

    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        // 将leaseTime转化为ms赋值给内部的成员变量
        internalLockLeaseTime = unit.toMillis(leaseTime);

        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; " +
                        // 这里返回的是 锁的剩余有效期,单位为ms
                        "return redis.call('pttl', KEYS[1]);",
                Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }
② watch dog自动续约

定时更新有效期的方法是在获取锁成功之后调用的,方法名为scheduleExpirationRenewal

	private static final ConcurrentMap<String, ExpirationEntry> EXPIRATION_RENEWAL_MAP = new ConcurrentHashMap<>();

	// name是锁的名称
	public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
        ...
        this.id = commandExecutor.getConnectionManager().getId(); // 当前连接id
		...
        this.entryName = id + ":" + name;
		...
    }

	protected String getEntryName() {
        return entryName;
    }

	// 解决有效期问题,即过期时间续约
	private void scheduleExpirationRenewal(long threadId) {
        ExpirationEntry entry = new ExpirationEntry();
        // EXPIRATION_RENEWAL_MAP是一个ConcurrentMap,key可以理解为就是锁的名称,entry就是锁的实例
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        if (oldEntry != null) {
            // 同一个线程重入的情况
            oldEntry.addThreadId(threadId);
        } else {
            // 线程第一次来
            entry.addThreadId(threadId);
            // 更新过期时间
            renewExpiration();
        }
    }
    // 更新过期时间
	private void renewExpiration() {
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ee == null) {
            return;
        }
		// 定时任务,有三个参数
        // TimerTask 定时任务
        // long delay 延迟时间
        // TimeUnit unit 时间单位 
        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                // 获取ExpirationEntry
                ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
                if (ent == null) {
                    return;
                }
                // 取出线程ID
                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) {
                        // 重新执行自己的逻辑,这样的话,如果锁存在,就会自动续约,10s执行一次
                        renewExpiration();
                    }
                });
            }
            // internalLockLeaseTime是内部锁释放时间(尝试获取锁的逻辑中赋值,是一个成员变量,即wtach dog默认时间30s)
            // 这里表示该定时任务10s后执行
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

        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));
    }

那么什么时候停止定时更新呢?

当然是在释放锁的过程中,我们回过来看释放锁的剩余代码:

    @Override
    public RFuture<Void> unlockAsync(long threadId) {
        RPromise<Void> result = new RedissonPromise<Void>();
        // unlockInnerAsync 是释放锁的方法
        RFuture<Boolean> future = unlockInnerAsync(threadId);
		
        // 释放锁完成后 
        future.onComplete((opStatus, e) -> {
            // 取消到期更新任务
            cancelExpirationRenewal(threadId);

            if (e != null) {
                result.tryFailure(e);
                return;
            }

            if (opStatus == null) {
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + threadId);
                result.tryFailure(cause);
                return;
            }

            result.trySuccess(null);
        });

        return result;
    }
    void cancelExpirationRenewal(Long threadId) {
        // 从map中获取到当前锁的定时任务
        ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (task == null) {
            return;
        }
        
        if (threadId != null) {
            // 从任务中移除线程id
            task.removeThreadId(threadId);
        }

        if (threadId == null || task.hasNoThreads()) {
            Timeout timeout = task.getTimeout();
            if (timeout != null) {
                // 将定时任务取消
                timeout.cancel();
            }
            // 最后将entry移除
            EXPIRATION_RENEWAL_MAP.remove(getEntryName());
        }
    }

这样,在锁被释放锁,所有的定时任务、订阅都被取消了。

文章来源:https://blog.csdn.net/qq_52805594/article/details/135698708
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。