六.Redis极简入门-Redis实现分布式锁原理

你的名字 2022-12-22 09:06 114阅读 0赞

老鸟飞过，学习使用，欢迎交流

## 理解分布式锁 ##

### 为什么要分布式锁 ###

在并发场景中，我们可以使用加锁的手段来保证业务方法或代码的原子性操作，从而防止数据被并发修改引发安全问题，在单体应用中我们可以使用互斥锁如： `synchronized` 同步代码块 或者 `Lock`锁来实现，如图:

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7lCrzyXy-1605714910565)(分布式锁.assets/1605692967171.png)\]][img-7lCrzyXy-1605714910565_.assets_1605692967171.png]

但是在`集群/分布式`应用中单纯的互斥锁是不能保证多个节点中对同一个数据的原性操作的，如图：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rGJm7d9A-1605714910569)(分布式锁.assets/1605695422135.png)\]][img-rGJm7d9A-1605714910569_.assets_1605695422135.png]

集群模式中，每个服务都加了锁但是只能锁住自己，每个服务做库存做扣减操作，当库存都剩1的时候，三个服务并发减库存可能会导致库存减到 -2 出现线程不安全问题。

那应该如何解决上述问题呢？我们需要使用`分布式事务`保证三个服务对库存的`原子性操作`

### 什么是分布式锁 ###

分布式锁的目的就是在`分布式/集群`环境中使用加锁手段保证`多个服务节点对同一个数据的原子性操作`，保证数据的安全性，如上图，三个服务都在同时扣减库存，我们需要`对减库存进行原子性操作`，如：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ITVue2Th-1605714910572)(分布式锁.assets/1605709890367.png)\]][img-ITVue2Th-1605714910572_.assets_1605709890367.png]

实现分布式锁的原理也很简单，就是需要得有一把唯一且共享的锁，多个服务同时去获取锁，但是只有一个服务才能获取到锁，其他没有获取到锁的服务需要等待或者自旋，等获取到锁的服务业务执行完成释放锁，其他的服务就可以再次尝试获取锁。

## Redis分布式锁的原理 ##

实现分布式锁的方案有很多，比如基于数据库实现分布式锁，使用`ZooKeeper`实现分布式锁，本文采用的是使用`Redis实现分布式`锁方案。

### 加锁和释放锁 ###

Redis提供了一个命令`setnx` 可以来实现分布式锁，该命令只在键 `key` 不存在的情况下 将键 `key` 的值设置为 `value` ，若键 `key` 已经存在， 则 `SETNX` 命令不做任何动作。根据这一特性我们就可以制定Redis实现分布式锁的方案了。

简单理解就是 ：如果三个服务同时抢锁，服务A抢先一步执行`setnx(lock_stock,1)`加上锁，那么当服务B在执行`setnx(lock_stock,1)`加锁的时候就会失败，服务C也一样，服务A抢到锁执行完业务逻辑后就会释放锁，可以使用`del(lock_stock)`删除锁，其他服务就可以执行`setnx(lock_stock,1)`加锁了，如图：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7U0ig190-1605714910578)(分布式锁.assets/1605709864403.png)\]][img-7U0ig190-1605714910578_.assets_1605709864403.png]

### 锁超时问题 ###

这里有一个问题，如果获取到锁的服务在`释放锁的时候宕机`了，那么Redis中`lock-stock`不就永远存在，那`锁不就释放不了`么，别的服务也就没办法获取到锁，就造成了`死锁`，为了解决这个问题，我们需要设置`锁的自动超时`也就是Key的超时自动删除，即使服务宕机没有调用del释放锁，那么锁本身也有超时时间，可以自动删除锁，别的服务就可以获取锁了，Redis中Key的过期时间可以使用Redis的 `expire（lock_stock，30）`命令实现，这里给出伪代码如下

if（jedis.setnx（lock_stock，1） == 1）{ 	//获取锁
        expire（lock_stock，5）		 //设置锁超时
        try { 
            业务代码
        } finally { 
            jedis.del（lock_stock）			 //释放锁
        }
    }

