Redisson源码-多线程之首个获取锁的线程加解锁流程

这篇具有很好参考价值的文章主要介绍了Redisson源码-多线程之首个获取锁的线程加解锁流程。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Redisson源码-多线程之首个获取锁的线程加解锁流程

简介
当有多个线程同时去获取同一把锁时,第一个获取到锁的线程会进行加解锁,其他线程需订阅消息并等待锁释放。

以下源码分析基于redisson-3.17.6版本,不同版本源码会有些许不同需注意。

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

当我们调用Redisson.lock()并且不设置锁时间时,我们进入RedissonLock的lock方法。

    public void lock() {
        try {
        	// -1L为锁时间,表示不限时
            this.lock(-1L, (TimeUnit)null, false);
        } catch (InterruptedException var2) {
            throw new IllegalStateException();
        }
    }
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
		// 获取当前线程id
        long threadId = Thread.currentThread().getId();
        // 尝试获取锁
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // lock acquired
        // 获取到锁直接返回
        if (ttl == null) {
            return;
        }
		// 订阅锁消息:当锁被释放的时候,会通过publish发布一条消息,通知其它等待这个锁的线程,锁已经释放。
        CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
        pubSub.timeout(future);
        RedissonLockEntry entry;
        if (interruptibly) {
            entry = commandExecutor.getInterrupted(future);
        } else {
            entry = commandExecutor.get(future);
        }

        try {
        	// 不停尝试获取锁
            while (true) {
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message
                // 如果锁的过期时间>=0
                if (ttl >= 0) {
                    try {
                    	// 等待超时时间过去
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        if (interruptibly) {
                            throw e;
                        }
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                 // 锁过期时间<0,代表锁的超时时间未设置
                } else {
                    if (interruptibly) {
                        entry.getLatch().acquire();
                    } else {
                    	// 无限等待直至获取锁
                        entry.getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
            unsubscribe(entry, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

我们按照代码逻辑先看一下尝试获取锁的代码

private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
    }
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        RFuture<Long> ttlRemainingFuture;
        if (leaseTime > 0) {
        	// 指定了超时时间
            ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        } else {
        	// 未指定超时时间
            ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                    TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        }
        CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
            // lock acquired
            // ttlRemaining为null, 本质就是加锁的LUA脚本中返回nil,表示获取锁成功
            if (ttlRemaining == null) {
                if (leaseTime > 0) {
                	// 如果设置了超时时间,则更新internalLockLeaseTime为指定的超时时间,并且不会启动看门狗
                    internalLockLeaseTime = unit.toMillis(leaseTime);
                } else {
                	// 自动续期实现,开启看门狗机制
                    scheduleExpirationRenewal(threadId);
                }
            }
            return ttlRemaining;
        });
        return new CompletableFutureWrapper<>(f);
    }

tryLockInnerAsync方法里的代码是加锁的核心代码之一:

    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        return evalWriteAsync(getRawName(), 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(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
    }

先介绍一下lua脚本的中的参数:
KEYS[1]:锁住的对象(锁住的key),相当于下图的HASH
ARGV[1]:锁的过期时间
ARGV[2]:UUID+当前线程id,相当于下图的key
nil:相当于null

初次获取锁:
Redisson源码-多线程之首个获取锁的线程加解锁流程
锁重入:
Redisson源码-多线程之首个获取锁的线程加解锁流程

接下来我们详细看一下lua脚本的逻辑:

 		 // 通过exists指令判断需要加锁的key是否存在,如果不存在,说明还没被加锁,可以直接进行加锁
		"if (redis.call('exists', KEYS[1]) == 0) then " +
        // 通过hincrby指令往redis中插入一个哈希结构的数据,key[1]=加锁key   ARGV[2]=uuid+当前线程ID  1=锁的重入次数
        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
        // 通过pexpire指令设置锁的过期时间:ARGV[1]=锁的过期时间
        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
        // 返回nil, 表示加锁成功:nil=null
        "return nil; " +
        "end; " +

        // 如下是可重入锁的逻辑
        // 通过hexists指令判断当前的锁是不是自己的,只有是自己的锁时,才支持可重入
        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
        // 通过hincrby指令更新hash结构的数据(锁结构数据),将value对应的可重入次数加一
        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
        // 通过pexpire指令设置锁的过期时间
        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
        // 返回nil, 表示加锁成功
        "return nil; " +
        "end; " +

        // 如果当前已经有人获取了锁,并且这个锁不是自己的,那么将会执行pttl指令,返回当前锁剩余的过期时间
        "return redis.call('pttl', KEYS[1]);"

从上面我们可以看出,redisson加锁的本质是通过执行lua脚本,返回nil(相当于null)或锁的剩余过期时间。如果返回并且未设置过期时间则开启看门狗机制。

接下来我们看下开启看门狗机制的代码

protected void scheduleExpirationRenewal(long threadId) {
        ExpirationEntry entry = new ExpirationEntry();
        // 将当前锁的名称和 ExpirationEntry 对象放入 EXPIRATION_RENEWAL_MAP中,并返回之前关联的对象(如果存在)。
        // 本质就是检查当前锁是否已开启自动续期。
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        // 已开启
        if (oldEntry != null) {
            oldEntry.addThreadId(threadId);
        // 未开启
        } else {
            entry.addThreadId(threadId);
            try {
            	// 执行自动续期机制
                renewExpiration();
            } finally {
                if (Thread.currentThread().isInterrupted()) {
                	// 如果线程中断则解除自动续期机制避免死锁
                    cancelExpirationRenewal(threadId);
                }
            }
        }
    }
private void renewExpiration() {
		// 获取当前锁名称的 ExpirationEntry 对象。
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ee == null) {
            return;
        }
        // 构建Timeout任务去执行锁续期,本质是调用了netty框架中的newTimeout方法,相当于一个延迟定时任务。
        // 相当于每隔 过期时间/3 (默认10秒)毫秒,递归调用renewExpiration方法去执行锁续期直至锁被释放。
        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;
                }
                // 获取 ExpirationEntry 对象中的第一个线程 ID。
                Long threadId = ent.getFirstThreadId();
                if (threadId == null) {
                    return;
                }
                // 锁续期
                CompletionStage<Boolean> future = renewExpirationAsync(threadId);
                future.whenComplete((res, e) -> {
                    if (e != null) {
                        log.error("Can't update lock " + getRawName() + " expiration", e);
                        EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                        return;
                    }
                    // 续期成功
                    if (res) {
                        // reschedule itself
                        // 递归调用自身,不断续期
                        renewExpiration();
                    // 续期失败,表示锁被释放
                    } else {
                    	// 取消定时任务等操作
                        cancelExpirationRenewal(null);
                    }
                });
            }
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
        
        ee.setTimeout(task);
    }

