百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术教程 > 正文

【redis实战五】Redisson锁机制源码分析

mhr18 2025-03-25 15:27 20 浏览 0 评论

1、了解分布式锁的特性

  • 1、锁的互斥性
    • 也就是说,在任意时刻,只能有一个客户端能获取到锁,不能同时有两个或多个客户端获取到锁。
    • 简单来说,就比如上厕所,一个厕所只有一个坑位,只能一个人上,不能同时两个人或多个人上。
  • 2、锁的同一性
    • 也就是说,锁只能被持有该锁的客户端进行删除(释放锁),不能由其他客户端删除。
    • 简单俩说,就是谁加的锁,就只能谁来解锁。也就是解铃还须系铃人。
  • 3、锁的可重入性
    • 也就是说,持有某个锁的客户端,可以继续对该锁进行加锁,实现锁的续租。
    • 简单来说,就是你上厕所的按时间收费的,时间快到了会按照时间给你续租,而会给你价钱。
    • 而Redisson则会增大的你的续租次数,也就是可重入次数。但绝不收费,因为Redis是开源的嘛。
  • 4、锁的容错性
    • 锁超过了最大续租时间后,会自动释放锁,其他客户端会继续获得该锁,从而防止死锁的发生。
    • 简单来说,比如你上个厕所上了五小时,厕管员觉得不对劲,就来测试,发现你悄悄逃票了,此时测试会自动变成解锁状态,其他人就可以去上了,只是厕管员血亏5块大洋。

2、带着几个特性去看Redisson源码

先回顾一下Redisson加解锁代码如何写的

public TestEntity getById2(Long id){
    RLock lock = redissonClient.getLock("demo2_lock");
    lock.lock(20, TimeUnit.SECONDS);
    index++;
    log.info("current index is : {}", index);
    TestEntity testEntity = new TestEntity(new Random().nextLong(), UUID.randomUUID().toString(), new Random().nextInt(20) + 10);
    log.info("模拟查询数据库:{}", testEntity);
    lock.unlock();
    return testEntity;
}

2.1、关注Redisson.getLock()方法

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

其实就是创建一个RedissonLock对象, 所以加锁的逻辑就在RedissonLock.lock()中,解锁的逻辑就在RedissonLock.unlock()。

2.2、关注RedissonLock.lock()方法

// RedissonLock.lock()的方法体
public void lock(long leaseTime, TimeUnit unit) {
    try {
        // 调用了lock的重载方法
        lock(leaseTime, unit, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

关注lock的重载方法

// leaseTime表示最大续时间,unit表示续约时间单位,interruptibly表示是否可以中断
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    // 获取当前线程的线程ID
    long threadId = Thread.currentThread().getId();
    // 尝试获取锁,结果为null表示此时没有客户端占用锁,绝不矫情,直接拿到锁就返回。
    // 结果ttl>0的话,表示此时已经有了其他不识好歹的客户端暂用了锁,那么就只能绝望的等待了    
    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return;
    }

    // 等待时订阅一个渠道,如果锁被其他客户端释放了,会通过发布订阅模式在publish上发一个消息,表示锁已经释放了
    CompletableFuture 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);
            // 如果TTL为空了,表示获取到了锁,那还等什么,长驱直入就是。
            if (ttl == null) {
                // 结束循环等待
                break;
            }

            // 如果ttl还是大于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 {
        // 最后,不管如何,我无论如何都要去取消订阅这个publish的消息,因为这会浪费我的精力,这已经是我最后的坚持了。
        // 其实是释放资源
        unsubscribe(entry, threadId);
    }
//        get(lockAsync(leaseTime, unit));
}

关注tryAcquire加锁方法

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

该方法调用了tryAcquireAsync来实现的,所以我们关注tryAcquireAsync方法,继续跟进。

关注tryAcquireAsync加锁方法

