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

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

理解分布式锁

为什么要分布式锁

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

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

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

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

什么是分布式锁

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

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

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）命令实现，这里给出伪代码如下

1
2
3
4
5
6
7
8

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命令解决，我们先来看一下这个命令的语法

1

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的组合命令来使用，而且是原子性的，改造代码如

1
2
3
4
5
6
7

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，代码如：

1
2
3
4
5
6
7
8
9
10
11

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脚本来解决一致性问题

1
2

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脚本去执行而已，现在代码可以这样写了

1
2
3
4
5
6
7
8
9
10
11

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 呢？如果是这样的话未获取到锁的线程就执行失败了，啥也没做，这是不可行的，我们是不是需要让未获取到锁的线程等待片刻之后再次尝试获取锁呢？如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

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的概述，官方文档

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的集成

导入依赖

1
2
3
4
5
6
7

<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.6</version>
</dependency>

配置一个单机Redis

1
2
3
4
5
6
7
8
9
10
11

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

可重入锁（Reentrant Lock）

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

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

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@Autowired
  private RedissonClient redissonClient;

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

我们可以手动指定锁的过期时间,但是手动指定了过期时间就不会再进行时间的自动续期了，如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@Test
  public void testLock(){
  
      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("释放锁....");
      }
  }

这里有一个问题，手动指定了过期时间就不会再自动续期，那么如果在线程A业务还没执行完时锁自动过期，这时候线程B获取到锁，那么等线程A业务执行完释放锁，就会把线程B的锁删除，当然这种情况Redisson会报异常，所以如果要手动设定过期时间需要让过期时间比业务逻辑执行的时间长才对

这里我们看到，其实Redisson底层帮我了处理了上述的一些列问题，那么Redisson是如何实现上面的功能的呢？

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

后话

文章结束，这里主要探讨了一下Redis实现分布式锁的核心原理，以及过程中的一些问题，最终采用Redisson实现分布式锁，希望文章对你有所帮助