下面我们看一下锁续期的核心代码:

protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
        return evalWriteAsync(getRawName(), 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(getRawName()),
                internalLockLeaseTime, getLockName(threadId));
    }

该lua脚本的中的参数与上面相同:
KEYS[1]:锁住的对象(锁住的key)
ARGV[1]:锁的过期时间
ARGV[2]:UUID+当前线程id

下面我们看一下这段lua脚本的逻辑:

	// 通过hexists指令判断当前的锁是不是自己的
	"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
	// 如果是的话通过pexpire指令设置锁的过期时间
	"redis.call('pexpire', KEYS[1], ARGV[1]); " +
	"return 1; " +
	"end; " +
	// 否则返回0
	"return 0;",

整理上文可以看出看门狗机制简单来说就是每隔 过期时间/3 毫秒去执行lua脚本,若锁未被释放则刷新其过期时间,直至锁被释放为止。

接下来我们再看下解锁的逻辑:

@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;
            }
        }
public RFuture<Void> unlockAsync(long threadId) {
		// 释放锁的核心代码
        RFuture<Boolean> future = unlockInnerAsync(threadId);

        CompletionStage<Void> f = future.handle((opStatus, e) -> {
        	// 取消看门狗机制(就是取消上文的定时任务等操作)
            cancelExpirationRenewal(threadId);

            if (e != null) {
                throw new CompletionException(e);
            }
            if (opStatus == null) {
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + threadId);
                throw new CompletionException(cause);
            }

            return null;
        });

        return new CompletableFutureWrapper<>(f);
    }