### 原子性问题 ###

上面的代码依然有问题，就是`setnx`获取锁和`expire`不是原子性操作，假设有一极端情况，当线程通过`setnx（lock_stock，1）`获取到锁，还没来得及执行`expire（lock_stock，30）`设置锁的过期时间，服务就宕机了，那是不是锁也永远得不到释放呢？？？又变成了死锁，这个问题可以使用`set`命令解决，我们先来看一下这个命令的语法

SET key value [EX seconds] [PX milliseconds] [NX|XX]

从 Redis 2.6.12 版本开始， `SET` 命令的行为可以通过一系列参数来修改：

*  `EX seconds` ： 将键的过期时间设置为 `seconds` 秒。 执行 `SET key value EX seconds` 的效果等同于执行 `SETEX key seconds value` 。
 *  `PX milliseconds` ： 将键的过期时间设置为 `milliseconds` 毫秒。 执行 `SET key value PX milliseconds` 的效果等同于执行 `PSETEX key milliseconds value` 。
 *  `NX` ： 只在键不存在时， 才对键进行设置操作。 执行 `SET key value NX` 的效果等同于执行 `SETNX key value` 。
 *  `XX` ： 只在键已经存在时， 才对键进行设置操作。

也就是说该命令可以当做`setnx`和`expire`的组合命令来使用，而且是原子性的，改造代码如

if（set(lock_stock,1,"NX","EX",5) == 1）{ 	//获取锁并设置超时
        try { 
            业务代码
        } finally { 
            del（lock_stock）			 		//释放锁
        }
    }

### 锁的误删除问题 ###

上面的方案依然有问题，就是在`del`释放锁的时候可能会误删除别人加的锁，例如服务A获取到锁`lock_stock`，过期时间为 5s，如果在服务A执行业务逻辑的这一段时间内，锁到期自动删除，且别的服务获取到了锁`lock_stock`，那么服务A业务执行完成执行`del(lock_stock)`是不是会把别人的锁给删除掉呢？？？如图：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rSEN6vhr-1605714910581)(分布式锁.assets/1605712581554.png)\]][img-rSEN6vhr-1605714910581_.assets_1605712581554.png]

那么这个问题怎么解决呢？我们可以在删除锁的时候先判断一下要删除的锁是不是自己上的锁，比如可以把锁的值使用一个UUID，在释放锁的时候先获取一下锁的值和当前业务中创建的UUID是不是同一个，如果是才执行·`del`删除锁，当然也可以使用线程的ID替代UUID，代码如：

String uuid = UUID.randomUUID().toString();
    if（jedis.set(lock_stock,uuid,"NX","EX",5) == 1）{ 	//获取锁并设置超时
        try { 
            业务代码
        } finally { 
            String lockValue = jedis.get(lock_stock);	//获取锁的值
            if(lockValue.equals(uuid)){ 			//判断是不是自己的锁
                jedis.del（lock_stock）			 	  //释放锁
            }
        }
    }

### Lua脚本保证原子性 ###

但是上面的代码依然有问题，就是判断锁的代码和删除锁的代码也不是原子性的，依然可能会导致锁的误删除问题，比如服务A在判断锁成功准备删除锁时，锁自动过期，别的服务B获取到了锁，然后服务A执行`DEL`就可能会把服务B的锁给删除掉，所以，我们必须保证 获取锁 -> 判断锁 -> 删除锁 的操作是原子性的才可以，解决方案可以使用Redis+Lua脚本来解决一致性问题

String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

这是一段Lua脚本，可以保证多个命令的原子性

*  redis.call(‘get’, KEYS\[1\]) ：是调用redis的get命令，key可以通过参数传入
 *  == ARGV\[1\] ：意思是是否和 某个值相等，这里的值也可以参数传入
 *  then return redis.call(‘del’, KEYS\[1\]) ：如果相等就执行 `redis.call('del', KEYS[1])`删除操作
 *  else return 0 end ：否则就返回 0

