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

本文涉及的产品
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
云数据库 Tair(兼容Redis),内存型 2GB
全局流量管理 GTM,标准版 1个月
简介: Redis进阶- Redisson分布式锁实现原理及源码解析


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

实例化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();
    }
  • super(commandExecutor, name); 父类name赋值,后续通过getName()获取
  • commandExecutor: 执行lua脚本的executor
  • id 是个UUID, 后面被用来当做 和threadId组成 value值,用作判断加锁和释放锁是否是同一个线程的校验。
  • internalLockLeaseTime : 取自 Config#lockWatchdogTimeout,默认30秒,这个参数还有另外一个作用,锁续命的执行周期 internalLockLeaseTime/3 = 10秒

redissonLock.lock()的逻辑

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

redissonLock.lock();

看看都干了啥

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

继续看 lockInterruptibly

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

继续看 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));
    }

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

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

继续看下

tryAcquireAsync(leaseTime, unit, threadId)
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)尝试加锁  待会细看 先把主要的逻辑梳理完
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);

看实现

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

lua 脚本

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

ARGV[1] ---------> internalLockLeaseTime

ARGV[2] ---------> getLockName(threadId) 实现如下

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

这个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]);"

那继续监听时间中的 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();
        }
    }

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

重点看 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));
    }

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

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

KEYS[2] ---------> getChannelName()

ARGV[1] ---------> LockPubSub.unlockMessage

ARGV[2] ---------> internalLockLeaseTime

ARGV[2] ---------> getLockName(threadId)


总结

需要用到续锁功能时,一要记住不要设置锁的过期时间,可以设置成-1.

一旦设了时间,RedissonLock就会认为你需要自己控制锁时间,而放弃执行续锁逻辑。

查看源码, 续锁逻辑需要起定时器。所以要注意这点,并不是所有分布式场景都需要续锁逻辑的。当我们很难判断业务逻辑的执行时间时,不妨开启续锁。

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


相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
打赏
0
1
0
0
99
分享
相关文章
【📕分布式锁通关指南 02】基于Redis实现的分布式锁
本文介绍了从单机锁到分布式锁的演变,重点探讨了使用Redis实现分布式锁的方法。分布式锁用于控制分布式系统中多个实例对共享资源的同步访问,需满足互斥性、可重入性、锁超时防死锁和锁释放正确防误删等特性。文章通过具体示例展示了如何利用Redis的`setnx`命令实现加锁,并分析了简化版分布式锁存在的问题,如锁超时和误删。为了解决这些问题,文中提出了设置锁过期时间和在解锁前验证持有锁的线程身份的优化方案。最后指出,尽管当前设计已解决部分问题,但仍存在进一步优化的空间,将在后续章节继续探讨。
318 128
【📕分布式锁通关指南 02】基于Redis实现的分布式锁
|
6天前
|
Springboot使用Redis实现分布式锁
通过这些步骤和示例,您可以系统地了解如何在Spring Boot中使用Redis实现分布式锁,并在实际项目中应用。希望这些内容对您的学习和工作有所帮助。
130 83
Redis,分布式缓存演化之路
本文介绍了基于Redis的分布式缓存演化,探讨了分布式锁和缓存一致性问题及其解决方案。首先分析了本地缓存和分布式缓存的区别与优劣,接着深入讲解了分布式远程缓存带来的并发、缓存失效(穿透、雪崩、击穿)等问题及应对策略。文章还详细描述了如何使用Redis实现分布式锁,确保高并发场景下的数据一致性和系统稳定性。最后,通过双写模式和失效模式讨论了缓存一致性问题,并提出了多种解决方案,如引入Canal中间件等。希望这些内容能为读者在设计分布式缓存系统时提供有价值的参考。感谢您的阅读!
102 6
Redis,分布式缓存演化之路
【📕分布式锁通关指南 03】通过Lua脚本保证redis操作的原子性
本文介绍了如何通过Lua脚本在Redis中实现分布式锁的原子性操作,避免并发问题。首先讲解了Lua脚本的基本概念及其在Redis中的使用方法,包括通过`eval`指令执行Lua脚本和通过`script load`指令缓存脚本。接着详细展示了如何用Lua脚本实现加锁、解锁及可重入锁的功能,确保同一线程可以多次获取锁而不发生死锁。最后,通过代码示例演示了如何在实际业务中调用这些Lua脚本,确保锁操作的原子性和安全性。
22 6
【SpringCloud Alibaba系列】一文全面解析Zookeeper安装、常用命令、JavaAPI操作、Watch事件监听、分布式锁、集群搭建、核心理论
一文全面解析Zookeeper安装、常用命令、JavaAPI操作、Watch事件监听、分布式锁、集群搭建、核心理论。
【SpringCloud Alibaba系列】一文全面解析Zookeeper安装、常用命令、JavaAPI操作、Watch事件监听、分布式锁、集群搭建、核心理论
|
2月前
|
使用lock4j-redis-template-spring-boot-starter实现redis分布式锁
通过使用 `lock4j-redis-template-spring-boot-starter`,我们可以轻松实现 Redis 分布式锁,从而解决分布式系统中多个实例并发访问共享资源的问题。合理配置和使用分布式锁,可以有效提高系统的稳定性和数据的一致性。希望本文对你在实际项目中使用 Redis 分布式锁有所帮助。
224 5
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?
Redis分布式锁在高并发场景下是重要的技术手段,但其实现过程中常遇到五大深坑:**原子性问题**、**连接耗尽问题**、**锁过期问题**、**锁失效问题**以及**锁分段问题**。这些问题不仅影响系统的稳定性和性能,还可能导致数据不一致。尼恩在实际项目中总结了这些坑,并提供了详细的解决方案,包括使用Lua脚本保证原子性、设置合理的锁过期时间和使用看门狗机制、以及通过锁分段提升性能。这些经验和技巧对面试和实际开发都有很大帮助,值得深入学习和实践。
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?
|
6月前
|
基于Redis的高可用分布式锁——RedLock
这篇文章介绍了基于Redis的高可用分布式锁RedLock的概念、工作流程、获取和释放锁的方法,以及RedLock相比单机锁在高可用性上的优势,同时指出了其在某些特殊场景下的不足,并提到了ZooKeeper作为另一种实现分布式锁的方案。
182 2
基于Redis的高可用分布式锁——RedLock
基于Redis海量数据场景分布式ID架构实践
【11月更文挑战第30天】在现代分布式系统中,生成全局唯一的ID是一个常见且重要的需求。在微服务架构中,各个服务可能需要生成唯一标识符,如用户ID、订单ID等。传统的自增ID已经无法满足在集群环境下保持唯一性的要求,而分布式ID解决方案能够确保即使在多个实例间也能生成全局唯一的标识符。本文将深入探讨如何利用Redis实现分布式ID生成,并通过Java语言展示多个示例,同时分析每个实践方案的优缺点。
106 8
|
3月前
|
Redis分布式锁如何实现 ?
Redis分布式锁通过SETNX指令实现,确保仅在键不存在时设置值。此机制用于控制多个线程对共享资源的访问,避免并发冲突。然而,实际应用中需解决死锁、锁超时、归一化、可重入及阻塞等问题,以确保系统的稳定性和可靠性。解决方案包括设置锁超时、引入Watch Dog机制、使用ThreadLocal绑定加解锁操作、实现计数器支持可重入锁以及采用自旋锁思想处理阻塞请求。
88 16

推荐镜像

更多