private  RFuture tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    RFuture ttlRemainingFuture;
    // 首先判断租约时间是否大于0
    if (leaseTime > 0) {
        // 大于零,调用tryLockInnerAsync获取锁
        ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    } else {
        // 否则,使用默认的租约时间 追溯下去发现private long lockWatchdogTimeout = 30 * 1000;  也就是30s的租约时间
        ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    }
    // 
    CompletionStage f = ttlRemainingFuture.thenApply(ttlRemaining -> {
        // lock acquired
        // 结果为空,如果leaseTime大于哦,更新internalLockLeaseTime为指定的超时时间,并且不会启动看门狗(watch dog)
        if (ttlRemaining == null) {
            if (leaseTime > 0) {
                internalLockLeaseTime = unit.toMillis(leaseTime);
            } else {
                // 使用定时任务,自动续约(使用看门狗(watch dog))
                scheduleExpirationRenewal(threadId);
            }
        }
        return ttlRemaining;
    });
    return new CompletableFutureWrapper<>(f);
}

可以看到,加锁最终会调用tryLockInnerAsync进行加锁,而续约会使用scheduleExpirationRenewal进行续约。

关注tryLockInnerAsync实现真正的加锁逻辑

 RFuture tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand 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脚本(整个lua脚本保障原子性),我们将脚本内容复制出来,详细解释一下。

-- KEYS[1] 加锁的对象(也就是我们传入的的锁名称)
-- ARGV[1] 表示锁的过期时间
-- ARGV[2]:UUID+当前线程id
-- 如果锁不存在。 == 0表示不存在 == 1表示存在
if (redis.call('exists', KEYS[1]) == 0) then
    -- 对我自己的锁执行一个incrby(自增,表示锁的可重入次数)操作
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
    -- 对key设置一个过期时间(过期时间就是保证锁的容错性)
    redis.call('pexpire', KEYS[1], ARGV[1]);
    -- 返回nil, 相当于null, 表示获取锁成功
    return nil;
end ;
-- 继续判断锁名成+UUID+当前线程id是否存在,其实就是判断我自己有没有已经拿到锁(保证锁的可重入性)
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
    -- 自己已经持有锁,执行一个incrby(自增,表示锁的可重入次数)操作
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
    -- 重新设置过期时间
    redis.call('pexpire', KEYS[1], ARGV[1]);
    -- 返回nil, 相当于null, 表示获取锁成功
    return nil;
end ;
-- 都不是,表示已经有其他客户端获取到了锁,此时返回key的过期时间,也就是别人释放锁的时间(但其他客户端可能出现续约,存在会等待更久的可能)
return redis.call('pttl', KEYS[1]);

整个lua脚本保障原子性,从而只会有一个客户端能够获取到锁,这样就保证了锁的互斥性。

打一个断点看获取到的锁信息

hash表中的第一个值表示UUID+线程ID,这二个值表示锁的重入次数,如果锁被多次获取,那么这个值就是大于1。

关注scheduleExpirationRenewal实现自动续约

protected void scheduleExpirationRenewal(long threadId) {
    ExpirationEntry entry = new ExpirationEntry();
    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);
            }
        }
    }
}

关注renewExpiration()方法

private void renewExpiration() {
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) {
        return;
    }
    // 创建一个定时任务去实现自动续约
    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;
            }
            // 锁续期
            CompletionStage 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); // internalLockLeaseTime / 3表示每隔锁时间的三分之一,去续约一次
    
    ee.setTimeout(task);
}

关注renewExpirationAsync方法

 protected CompletionStage 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] 加锁的对象(也就是我们传入的的锁名称)
-- ARGV[1] 表示锁的过期时间
-- ARGV[2]:UUID+当前线程id
-- 使用hexists判断锁是不是自己持有的, == 1表示是自己持有,== 0 表示被其他客户端持有
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
    -- 重新设置过期时间
    redis.call('pexpire', KEYS[1], ARGV[1]);
    -- 返回1 表示续约成功
    return 1;
end ;
-- 返回0 表示续约失败,也意味着锁已经被释放或者被其他客户端获取了
return 0;