如果我们把数据带入`KEYS[1]`的值为“lock\_stock”,`ARGV[1]`的值为UUID如“xoxoxo”，所以大概的含义是如果调用get(“lock\_stock”)获取到的值 等于 “xoxoxo” ,那就调用 del(“lock\_stock”)，否则就返回 0 。 说白了就是把我们上面的判断锁和删除锁的动作使用Lua脚本去执行而已，现在代码可以这样写了

String uuid = UUID.randomUUID().toString();
    if（jedis.set(lock_stock,uuid,"NX","EX",5) == 1）{ 	//获取锁并设置超时
        try { 
            业务代码
        } finally { 
            //lua脚本
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            //执行脚本
            jedis.eval(script, Collections.singletonList("lock_stock"),Collections.singletonList(uuid));
        }
    }

*  Arrays.asList(“lock\_stock”) 转给 KEYS\[1\]
 *  Arrays.asList(uuid)转给 ARGV\[1\]

### 可重入锁 ###

上面的代码是不完整的，如果某个线程没有获取到锁是不是就不会进入 IF 呢？如果是这样的话未获取到锁的线程就执行失败了，啥也没做，这是不可行的，我们是不是需要让未获取到锁的线程等待片刻之后再次尝试获取锁呢？如下：

public void method(){ 
    	String uuid = UUID.randomUUID().toString();
    	if（jedis.set(lock_stock,uuid,"NX","EX",5) == 1）{ 	//获取锁并设置超时
    	    try { 
    	        业务代码
    	    } finally { 
    	        //lua脚本
    	        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    	        //执行脚本
    	        jedis.eval(script, Collections.singletonList("lock_stock"),Collections.singletonList(uuid));
    	    }
    	}else{ 
    		//休眠一会儿，重入方法，尝试获取锁
    		Thread.sleep(100);
    		method();	//自旋，重新进入方法
    	}
    }

上面的代码增加了else获取锁失败的逻辑，休眠一会儿后`重入方法尝试重新获取锁`，休眠时间结合业务逻辑的执行时间设定

### 分布式锁的特点 ###

当然要实现一个分布式锁还需要考虑一些东西，比如Redis的健壮性，它不能随便挂掉，这里总结一下分布式锁的一些要素  
首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

*  互斥性：同一时间只能一个节点获取到锁，其他节点需要等待获取到锁的节点释放了锁才可以获取到锁，而这里的等待一般是通过阻塞，和自旋两种方式
 *  安全性：解铃还须系铃人，只能释放自己的锁不能误删别人的锁
 *  死锁：比如在节点宕机时最容易出现锁没被释放的问题，然后出现死锁，所以做锁的过期
 *  容错：当Redis宕机，客户端仍然可以释放锁
 *  可重入：获取锁失败可以重新尝试获取锁

要实现一个分布式锁是不是要考虑很多细节呢，其实不用做什么麻烦，我们有更专业的工具已经帮我们封装好上面的所有细节

## Redisson的实现分布式锁 ##

### Redisson是什么 ###

我们操作Redis的手段有很多，在Java中可以使用Jedis或者Redisson，本文章在于讨论Redisson是如何操作Java的，下面是对Redisson的概述，[官方文档][Link 1]

> Redisson是一个实现的Java操作Redis的工具包，它不仅提供了一系列常用的操作Redis的API，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法，Redisson的宗旨是促进使用者对Redis的关注分离，从而让使用者能够将精力更集中地放在处理业务逻辑上。

### Redisson的集成 ###

导入依赖

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

配置一个单机Redis

@Configuration
    public class RedissonConfig { 
    
        //创建客户端
        @Bean
        public RedissonClient redissonClient(){ 
            Config config = new Config();
            config.useSingleServer().setAddress("redis://127.0.0.1:6379");//.setPassword("123456");
            return Redisson.create(config);
        }
    }

### Redisson实现分布式锁 ###

官方对分布式锁的定义

> 大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。  
> 另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

以上是Redisson官方文档对分布式锁的解释总结下来有两点

*  Redisson加锁自动有过期时间30s，监控锁的看门狗发现业务没执行完，会自动进行锁的续期(重回30s)，这样做的好处是防止在程序执行期间锁自动过期被删除问题
 *  当业务执行完成不再给锁续期，即使没有手动释放锁，锁的过期时间到了也会自动释放锁