下面我们看一下解锁的核心代码:

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return evalWriteAsync(getRawName(), 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(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
    }

先说明一下该lua脚本的中的参数:
KEYS[1]:锁住的对象(锁住的key)
KEYS[2]:监听该锁的频道
ARGV[1]:解锁消息
ARGV[2]:锁的过期时间
ARGV[3]:UUID+当前线程id

下面我们看下这段lua脚本的逻辑:

	// 通过hexists指令判断当前的锁是不是自己的
	"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
	// 不是自己的则返回nil(相当于null)
	"return nil;" +
	"end; " +
	// 则使用 hincrby 指令将字段锁的可重入次数减去 1,即减少持有锁的线程数。
	"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
	// 如果结果 > 0,代表解锁成功,但是锁仍然存在
	"if (counter > 0) then " +
	// 通过pexpire指令设置锁的过期时间
	"redis.call('pexpire', KEYS[1], ARGV[2]); " +
	// 返回0
	"return 0; " +
	"else " +
	// 如果结果 < 0 ,代表持有锁的线程数为0,这时需要完全释放锁,通过del指令删除指定key的锁
	"redis.call('del', KEYS[1]); " +
	// 通过publish指令向订阅该锁的频道发送解锁消息
	"redis.call('publish', KEYS[2], ARGV[1]); " +
	// 返回1
	"return 1; " +
	"end; " +
	"return nil;",

以上我们解释说明了单个线程在不等待锁的情况下,直接获取锁,对锁进行续期和解锁的代码逻辑,可以看出加解锁本质上都是通过lua脚本去执行,当有多个线程同时去获取锁时,第一个获取到锁的线程会按照此逻辑执行,其他线程需订阅消息并等待锁释放。文章来源地址https://www.toymoban.com/news/detail-507339.html

到了这里,关于Redisson源码-多线程之首个获取锁的线程加解锁流程的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 多线程Synchronized锁的使用与线程之间的通讯

    多线程同时对同一个全局变量做写操作,可能会受到其他线程的干扰,就会发生线程安全问题。 Java中的全局变量是存放在堆内存中的,而堆内容对于所有线程来说是共享的。 比如下面一个简单的代码案例: 代码比较简单,我们看下面控制台的打印: 可以看到两个线程之间

    2024年02月04日
    浏览(41)
  • cpp多线程(二)——对线程的控制和锁的概念

    这篇文章是笔者学习cpp多线程操作的第二篇笔记,没有阅读过第一篇的读者可以移步此处: Cpp多线程(一)-CSDN博客 如果读者发现我的文章里有问题,欢迎交流哈!   一、如何控制线程呢? c++11在std::this_thread名称空间(显然,这是一个嵌套在大名称空间里的小名称空间)内

    2024年01月20日
    浏览(33)
  • 多线程锁的升级原理是什么

            在 Java 中,锁共有 4 种状态,级别从低到高依次为:无状态锁,偏向锁,轻量级锁和重量级锁状态,这几个状态会随着竞争情况逐渐升级。锁可以升级但不能降级。         如下图所示         多线程锁的升级过程主要指的是锁从偏向锁状态升级为轻量级

    2024年02月06日
    浏览(36)
  • C# 中多线程锁的使用经验

    C# 中多线程锁的使用经验:全局锁,实例锁         private static object _exeLock = new object();        static 静态的是全应用程序的资源。如果在一个类里定义了这样一个锁,你在调用使用这个类的时候,是NEW了一个对象,并把这个对象给了一个静态全局变量中保存。这时这个锁

    2024年03月14日
    浏览(55)
  • openxr runtime Monado 源码解析 源码分析:整体介绍 模块架构 模块作用 进程 线程模型 整体流程

    monado系列文章索引汇总: openxr runtime Monado 源码解析 源码分析:源码编译 准备工作说明 hello_xr解读 openxr runtime Monado 源码解析 源码分析:整体介绍 模块架构 模块作用 进程 线程模型 整体流程 openxr runtime Monado 源码解析 源码分析:CreateInstance流程(设备系统和合成器系统)C

    2024年02月11日
    浏览(83)
  • 基于springboot实现多线程抢锁的demo

    1、本代码基于定时调度和异步执行同时处理,如果只加异步处理,会导致当前任务未执行完,下个任务到点也不会触发执行 2、日志信息: 3、使用 jstack 88023 |grep SimpleAsyncTaskExecutor-62 命令可以查看相关线程是否存在,可以看到,相关线程任务执行完成后,会自动消失 4、如果

    2024年02月09日
    浏览(38)
  • 【Java | 多线程】可重入锁的概念以及示例

    可重入锁(又名递归锁)是一种特殊类型的锁,它允许 同一个线程在获取锁后再次进入该锁保护的代码块或方法,而不需要重新获取锁 。 说白了,可重入锁的特点就是同一个线程可以多次获取同一个锁,而不会因为之前已经获取过锁而阻塞。 可重入锁的一个优点是可以一定

    2024年04月24日
    浏览(38)
  • 【多线程面试题二十二】、 说说你对读写锁的了解

    文章底部有个人公众号: 热爱技术的小郑 。主要分享开发知识、学习资料、毕业设计指导等。有兴趣的可以关注一下。为何分享? 踩过的坑没必要让别人在再踩,自己复盘也能加深记忆。利己利人、所谓双赢。 面试官:说说你对读写锁的了解 参考答案: 与传统锁不同的是

    2024年02月06日
    浏览(39)
  • 【Linux从入门到精通】线程互斥与互斥锁的使用与原理详解

      上篇文章我们对线程 | 线程介绍线程控制介绍后,本篇文章将会对多线程中的线程互斥与互斥锁的概念进行详解。同时结合实际例子解释了可重入与不被重入函数、临界资源与临界区和原子性的概念。希望本篇文章会对你有所帮助。 文章目录 引入 一、重入与临界 1、1 可重

    2024年02月07日
    浏览(44)
  • (五)、Redisson锁机制源码分析

    1、锁的互斥性 也就是说,在任意时刻,只能有一个客户端能获取到锁,不能同时有两个或多个客户端获取到锁。 简单来说,就比如上厕所,一个厕所只有一个坑位,只能一个人上,不能同时两个人或多个人上。 2、锁的同一性 也就是说,锁只能被持有该锁的客户端进行删除

    2024年02月11日
    浏览(38)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包