Redisson分布式锁实现原理和使用

我会带着你远行 2023-01-08 06:29 159阅读 0赞

## 常见的锁 ##

内存锁lock,synchronize  
分布式锁redis，zookeeper实现  
Redisson基于redis实现了Lock接口的分布式集群锁，是可重入锁，功能强大，源码复杂，比redis单机模式分布式锁可靠，稳定性更高，支持集群模式，支持锁根据业务时长自动延迟释放

redis普通分布式锁存在一定的缺陷——它加锁只作用在一个Redis节点上，如果通过sentinel和cluster保证高可用，如果master节点由于某些原因发生了主从切换，那么就会出现锁丢失的情况：

**高可用问题**

*  客户端1在Redis的master节点上拿到了锁
 *  Master宕机了，存储锁的key还没有来得及同步到Slave上
 *  master故障，发生故障转移，slave节点升级为master节点
 *  客户端2从新的Master获取到了对应同一个资源的锁  
    　　于是，客户端1和客户端2同时持有了同一个资源的锁。锁的安全性被打破了。针对这个问题。Redis作者antirez提出了RedLock算法来解决这个问题

**业务线超时问题**

*  节点1：如果设置锁的过期时间为30MS,但是业务线可能因为网络或者数据量峰值出现导致执行时间超过了30MS，那么这时redis已经把锁给释放了，但是业务线却仍然在执行
 *  节点2这时去获取锁，发现锁可以获取成功，这就造成了 同时有两个节点在执行同一个业务逻辑，则无法保证业务的幂等性（数据加上版本号处理，但仍然会对累加结果造成重复性错误），会造成数据重复处理，或者日志主键ID重复，同一订单两次计算金额，客户两次扣款等问题

Redisson在这个问题上加上了自动续时，如果锁已经被持有，那么另一个线程试图再次获取锁时，会把已存在的锁重置过期时间，就相当于延长了当前锁的时间从而避免过期造成幂等性问题

## 源码RedLock算法的实现思路 ##

全流程图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]

### 加锁机制 ###

lock()方法直接调用 lockInterruptibly 方法，tryAcquire就是获取锁的具体实现