#### 可重入锁（Reentrant Lock） ####

> 基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口

一个简单的锁分布式锁案例如下：

@Autowired
        private RedissonClient redissonClient;
    
        @Test
        public void testLock1(){ 
            RLock rLock = redissonClient.getLock("lock_stock");
            rLock.lock();	//阻塞式等待，过期时间30s
            try{ 
                System.out.println("加锁成功....");
                System.out.println("执行业务....");
            }finally { 
                rLock.unlock();
                System.out.println("释放锁....");
            }
        }

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了，如下

@Test
        public void testLock2(){ 
        
            RLock rLock = redissonClient.getLock("lock_stock");
        	// 加锁以后10秒钟自动解锁
    		// 无需调用unlock方法手动解锁
    		rLock.lock(10, TimeUnit.SECONDS);
            try{ 
                System.out.println("加锁成功....");
                System.out.println("执行业务....");
            }finally { 
                rLock.unlock();
                System.out.println("释放锁....");
            }
        }

Redisson对分布式锁实现细节进行了封装，帮我们处理了分布式锁面临的一些列问题，那么Redisson是如何工作的呢？？？

*  如果没有设置过期时间，Redisson以 30s 作为锁的默认过期时间，获取锁成功后(底层也用到了Lua脚本保证原子性)会开启一个定时任务定时进行锁过期时间续约，即每次都把过期时间设置成 30s，定时任务 10s执行一次(看门狗)
 *  如果设置了过期时间，直接把设定的过期时间作为锁的过期时间，然后使用Lua脚本获取锁，没获取到锁的线程会while自旋重入不停地尝试获取锁

这里需要注意，`rLock.lock(10, TimeUnit.SECONDS)`指定了解锁时间，Redisson就不会再自动续期，那么如果在线程A业务还没执行完就自动解锁了，这时候线程B获取到锁，继续执行业务，那么等线程A业务执行完释放锁就可能会把线程B的锁删除，当然这种情况Redisson会报异常，但是这种情况是没有把所有线程都锁住的，所以如果要手动设定过期时间需要让过期时间比业务逻辑执行的时间长才对

Redisson同时还为分布式锁提供了异步执行的相关方法

@Test
    public void testLock3() { 
    
        RLock rLock = redissonClient.getLock("lock_stock");
        try{ 
            //rLock.lockAsync();
    
            //10秒自动释放锁
            //rLock.lockAsync(10, TimeUnit.SECONDS);
    
            //尝试加锁等待2秒，上锁以后10秒自动释放锁
            Future<Boolean> res = rLock.tryLockAsync(2, 10, TimeUnit.SECONDS);
                if(res.get()){ 
                    System.out.println("加锁成功....");
                    System.out.println("执行业务....");
                }
        } catch (InterruptedException e) { 
            e.printStackTrace();
        } catch (ExecutionException e) { 
            e.printStackTrace();
        }finally { 
            rLock.unlock();
            System.out.println("释放锁....");
        }
    }

RLock对象完全符合Java的Lock规范。也就是说只有拥有锁的进程才能解锁，其他进程解锁则会抛出IllegalMonitorStateException错误。但是如果遇到需要其他进程也能解锁的情况，请使用分布式信号量Semaphore 对象。

#### 公平锁（Fair Lock） ####

基于Redis的Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待5秒后继续下一个线程，也就是说如果前面有5个线程都处于等待状态，那么后面的线程会等待至少25秒

@Test
     public void testLock5() { 
         RLock fairLock= redissonClient.getFairLock("anyLock");
         try{ 
             // 最常见的使用方法
             fairLock.lock();
         }finally { 
             fairLock.unlock();
             System.out.println("释放锁....");
         }
     }

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了

// 10秒钟以后自动解锁
    // 无需调用unlock方法手动解锁
    fairLock.lock(10, TimeUnit.SECONDS);
    
    // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
    boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    fairLock.unlock();

Redisson同时还为分布式可重入公平锁提供了异步执行的相关方法

RLock fairLock = redisson.getFairLock("anyLock");
    fairLock.lockAsync();
    fairLock.lockAsync(10, TimeUnit.SECONDS);
    Future<Boolean> res = fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);

#### 联锁（MultiLock） ####

基于Redis的Redisson分布式联锁RedissonMultiLock对象可以将多个RLock对象关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例。

RLock lock1 = redissonInstance1.getLock("lock1");
    RLock lock2 = redissonInstance2.getLock("lock2");
    RLock lock3 = redissonInstance3.getLock("lock3");
    
    RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
    // 同时加锁：lock1 lock2 lock3
    // 所有的锁都上锁成功才算成功。
    lock.lock();
    ...
    lock.unlock();

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
    // 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
    lock.lock(10, TimeUnit.SECONDS);
    
    // 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
    boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

#### 红锁（RedLock） ####

Redis常用的方式有单节点、主从模式、哨兵模式、集群模式，在后三种模式中可能会出现 ，异步数据丢失，脑裂问题，Redis官方提供了解决方案：RedLock,RedLock是基于redis实现的分布式  
锁，它能够保证以下特性：

*  容错性：只要多数节点的redis实例正常运行就能够对外提供服务,加锁释放锁
 *  互斥性：只能有一个客户端能获取锁，即使发生了网络分区或者客户端宕机，也不会发生死锁

基于Redis的Redisson红锁RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁，每个RLock对象实例可以来自于不同的Redisson实例

RLock lock1 = redissonInstance1.getLock("lock1");
    RLock lock2 = redissonInstance2.getLock("lock2");
    RLock lock3 = redissonInstance3.getLock("lock3");
    
    RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
    // 同时加锁：lock1 lock2 lock3
    // 红锁在大部分节点上加锁成功就算成功。
    lock.lock();
    ...
    lock.unlock();

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
    // 给lock1，lock2，lock3加锁，如果没有手动解开的话，10秒钟后将会自动解开
    lock.lock(10, TimeUnit.SECONDS);
    
    // 为加锁等待100秒时间，并在加锁成功10秒钟后自动解开
    boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

#### 读写锁（ReadWriteLock） ####

基于Redis的Redisson分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。其中读锁和写锁都继承了RLock接口。

分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态,即：使用同一个RReadWriteLock加写锁和读锁，多个读锁是需要等待写释放锁才能加锁成功，如下：

@Test
        public void testWriteLock() { 
            //获取读写锁
            RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("ReadWriteLock");
            //获取写锁
            RLock rLock = readWriteLock.writeLock();
            try{ 
                //加上写锁，读会等待
                rLock.lock();
                System.out.println("写锁加锁成功");
                Thread.sleep(200000);
                System.out.println("处理写业务...");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }finally { 
                rLock.unlock();
                System.out.println("释放写锁....");
            }
        }
        @Test
        public void testReadLock() { 
            //获取读写锁
            RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("ReadWriteLock");
            //获取读锁
            RLock rLock = readWriteLock.readLock();
            try{ 
                //加上读锁，如果写锁没释放会等待
                rLock.lock();
                System.out.println("读锁加锁成功");
                System.out.println("处理读业务...");
            }finally { 
                rLock.unlock();
                System.out.println("释放读锁....");
            }
        }

如果 testWriteLock 写方法先自行，先加上写锁 ，那么 testReadLock读方法中的加锁代码会等待，直到写锁释放。  
当然如果多个线程全是读锁没有写锁那相当于是没有加锁，不会等待，其他情况只要有写锁参与，后执行加锁的线程都要等先执行加锁的线程释放锁，不管是先读还是先写，又或者是写和写。这种锁能够保证读锁能读到的数据始终是写完之后的数据。

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了

// 10秒钟以后自动解锁
    // 无需调用unlock方法手动解锁
    rwlock.readLock().lock(10, TimeUnit.SECONDS);
    // 或
    rwlock.writeLock().lock(10, TimeUnit.SECONDS);
    
    // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
    boolean res = rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS);
    // 或
    boolean res = rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS);
    ...
    lock.unlock();

