时间戳

首页 > 技术文章 > Redis 分布式锁

Redis 分布式锁

zhengzhaoxiang 2020-11-20 11:18 原文

分布式应用进行逻辑处理时经常会遇到并发问题，我们首先肯定会想到锁。关于锁大家都很熟悉。在并发编程中，我们通过锁，来避免由于竞争而造成的数据不一致问题。通常我们使用 synchronized 、Lock 来加锁。但是 Java中的锁，只能保证在同一个 JVM进程内中执行。如果在分布式集群环境下呢？

一、分布式锁

分布式锁的本质与 Java中的锁一样，就是在 Redis里面占了一个“坑”（一个固定 key 的值），当别的进程也要来占坑时（给key 设置值时）发现坑已经有人了（key 已经有值）了，就只好放弃或者稍后再试。一般使用 setnx(set if not exists)指令，当 key不存在时返回1，否则返回0；调用 del指令释放key 的值。

1 > setnx lock true                #加锁
2 OK
3 ... do something critical ...    #业务处理
4 > del lock                       #释放锁
5 (integer> 1

存在一个问题：如果逻辑执行中间出现异常，可能会导致 del指令没有被调用，这样就是陷入死锁。于是当拿到锁后，应该给锁加上一个过期时间，这样即使中间出现了异常，也可以保证在规定的时间内自动释放锁。

1 > setnx lock true
2 OK
3 > expire lock 5
4 ... do something critical ...
5 >del lock
6 (integer) 1

但是上述的逻辑也存在问题，如果在 setnx 与 expire 之间服务器进程突然挂掉了，也会导致死锁。这个问题的根源是因为 setnx 与 expire 不是原子的。如果这两个命令一块执行就没问题了。redis2.8版本中，提供了如下命令（重点）：加锁保证了原子性

1 > set lock true ex 5 nx    #设置key=lock vlue=true 过期时间5s
2 OK
3 ... do something critical ...
4 > del lock

但是在 Redis 集群环境下，这种方式是有缺陷的，它不是绝对的安全。在哨兵模式中，当主节点挂掉时，从节点会取而代之，但客户端上却没有明显感知。比如，原先第一个客户端在主节点中申请成功了一把锁，但是这把锁还没来得及同步到从节点，主节点就挂掉了，然后从节点变成了主节点，这个新的主节点内部没有这个锁，所以当另一个客户端过来请求加锁，立即就批准了。这样就会导致系统中同样一把锁被两个客户端同时持有，不安全性由此产生。不过这种不安全也仅在主从发生 failover 的情况下才会产生，而且持续时间极短，业务系统多数情况下可以容忍。

为了解决这个问题，Antirez 发明了 Redlock算法，它的流程比较复杂，不过已经有了很多开源的 library 做了良好的封装，用户可以拿来即用，比如 redlock-py。

 1 addrs = [{
 2     "host" : "localhost",
 3     "port" : 6379,
 4     "db": 0
 5 },{
 6     "host" : "localhost",
 7     "port" : 6479,
 8     "db": 0
 9 },{
10     "host" : "localhost",
11     "port" : 6579,
12     "db": 0
13 }]
14 
15 dlm = redlock.Redlock(addrs)
16 success = dlm.lock("user-lck",5000)
17 if success:
18     print "lock success"
19     dlm.unlock('user-lck')
20 else:
21     print 'ock failed'

为了使用 Redlock，需要提供多个 Redis 实例，这些实例之间相互独立，没有主从关系。同很多分布式算法一样，Redis 也使用“大多数机制”。加锁时，它会向过半节点发送set(key,value,nx=True,ex=xxx)指令，只要过半节点set 成功，就认为加锁成功。释放锁时，需要向所有节点发送del 指令。不过 Redlock 算法还需要考虑出错重试，时钟漂移等很多细节问题，同时因为Redlock 需要向多个节点进行读写，意味着其相比单实例 Redis 的性能会下降一些。

Redlock 使用场景：如果你很在乎高可用性，希望即使挂了一台 Redis 也完全不受影响，就应该考虑 Redlock。不过代价也是有的，需要更多的 Redis 实例，性能也下降了，代码上还需要引入额外的 library，运维上也需要特殊对待，这些都是需要考虑的成本。

二、超时问题

Redis 的分布式锁不能解决超时问题，如果在加锁和释放锁之间的逻辑执行时间超出了锁的过期时间，就会出现问题：线程1删除的锁不是自己的（自己的已经过期，自动删除），而是刚获取到分布式锁的线程2的 key值。【简单点就是释放了别人刚拿到的锁】，有一种稍微安全一点的方案是将 set 指令的 value 参数设置为一个随机数，释放锁时，先匹配随机数是否一致，然后再删除key，就可以避免当前线程占用的锁，不会被其他线程所删除。除非这个锁是因为过期了而被服务器自动释放了。但是匹配 value 和删除 key 不是一个原子操作，Redis 也没有提供类似原子性的操作。这就需要使用 Lua 脚本来处理了，因为 Lua 脚本可以保证连续多个指令的原子性执行。

1 if redis.call("get",KEYS[1]) == ARGV[1] then
2    return redis.call("del",KEYS[1])
3 else
4    return 0
5 end

但是这也不是一个完美的方案，它只是相对安全一点，因为如果真的超时了，当前线程的逻辑没有执行完，其他线程也会乘虚而入。

三、可重入性

可重入性是指线程在持有锁的情况下再次请求加锁，如果一个锁支持同一个线程多次加锁，那么这个锁就是可重入锁。比如Java语言中的 ReentrantLock 就是可重入锁。Redis分布式锁如果支持可重入，需要对客户端的set 方法进行包装，使用线程的 Threadlocal 变量存储当前持有锁的计数。还需要考虑内存锁计数的过期时间。

 1 /**
 2  * 可重入锁
 3  */
 4 public class RedisWithReentrantLock {
 5 
 6     //每一个线程都可以独立地改变自己的副本，而不会影响其它线程所对应的副本
 7     private ThreadLocal<Map<String,Integer>> lockers = new ThreadLocal<>();
 8 
 9     private Jedis jedis;
10     //构造器
11     public RedisWithReentrantLock(Jedis jedis) {
12         this.jedis = jedis;
13     }
14 
15     //判断当前key是否已存在
16     private boolean _lock(String key){
17         return jedis.set(key,"","nx","ex",5L) != null;
18     }
19 
20     //释放锁
21     private void _unlock(String key){
22         jedis.del(key);
23     }
24 
25     //从当前线程中获取锁信息
26     private Map<String,Integer> currentLockers(){
27         //从当前线程中获取存放的值
28         Map<String,Integer> refs = lockers.get();
29         if(refs != null){
30             return refs;
31         }
32         lockers.set(new HashMap<>());
33         return lockers.get();
34     }
35 
36     //加锁
37     public boolean lock(String key){
38         //给当前线程设置值
39         Map<String, Integer> refs = currentLockers();
40         Integer refCnt = refs.get(key);
41         if(refCnt != null){
42             refs.put(key,refCnt+1);
43             return true;
44         }
45         boolean ok = this._lock(key);
46         if(!ok){
47             return false;
48         }
49         refs.put(key,1);
50         return true;
51     }
52     
53     //释放锁
54     public boolean unlock(String key){
55         Map<String, Integer> refs = currentLockers();
56         Integer refCnt = refs.get(key);
57         if(refCnt == null){
58             return false;
59         }
60         refCnt-=1;
61         if(refCnt > 0){
62             refs.put(key,refCnt);
63         }else{
64             refs.remove(key);
65             this._unlock(key);
66         }
67         return true;
68     }
69 
70     public static void main(String[] args) {
71         Jedis jedis = new Jedis();
72         RedisWithReentrantLock redis = new RedisWithReentrantLock(jedis);
73         redis.lockers("codehole");
74         redis.lockers("codehole");
75         redis.unlock("codehole");
76         redis.unlock("codehole");
77     }
78 }

四、Redisson

在并发较大的情况下，直接在 Redis 中扣减库存一定会导致商品出现超买现象，可以引入分布式锁来避免超卖。当然分布式锁自身必须满足一下三点要求：
【1】在任何情况下分布式锁都不能沦为系统瓶颈；
【2】不能产生死锁；
【3】支持锁重入；
相比 Jedis，Redisson 确实算得上一款崭新的 Redis 客户端 API，它支持丰富的数据类型，并且是线程安全的，底层还使用了 Netty4 进行网络通信。那么我们能够在程序中用 Redisson 代替 Jedis 来与 Redis 进行交互？其实，Redisson 仅仅是为了扩展 Jedis 的部分功能，两者是并存的，比如Redisson 并不支持 String 类型的数据结构。

1 <dependency>
2     <groupId>org.redisson</groupId>
3     <artifactId>redisson</artifactId>
4     <version>2.2.11</version>
5 </dependency>

 在程序中使用 Redisson 客户端实现基于 Redis 的分布式锁，如下：

 1 private static Config config;
 2 private static ClusterServersConfig clusterServersConfig;
 3 private static String address = "127.0.0.1:6379";
 4 public static @BeforeClass void init(){
 5     config = new Config();
 6     //使用集群模式
 7     clusterServersConfig = config.useClusterServers();
 8     clusterServersConfig.addNodeAddress(address);
 9     clusterServersConfig.setMasterConnectionPoolSize(100);
10     clusterServersConfig.setSlaveConnectionPoolSize(100);
11     clusterServersConfig.setTimeout(1000);
12 }
13 
14 public @Test void testLock(){
15     Redisson Redisson = null;
16     try{
17         redisson = Redisson.create(config);
18         RLock lock = redisson.getLock("testLock");
19         //获取锁
20         lock.lock(20,TimeUnit.MILLISECONDS);
21         //释放锁
22         lock.unlock();
23         //尝试获取锁
24         boolean result = lock.tryLock(10,20,TimeUnit.MILLISECONDS);
25         //判断是否成功获取到锁
26         if(result){
27             lock.forceUnlock();
28         }
29     } catch (Exception e){
30         e.printStackTrace();
31     } finally {
32         if(null != redisson){
33             redisson.shutdown();
34         }
35     }
36 }

上述程序中，使用了两种获取分布式锁的方法。lock(long leaseTime,TimeUnit unit) 方法中的第1个参数用于设定分布式锁的租约时间，而第2个参数则为时间单位。使用这种方式意味着在某一个获取到锁的线程未释放锁之前，其他线程只能够在队列中阻塞等待，这和 InnoDB 引擎提供的行锁机制如出一辙，并发越高等待的线程越多。因此，在并发较大时，建议使用 tryLock(long waitTime,long leaseTime,TimeUnit unit) 方法获取分布式锁。

在 tryLock() 方法中开发人员可以通过参数 “waitTime” 来设定获取分布式锁的等待时间，超出规定的时间阈值后，线程将不再继续等待拿锁；那么为了提升库存扣减的成功率，可以在获取锁失败后尝试多次。相比 lock() 方法的拿锁方式，后者在并发较大的情况下不会使分布式锁沦为系统瓶颈，但是商品库存的扣减成功率会受到一定影响。

例如将商品库存的扣减操作转移到 Redis 中主要是为了避免数据库沦为系统瓶颈。既然性能问题得到了解决，那么变化后的实时库存应该如何同步到数据库？当系统获取到分布式锁并成功扣减 Redis 中的实时库存后，可以将消息写入到消息队列中，由消费者负责实际库存的扣减。由于采用了排队机制，并发写入数据库时的流量可控，因此数据库的负载压力就会始终保持在一个恒定的范围内，不会因为流量的影响而导致数据库性能下降。

 

推荐阅读