分布式锁实现，Redisson篇

墨蓝 2023-01-22 12:55 150阅读 0赞

![c9a22365e97aa79f254ee8a021a04f21.png][]

针对项目中使用的分布式锁进行简单的示例配置以及源码解析，并列举源码中使用到的一些基础知识点，但是没有对redisson中使用到的netty知识进行解析。

本篇主要是对以下几个方面进行了探索

*  Maven配置
 *  RedissonLock简单示例
 *  源码中使用到的Redis命令
 *  源码中使用到的lua脚本语义
 *  源码分析

##     ##

## **\# Maven配置** ##

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>2.2.12</version>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-annotations</artifactId>
        <version>2.6.0</version>
    </dependency>

## **\# RedissonLock简单示例** ##

redission支持4种连接redis方式，分别为单机、主从、Sentinel、Cluster 集群，项目中使用的连接方式是Sentinel。

redis服务器不在本地的同学请注意权限问题。

###     ###

### **\# Sentinel配置** ###

Config config = new Config();
    config.useSentinelServers().addSentinelAddress("127.0.0.1:6479", "127.0.0.1:6489").setMasterName("master").setPassword("password").setDatabase(0);
    RedissonClient redisson = Redisson.create(config);

**\# 简单使用**

RLock lock = redisson.getLock("test_lock");
    try{
        boolean isLock=lock.tryLock();
        if(isLock){
            doBusiness();
        }
    }catch(exception e){
    }finally{
        lock.unlock();
    }

## **\# 源码中使用到的Redis命令** ##

分布式锁主要需要以下redis命令，这里列举一下。在源码分析部分可以继续参照命令的操作含义。

*  EXISTS key :当 key 存在，返回1；若给定的 key 不存在，返回0。
 *  GETSET key value:将给定 key 的值设为 value ，并返回 key 的旧值 (old value)，当 key 存在但不是字符串类型时，返回一个错误，当key不存在时，返回nil。
 *  [GET key:返回 key 所关联的字符串值，如果 key 不存在那么返回 nil。][GET key_ key _ key _ nil]

*  DEL key \[KEY …\]:删除给定的一个或多个 key ,不存在的 key 会被忽略,返回实际删除的key的个数（integer）。

*  [HSET key field value：给一个key 设置一个\{field=value\}的组合值，如果key没有就直接赋值并返回1，如果field已有，那么就更新value的值，并返回0.][GET key_ key _ key _ nil]

*  [HEXISTS key field:当key中存储着field的时候返回1，如果key或者field至少有一个不存在返回0。][GET key_ key _ key _ nil]

*  [HINCRBY key field increment:将存储在key中的哈希（Hash）对象中的指定字段field的值加上增量increment。如果键key不存在，一个保存了哈希对象的新建将被创建。如果字段field不存在，在进行当前操作前，其将被创建，且对应的值被置为0，返回值是增量之后的值][GET key_ key _ key _ nil]

*  [PEXPIRE key milliseconds：设置存活时间，单位是毫秒。expire操作单位是秒。][GET key_ key _ key _ nil]

*  [PUBLISH channel message:向channel post一个message内容的消息，返回接收消息的客户端数。][GET key_ key _ key _ nil]

##     ##

## **\# 源码中使用到的lua脚本语义** ##

Redisson源码中，执行redis命令的是lua脚本，其中主要用到如下几个概念。

*  redis.call() 是执行redis命令.
 *  KEYS\[1\] 是指脚本中第1个参数
 *  ARGV\[1\] 是指脚本中第一个参数的值
 *  返回值中nil与false同一个意思。

需要注意的是，在redis执行lua脚本时，相当于一个redis级别的锁，不能执行其他操作，类似于原子操作，也是redisson实现的一个关键点。

另外，如果lua脚本执行过程中出现了异常或者redis服务器直接宕掉了，执行redis的根据日志回复的命令，会将脚本中已经执行的命令在日志中删除。

## **\# 源码分析** ##

### RLOCK结构 ###

public interface RLock extends Lock, RExpirable {
        void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException;
        boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
        void lock(long leaseTime, TimeUnit unit);
        void forceUnlock();
        boolean isLocked();
        boolean isHeldByCurrentThread();
        int getHoldCount();
        Future<Void> unlockAsync();
        Future<Boolean> tryLockAsync();
        Future<Void> lockAsync();
        Future<Void> lockAsync(long leaseTime, TimeUnit unit);
        Future<Boolean> tryLockAsync(long waitTime, TimeUnit unit);
        Future<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit);
    }