@Override
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { 
        // 获取当前线程的id
        long threadId = Thread.currentThread().getId();
        // 尝试获取锁
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired ，ttl为空，说明成功获取锁，返回
        if (ttl == null) { 
            return;
        }
     
        // 如果获取锁失败，则订阅到对应这个锁的channel
        RFuture<RedissonLockEntry> future = subscribe(threadId);
        commandExecutor.syncSubscription(future);
     
        try { 
            // 没有获取倒锁就不断的自循环获取锁
            while (true) { 
            	// 再次尝试获取锁
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired ， ttl为空，说明成功获取锁，返回
                if (ttl == null) { 
                    break;
                }
     
                // waiting for message ， ttl大于0 则等待ttl时间后继续尝试获取
                if (ttl >= 0) { 
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else { 
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally { 
            // 取消对channel的订阅
            unsubscribe(future, threadId);
        }
    	// get(lockAsync(leaseTime, unit));

这里会订阅Channel，当资源可用时可以及时知道，并抢占，防止无效的轮询而浪费资源。  
当资源可用用的时候，循环去尝试获取锁，由于多个线程同时去竞争资源，所以这里用了**信号量**，对于同一个资源只允许一个线程获得锁，其它的线程阻塞，从而避免CPU一致执行while  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]

tryAcquireAsync 方法,有自定义过期时间则直接获取，没有则使用系统配置的默认过期时间获取

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { 
        // 如果存在过期时间，按照正常费那事获取锁
        if (leaseTime != -1) { 
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        // 读取配置的默认加锁时间30秒执行获取锁的 private long lockWatchdogTimeout = 30 * 1000; 
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        // 如果一直持有这个锁，开启监听通过定时任务不断刷新锁的过期时间
        ttlRemainingFuture.addListener(new FutureListener<Long>() { 
            @Override
            public void operationComplete(Future<Long> future) throws Exception { 
                if (!future.isSuccess()) { 
                    return;
                }
     
                Long ttlRemaining = future.getNow();
                // lock acquired
                if (ttlRemaining == null) { 
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;

看 tryLockInnerAsync 方法，采用LUA脚本代码加锁，LUA脚本在redis中具有原子性操作。

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { 
    	// 设置过期时间
        internalLockLeaseTime = unit.toMillis(leaseTime);
     
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
        		  // 如果锁不存在，则通过hset设置它的值，并设置过期时间
                  "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; " +
                  // 如果锁已存在，并且锁的是当前线程，则通过hincrby给数值递增1
                  "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; " +
                  // //如果锁已存在，但并非本线程，则返回过期时间ttl
                  "return redis.call('pttl', KEYS[1]);",
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

这里KEYS\[1\]就是getName()，ARGV\[2\]是getLockName(threadId)

假设前面获取锁时传的name是“abc”，假设调用的线程ID是Thread-1，假设成员变量UUID类型的id是6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c

那么KEYS\[1\]=abc，ARGV\[2\]=6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1  
这段脚本的意思是  
　　1、判断有没有一个叫“abc”的key  
　　2、如果没有，则在其下设置一个字段为“6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1”，值为“1”的键值对 ，并设置它的过期时间  
　　3、如果存在，则进一步判断“6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1”是否存在，若存在，则其值加1为2，并重新设置过期时间  
　　4、返回“abc”的生存时间（毫秒）  
这里用的数据结构是hash，hash的结构是： key 字段1 值1，字段2 值2 ，字段3 值3 …

用在锁这个场景下，key就表示锁的名称，也可以理解为临界资源，字段就表示当前获得锁的线程,  
所有竞争这把锁的线程都要判断在这个key下有没有自己线程的字段，如果没有则不能获得锁，如果有，则相当于重入，字段值加1（次数）,重入锁也就基本实现了

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 2]

### 锁互斥机制 ###

那么在这个时候，如果客户端2来尝试加锁，执行了同样的一段lua脚本，会咋样呢？  
很简单，第一个if判断会执行“exists myLock”，发现myLock这个锁key已经存在了。

接着第二个if判断，判断一下，myLock锁key的hash数据结构中，是否包含客户端2的ID，但是明显不是的，因为那里包含的是客户端1的ID。

所以，客户端2会获取到pttl myLock返回的一个数字，这个数字代表了myLock这个锁key的剩余生存时间。比如还剩15000毫秒的生存时间。

此时客户端2会进入一个while循环，不停的尝试加锁。

### 释放锁机制 ###

核心解锁代码，也用LUA脚本

protected RFuture<Boolean> unlockInnerAsync(long threadId) { 
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
        		// 如果锁已经不存在， 发布锁释放的消息
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                "end;" +
                // 如果释放锁的线程和已存在锁的线程不是同一个线程，返回null
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                    "return nil;" +
                "end; " +
                // 通过hincrby递减1的方式，释放一次锁
                // 若剩余次数大于0 ，则刷新过期时间
                "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                "if (counter > 0) then " +
                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                // 否则证明锁已经释放，删除key并发布锁释放的消息
                "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(threadId));
     
    }

假设name=abc，假设线程ID是Thread-1

同理，我们可以知道  
KEYS\[1\]是getName()，即KEYS\[1\]=abc  
KEYS\[2\]是getChannelName()，即KEYS\[2\]=redisson\_lock\_\_channel:\{abc\}  
ARGV\[1\]是LockPubSub.unlockMessage，即ARGV\[1\]=0  
ARGV\[2\]是生存时间  
ARGV\[3\]是getLockName(threadId)，即ARGV\[3\]=6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1  
因此，上面脚本的意思是：  
　　1、判断是否存在一个叫“abc”的key  
　　2、如果不存在，向Channel中广播一条消息，广播的内容是0，并返回1  
　　3、如果存在，进一步判断字段6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1是否存在  
　　4、若字段不存在，返回空，若字段存在，则字段值减1  
　　5、若减完以后，字段值仍大于0，则返回0  
　　6、减完后，若字段值小于或等于0，则广播一条消息，广播内容是0，并返回1；

可以猜测，广播0表示资源可用，即通知那些等待获取锁的线程现在可以获得锁了

**一句话说:每次释放都对myLock数据结构中的那个加锁次数减1  
如果发现加锁次数是0了，说明这个客户端已经不再持有锁了，此时就会用：  
“del myLock”命令，从redis里删除这个key。然后呢，另外的客户端2就可以尝试完成加锁了**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 3]

### watch dog自动延期机制 ###

客户端1加锁的锁key默认生存时间才30秒，如果业务执行时间超过了30秒，客户端1还是需要一直持有这把锁，怎么办呢？

简单！只要客户端1一旦加锁成功，就会启动一个watch dog看门狗，他是**一个后台线程，会每隔10秒检查一下，如果客户端1还持有锁key，那么就会不断的延长锁key的生存时间**。

### Redisson分布式锁的缺点 ###

其实上面那种方案最大的问题，就是如果你对某个redis master实例，写入了myLock这种锁key的value，此时会异步复制给对应的master slave实例。

但是这个过程中一旦发生redis master宕机，主备切换，redis slave变为了redis master。  
接着就会导致，客户端2来尝试加锁的时候，在新的redis master上完成了加锁，而客户端1也以为自己成功加了锁。

此时就会导致多个客户端对一个分布式锁完成了加锁。

这时系统在业务语义上一定会出现问题，导致各种脏数据的产生。

所以这个就是redis cluster，或者是redis master-slave架构的主从异步复制导致的redis分布式锁的最大缺陷：**在redis master实例宕机的时候，可能导致多个客户端同时完成加锁**，如果要追求强一致性，那么只能考虑zookeeper分布式锁，当然它们各有自己的优缺点。

## spingboot中配置和使用 ##

redisson配置不会影响jedis的使用，data-redis配置方式不变，因此可以同时两个一起使用

<dependency>
    	<groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    
    
    <dependency>
       	<groupId>org.redisson</groupId>
        <artifactId>redisson-spring-boot-starter</artifactId>
        <version>3.13.6</version>
    </dependency>

redisson-dev.yml集群模式，这里使用redis的cluster

clusterServersConfig:
      idleConnectionTimeout: 10000
      connectTimeout: 10000
      timeout: 3000
      retryAttempts: 3
      retryInterval: 1500
      subscriptionsPerConnection: 5
      clientName: null
      loadBalancer: !<org.redisson.connection.balancer.RoundRobinLoadBalancer> { }
      slaveConnectionMinimumIdleSize: 32
      slaveConnectionPoolSize: 64
      masterConnectionMinimumIdleSize: 32
      masterConnectionPoolSize: 64
      readMode: "SLAVE"
      password: u111111111
      nodeAddresses:
        - "redis://192.168.32.176:6371"
        - "redis://192.168.32.176:6372"
        - "redis://1192.168.32.176:6373"
      scanInterval: 1000
    #threads: 0
    #nettyThreads: 0
    codec: !<org.redisson.codec.JsonJacksonCodec> { }
    transportMode : "NIO"

RedissonConfig这里根据环境使用相应的配置

@Configuration
    public class RedissonConfig { 
    
        @Autowired
        private Environment env;
        @Autowired
        private RedissonProperties redssionProperties;
    
        @Bean(destroyMethod = "shutdown")
        public RedissonClient redissonClient() throws IOException { 
            String[] profiles = env.getActiveProfiles();
            String profile = "";
            if(profiles.length > 0) { 
                profile = "-" + profiles[0];
            }
    
            Config config = Config.fromYAML(new ClassPathResource("redisson" + profile + ".yml").getInputStream());
            RedissonClient redisson = Redisson.create(config);
            return redisson;
        }
    
    
    }

Java中使用很简单直接调用就可以了

@Autowired
        private RedissonClient redissonClient;
        /** * 每天凌晨0点执行一次 */
     @Scheduled(cron = "0 0 0 */1 * ?")
      public void executeCountMonitorDataTask() { 
          log.info("*******执行记录状态标记任务*******");
          String lock_key = RedisKeyEnum.REDIS_LOCK_PUBLISH_TASK.getRedisPrefix();
          RLock lock = redissonClient.getLock(lock_key);
          try { 
              lock.lock();
              log.info("get redis lock success:{}",lock_key);
              execJob();
          } catch (Exception e) { 
    
          } finally { 
              lock.unlock();
              log.info("unlock redis lock success:{}",lock_key);
          }
    
    
      }

主要枷锁解锁  
redissonClient.getLock(lock\_key)  
lock.unlock();

## 参考资料 ##

图片来源:https://www.cnblogs.com/cjsblog/p/9831423.html

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]: /images/20221119/5a15fb5974e548e7aaf84dc60d24ce58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]: /images/20221119/3e54b276438841e3983319d645b9cba8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 2]: /images/20221119/be35da8018e147c7aec7013b109de2db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 3]: /images/20221119/7267c6f64f9c428c84941f97dd9b31d0.png