#### 信号量（Semaphore） ####

基于Redis的Redisson的分布式信号量（Semaphore）Java对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

信号量可以看做是在Redis中保存了一个数字，然后可以实现原子性的加或者减，比如说有一商品需要拿100个做秒杀，我们就可以把这个库存数量做成信号量，然后实现原子性加减操作：

@Test
    public void testReadLock5() throws InterruptedException { 
        //获得到一个信号量
        RSemaphore semaphore = redissonClient.getSemaphore("semaphore");
        //设置信号量的值
        boolean setPermits = semaphore.trySetPermits(1000);
        System.out.println(setPermits);
        System.out.println("可用数量："+semaphore.availablePermits());
    
    }
    @Test
    public void testReadLock6() throws InterruptedException { 
        //获得到一个信号量
        RSemaphore semaphore = redissonClient.getSemaphore("semaphore");
        //获取 2 个信号量 ， 值会减去 2 ， 如果获取不到，方法会阻塞
        semaphore.acquire(2);
        System.out.println("可用数量："+semaphore.availablePermits());
    
        //尝试获取 2 个信号量 ， 值会减去 2 ， 如果获取不到，方法不会
        boolean tryAccquireSuccess = semaphore.tryAcquire(2);
        System.out.println(tryAccquireSuccess);
        System.out.println("可用数量："+semaphore.availablePermits());
    
    }
    
    @Test
    public void testReadLock7() throws InterruptedException { 
        //获得到一个信号量
        RSemaphore semaphore = redissonClient.getSemaphore("semaphore");
        //释放2个值，数量会加回去
        semaphore.release(2);
        System.out.println("可用数量："+semaphore.availablePermits());
    }

#### 可过期性信号量（PermitExpirableSemaphore） ####

基于Redis的Redisson可过期性信号量（PermitExpirableSemaphore）是在RSemaphore对象的基础上，为每个信号增加了一个过期时间。每个信号可以通过独立的ID来辨识，释放时只能通过提交这个ID才能释放。它提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

RPermitExpirableSemaphore semaphore = redisson.getPermitExpirableSemaphore("mySemaphore");
    String permitId = semaphore.acquire();
    // 获取一个信号，有效期只有2秒钟。
    String permitId = semaphore.acquire(2, TimeUnit.SECONDS);
    // ...
    semaphore.release(permitId);

#### 闭锁（CountDownLatch） ####

基于Redisson的Redisson分布式闭锁（CountDownLatch）Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。  
闭锁可以实现多个线程都执行完才是完成的效果，否则闭锁会等待

RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
    //设置2个数量
    latch.trySetCount(2);
    
    //await方法会等待，等待其他线程 countDown 完成所有的trySetCount(2)次就结束闭锁
    latch.await();
    
    // 在其他线程或其他JVM里
    RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
    //完成第1个
    latch.countDown();
    
    // 在其他线程或其他JVM里
    RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
    //完成第2个 ， 闭锁完成
    latch.countDown();

## 后话 ##

文章结束，这里主要探讨了一下Redis实现分布式锁的核心原理，以及Redisson实现分布式锁的几种类型，希望文章对你有所帮助

[img-7lCrzyXy-1605714910565_.assets_1605692967171.png]: /images/20221120/45714d48569444cfb43e8cec187dbc4e.png
[img-rGJm7d9A-1605714910569_.assets_1605695422135.png]: /images/20221120/290f35c467264adbbc84eb8824ac87af.png
[img-ITVue2Th-1605714910572_.assets_1605709890367.png]: /images/20221120/b00ce1eb9c02499ca68cbef95faf3f06.png
[img-7U0ig190-1605714910578_.assets_1605709864403.png]: /images/20221120/90de6d60293c4c528be2f53c32a132ad.png
[img-rSEN6vhr-1605714910581_.assets_1605712581554.png]: /images/20221120/e5c475bc4b054253a77a7bede1c0e48f.png
[Link 1]: https://github.com/redisson/redisson/wiki/Table-of-Content