该接口主要继承了Lock接口, 并扩展了部分方法, 比如:boolean tryLock(long waitTime, long leaseTime, TimeUnit unit)新加入的leaseTime主要是用来设置锁的过期时间, 如果超过leaseTime还没有解锁的话, redis就强制解锁. leaseTime的默认时间是30s

### RedissonLock获取锁 tryLock源码 ###

Future<Long> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId) {
           internalLockLeaseTime = unit.toMillis(leaseTime);
           return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
                     "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));
       }

其中：

*  [KEYS\[1\] 表示的是 getName() ，代表的是锁名 test\_lock][GET key_ key _ key _ nil]
 *  [ARGV\[1\] 表示的是 internalLockLeaseTime 默认值是30s][GET key_ key _ key _ nil]
 *  [ARGV\[2\] 表示的是 getLockName(threadId) 代表的是 id:threadId 用锁对象id+线程id， 表示当前访问线程，用于区分不同服务器上的线程。][GET key_ key _ key _ nil]

逐句分析：

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('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(‘exists’, KEYS\[1\]) == 0) 如果锁名称不存在

then redis.call(‘hset’, KEYS\[1\], ARGV\[2\],1) 则向redis中添加一个key为test\_lock的set，并且向set中添加一个field为线程id，值=1的键值对，表示此线程的重入次数为1

redis.call(‘pexpire’, KEYS\[1\], ARGV\[1\]) 设置set的过期时间，防止当前服务器出问题后导致死锁，return nil; end;返回nil 结束

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;

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;返回nil，结束

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

锁存在, 但不是当前线程加的锁，则返回锁的过期时间。

### RedissonLock解锁 unlock源码 ###

@Override
        public void unlock() {
            Boolean opStatus = commandExecutor.evalWrite(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(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();
            }
        }

其中：

*  KEYS\[1\] 表示的是getName() 代表锁名test\_lock
 *  KEYS\[2\] 表示getChanelName() 表示的是发布订阅过程中使用的Chanel
 *  ARGV\[1\] 表示的是LockPubSub.unLockMessage 是解锁消息，实际代表的是数字 0，代表解锁消息
 *  ARGV\[2\] 表示的是internalLockLeaseTime 默认的有效时间 30s
 *  ARGV\[3\] 表示的是getLockName(thread.currentThread().getId())，是当
    
    前锁id+线程id

语义分析:

if (redis.call('exists', KEYS[1]) == 0) then
             redis.call('publish', KEYS[2], ARGV[1]);
             return 1;
             end;

if (redis.call(‘exists’, KEYS\[1\]) == 0) 如果锁已经不存在(可能是因为过期导致不存在，也可能是因为已经解锁)

then redis.call(‘publish’, KEYS\[2\], ARGV\[1\]) 则发布锁解除的消息

return 1; end 返回1结束

if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then 
             return nil;
             end;

if (redis.call(‘hexists’, KEYS\[1\], ARGV\[3\]) == 0) 如果锁存在，但是若果当前线程不是加锁的线

then return nil;end则直接返回nil 结束

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;

local counter = redis.call(‘hincrby’, KEYS\[1\], ARGV\[3\], -1) 如果是锁是当前线程所添加，定义变量counter，表示当前线程的重入次数-1,即直接将重入次数-1

if (counter > 0)如果重入次数大于0，表示该线程还有其他任务需要执行

then redis.call(‘pexpire’, KEYS\[1\], ARGV\[2\]) 则重新设置该锁的有效时间

return 0 返回0结束

else redis.call(‘del’, KEYS\[1\])否则表示该线程执行结束，删除该锁

redis.call(‘publish’, KEYS\[2\], ARGV\[1\])并且发布该锁解除的消息

return 1; end;返回1结束

return nil;

其他情况返回nil并结束

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

脚本执行结束之后，如果返回值不是0或1，即当前线程去解锁其他线程的加锁时，抛出异常。

### RedissonLock强制解锁源码 ###

@Override
        public void forceUnlock() {
            get(forceUnlockAsync());
        }
        Future<Boolean> forceUnlockAsync() {
            cancelExpirationRenewal();
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                    "if (redis.call('del', KEYS[1]) == 1) then "
                    + "redis.call('publish', KEYS[2], ARGV[1]); "
                    + "return 1 "
                    + "else "
                    + "return 0 "
                    + "end",
                    Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage);
        }

以上是强制解锁的源码,在源码中并没有找到forceUnlock()被调用的痕迹(也有可能是我没有找对),但是forceUnlockAsync()方法被调用的地方很多，大多都是在清理资源时删除锁。此部分比较简单粗暴，删除锁成功则并发布锁被删除的消息，返回1结束，否则返回0结束。

## **\# 总结** ##

这里只是简单的一个redisson分布式锁的测试用例，并分析了执行lua脚本这部分，如果要继续分析执行结束之后的操作，需要进行netty源码分析 ，redisson使用了netty完成异步和同步的处理。

往期推荐

[面试官：请你谈谈Redis 6.0 之前为什么一直不使用多线程？][Redis 6.0]

[Redis分布式锁使用不当，导致超卖了100瓶飞天茅台！！！][Redis_100]

[从几百个生产环境总结出的Redis性能问题排查解决手册（值得收藏）][Redis]

[面试官：作为架构师，请你谈谈Saas 应用如何搭建？][Saas]

[面试官：请你谈谈用Redis实现一个轻量级的搜索引擎！][Redis 1]

[接商城类私活就靠他了,前后端都有][Link 1]

[大厂实践｜Apache Pulsar 消息队列在拉卡拉的应用][Apache Pulsar]

[面试官：请你从架构演进的角度讲讲redis、kafka和 pulsar消息队列][redis_kafka_ pulsar]

![c6b0076d85c3607ace1ce09138c6b29b.png][]

![17538c0d4fa4069af11aaf3466d77dc4.png][]

有帮助？点点 在看

好看的人，从不白嫖，点赞在看分享，一键三连吧!

![ad72a73bac65e2d6208cce649a58b853.gif][]

我在等你

[c9a22365e97aa79f254ee8a021a04f21.png]: /images/20221020/4fca60275ded4cbb87f0c7baa1f1997e.png
[GET key_ key _ key _ nil]: http://mp.weixin.qq.com/s?__biz=MzIzMzgxOTQ5NA%3D%3D&chksm=e8fd2f43df8aa655fbf2c75d3c7e60412519d0aaa17dbcb77df628901e979c43a25bdd1e2a27&idx=1&mid=2247512906&scene=21&sn=0d38d67ce09b9042fb31799d5c0d1042#wechat_redirect
[Redis 6.0]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e6c3cb336fd5f249e121f73839526ec21faa323a5c095fd8da074b6f0416736ff9b68d91&idx=1&mid=2247488506&scene=21&sn=4bcd5734a144abf706032b376ece5a23#wechat_redirect
[Redis_100]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e6d5cb336fc3cf85e952dcc135a1a8e1b7d2e525f2aef91906aa916f5992096b2540e3b0&idx=1&mid=2247488492&scene=21&sn=d8d64692e290cc58bb13017be933a6c2#wechat_redirect
[Redis]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e6e2cb336ff46fbf5a7e44ede2480935c243ede960289181bb1f42f8f1ee1d3b621a6536&idx=1&mid=2247488475&scene=21&sn=0456c9ec0a73d77f2630aa7905444932#wechat_redirect
[Saas]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e6a0cb336fb6db7868e3643ab4cb503c47d657882ed5d6d2aadca7a84f9c4ccf9bf2a613&idx=1&mid=2247488409&scene=21&sn=931946d92b5a667a102ed9162d68e481#wechat_redirect
[Redis 1]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e645cb336f53ce15b2c4c5adfb934c4c64a515d39ba65c619eaf1e514f6e7308933de221&idx=1&mid=2247488380&scene=21&sn=a8e0c8d072c0e1d63cd4c17cbf4de124#wechat_redirect
[Link 1]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e649cb336f5fe33481c3dd6e162401ce9ee404e1d87df8b43c15c87815170b1da3173a5b&idx=1&mid=2247488368&scene=21&sn=474ded5d359ef1db87a9010cbf0dba33#wechat_redirect
[Apache Pulsar]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e652cb336f44abb008b5b35764365decb4c8f5765c05af834297b223aa6517e01ed9d26f&idx=1&mid=2247488363&scene=21&sn=1d2e50bb4b7ccfa4ada3cf6e2f7b227b#wechat_redirect
[redis_kafka_ pulsar]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e659cb336f4f7ef9969f1c9535bee6dcff8578370101cbf44ae668f3a549cf9ded726743&idx=1&mid=2247488352&scene=21&sn=9d09bd1537af44332e537a7c9c373142#wechat_redirect
[c6b0076d85c3607ace1ce09138c6b29b.png]: /images/20221020/8966469e3d5c40c4b8852b25fc1716e5.png
[17538c0d4fa4069af11aaf3466d77dc4.png]: /images/20221020/b4a0dd514ae24870b21b1f8332ffc6df.png
[ad72a73bac65e2d6208cce649a58b853.gif]: /images/20221020/cfa2d6ed5def4f909f806848925cd7a7.png