所以续约的逻辑就是,启动一个定时任务,每隔续约时间的三分之一次就执行一次。尝试去续约,续约成功则会一直递归续约下去。续约失败表示锁已被释放,则停止续约任务。

而续约的操作就是,判断是否是自己持有锁,是的话就重新设置过期时间,并且返回1表示续约成功,否则返回0表示续约失败。

2.3、关注RedissonLock.unlock()方法

@Override
public void unlock() {
    try {
        // 其实就是调用了unlockAsync进行解锁
        get(unlockAsync(Thread.currentThread().getId()));
    } catch (RedisException e) {
        if (e.getCause() instanceof IllegalMonitorStateException) {
            throw (IllegalMonitorStateException) e.getCause();
        } else {
            throw e;
        }
    }
    
//        Future future = unlockAsync();
//        future.awaitUninterruptibly();
//        if (future.isSuccess()) {
//            return;
//        }
//        if (future.cause() instanceof IllegalMonitorStateException) {
//            throw (IllegalMonitorStateException)future.cause();
//        }
//        throw commandExecutor.convertException(future);
}

我们可以看到,会使用unlockAsync方法进行解锁,并且在这里传入了当前的线程ID。

关注unlockAsync方法

@Override
public RFuture unlockAsync(long threadId) {
    // 调用unlockInnerAsync实现异步解锁
    RFuture future = unlockInnerAsync(threadId);

    // 释放之后再处理一些事情
    CompletionStage f = future.handle((opStatus, e) -> {
        // 取消(停止)续约任务,这里也会停止watch dog
        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);
}

关注解锁的核心逻辑unlockInnerAsync方法

protected RFuture 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] 加锁的对象(也就是我们传入的的锁名称)
-- KEYS[2] 监听该锁的频道 也就是publish要发送锁被释放的频道,用于在锁释放时通知其他客户端可以重新获取锁了
-- ARGV[1]:解锁消息
-- ARGV[2] 表示锁的过期时间
-- ARGV[3]:UUID+当前线程id
-- 先判断自己的锁是不是已经释放了 ==0 表示key不存在了,也就是锁被释放了
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
    -- 返回nil,也就是null, 表示释放锁成功
    return nil;
end ;
-- 对锁的重入次数减一  因为重入一次counter会+1,所以释放时每次也只能-1,跟重入次数匹配
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
-- 如果重入次数仍然大于0,续约过期时间
if (counter > 0) then
    redis.call('pexpire', KEYS[1], ARGV[2]);
    -- 返回解说失败
    return 0;
else
    -- 表示重入次数已经为0了,删除锁的key
    redis.call('del', KEYS[1]);
    -- 使用publish发布一个消息,其他订阅了的客户端收到消息,就说明解锁成功了饿、然后可以重新获取锁了
    redis.call('publish', KEYS[2], ARGV[1]);
    -- 返回1 表示解锁成功
    return 1;
end ;
return nil;

其实就是在解锁的时候,已经解锁了直接返回成功,可重入次数没有到0,将会解锁失败,直到可重入次数重新减到0后,开始删除锁的key.

并且此时会使用publish发送一个消息在渠道上,订阅者们订阅到了,就说明锁已经被释放了,然后可以从重新获取锁了。

3、小结

Redisson实现分布式锁,就是使用lua脚本保证原子性和互斥性的。每次都判断是不是自己持有锁,才进行操作,这就保证了同一性。

在加锁时使用incrby对key对应的value值进行自增,减锁时自减实现锁的可重入性。

使用redis的超时自动过期来保证锁的容错性,不会一直锁死下去。所以锁的最大续约时间是防止思索的一个有效的方法。

相关推荐

B站收藏视频失效?mybili 收藏夹备份神器完整部署指南

本内容来源于@什么值得买APP,观点仅代表作者本人|作者:羊刀仙很多B站用户都有过类似经历:自己精心收藏的视频突然“消失”,点开一看不是“已被删除”,就是“因UP主设置不可见”。而B站并不会主动通知...

中间件推荐初始化配置

Redis推荐初始化配置bind0.0.0.0protected-modeyesport6379tcp-backlog511timeout300tcp-keepalive300...

