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

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

理解分布式锁

为什么要分布式锁

在java中可以通过synchronize和lock等手段来实现。

什么是分布式锁

很多时候我们需要保证一个方法在同一时间内只能被同一个线程执行。在单机环境中，通过 Java 提供的并发 API 我们可以解决，但是在分布式环境下，就没有那么简单啦。

• 分布式与单机情况下最大的不同在于其不是多线程而是多进程。
• 多线程由于可以共享堆内存，因此可以简单的采取内存作为标记存储位置。而进程之间甚至可能都不在同一台物理机上，因此需要将标记存储在一个所有进程都能看到的地方。

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

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

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)加锁了，如图：

锁超时问题

这里有一个问题，如果获取到锁的服务在释放锁的时候宕机了，那么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)是不是会把别人的锁给删除掉呢？？？如图：

那么这个问题怎么解决呢？我们可以在删除锁的时候先判断一下要删除的锁是不是自己上的锁，比如可以把锁的值使用一个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宕机，客户端仍然可以释放锁
• 可重入：获取锁失败可以重新尝试获取锁

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

Redisson的实现分布式锁

Redisson是什么

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

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(""); return Redisson.create(config); } } 

Redisson实现分布式锁

官方对分布式锁的定义

以上是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的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(); 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("释放读锁...."); } } 

另外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）

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实现分布式锁的几种类型，希望文章对你有所帮助