Redis中缓存穿透问题与解决方法

缓存穿透问题概述在Redis作为缓存使用时,缓存穿透是常见问题。正常查询流程是先从Redis缓存获取数据,若有则直接使用;若没有则去数据库查询,查到后存入缓存。但当请求的数据在缓存和数据库中都...

后端开发必看!Redis 哨兵机制如何保障系统高可用?

你是否曾在项目中遇到过Redis主服务器突然宕机,导致整个业务系统出现数据读取异常、响应延迟甚至服务中断的情况?面对这样的突发状况,作为互联网大厂的后端开发人员,如何快速恢复服务、保障系统的高可用...

Redis合集-大Key处理建议

以下是Redis大Key问题的全流程解决方案,涵盖检测、处理、优化及预防策略,结合代码示例和最佳实践:一、大Key的定义与风险1.大Key判定标准数据类型大Key阈值风险场景S...

深入解析跳跃表:Redis里的&quot;老六&quot;数据结构,专治各种不服

大家好,我是你们的码农段子手,今天要给大家讲一个Redis世界里最会"跳科目三"的数据结构——跳跃表(SkipList)。这货表面上是个青铜,实际上是个王者,连红黑树见了都要喊声大哥。...

Redis 中 AOF 持久化技术原理全解析,看完你就懂了!

你在使用Redis的过程中,有没有担心过数据丢失的问题?尤其是在服务器突然宕机、意外断电等情况发生时,那些还没来得及持久化的数据,是不是让你夜不能寐?别担心,Redis的AOF持久化技术就是...

Redis合集-必备的几款运维工具

Redis在应用Redis时,经常会面临的运维工作,包括Redis的运行状态监控,数据迁移,主从集群、切片集群的部署和运维。接下来,从这三个方面,介绍一些工具。先来学习下监控Redis实时...

别再纠结线程池大小 + 线程数量了,没有固定公式的!

我们在百度上能很轻易地搜索到以下线程池设置大小的理论:在一台服务器上我们按照以下设置CPU密集型的程序-核心数+1I/O密集型的程序-核心数*2你不会真的按照这个理论来设置线程池的...

网络编程—IO多路复用详解

假如你想了解IO多路复用,那本文或许可以帮助你本文的最大目的就是想要把select、epoll在执行过程中干了什么叙述出来,所以具体的代码不会涉及,毕竟不同语言的接口有所区别。基础知识IO多路复用涉及...

5分钟学会C/C++多线程编程进程和线程

前言对线程有基本的理解简单的C++面向过程编程能力创造单个简单的线程。创造单个带参数的线程。如何等待线程结束。创造多个线程,并使用互斥量来防止资源抢占。会使用之后,直接跳到“汇总”,复制模板来用就行...

尽情阅读,技术进阶,详解mmap的原理

1.一句话概括mmapmmap的作用,在应用这一层,是让你把文件的某一段,当作内存一样来访问。将文件映射到物理内存,将进程虚拟空间映射到那块内存。这样,进程不仅能像访问内存一样读写文件,多个进程...

C++11多线程知识点总结

一、多线程的基本概念1、进程与线程的区别和联系进程:进程是一个动态的过程,是一个活动的实体。简单来说,一个应用程序的运行就可以被看做是一个进程;线程:是运行中的实际的任务执行者。可以说,进程中包含了多...

微服务高可用的2个关键技巧,你一定用得上

概述上一篇文章讲了一个朋友公司使用SpringCloud架构遇到问题的一个真实案例,虽然不是什么大的技术问题,但如果对一些东西理解的不深刻,还真会犯一些错误。这篇文章我们来聊聊在微服务架构中,到底如...

Java线程间如何共享与传递数据

1、背景在日常SpringBoot应用或者Java应用开发中,使用多线程编程有很多好处,比如可以同时处理多个任务,提高程序的并发性;可以充分利用计算机的多核处理器,使得程序能够更好地利用计算机的资源,...

取消回复欢迎 发表评论: