网站推广.NET

网站推广.NET

Redis分布式锁该怎么实现续期

来源:互联网

Redis分布式锁如何续期

Redis分布式锁的正确姿势

据肥朝了解,很多同学在用分布式锁时,都是直接百度搜索找一个redis分布式锁工具类就直接用了.关键是该工具类中还充斥着很多system.out.println();等语句.其实redis分布式锁比较正确的姿势是采用redisson这个客户端工具.具体介绍可以搜索最大的同性交友网站github.

如何回答

首先如果你之前用Redis的分布式锁的姿势正确,并且看过相应的官方文档的话,这个问题So easy.我们来看

坦白说,如果你英文棒棒哒那么看英文文档可能更好理解

但是你如果看的是中文文档

这句话肥朝从语文角度分析就是一个歧义句,他有两个意思

看到这里,我希望大家不要黑我的小学体育老师,虽然他和语文老师是同个人.语文不行,我们可以源码来凑!

源码分析

我们根据官方文档给出的例子,写了一个最简单的demo,例子根据上面截图中Ctr+C和Ctr+V一波操作,如下

public class DemoMain {    public static void main(String[] args) throws Exception {        Config config = new Config();        config.useSingleServer().setAddress("redis://127.0.0.1:6379");        RedissonClient redisson = Redisson.create(config);        RLock lock = redisson.getLock("anyLock");        lock.lock();        //lock.unlock();    }}

create

从这里我们知道,internalLockLeaseTime 和 lockWatchdogTimeout这两个参数是相等的.

lockWatchdogTimeout默认值如下

public class Config {private long lockWatchdogTimeout = 30 * 1000;public long getLockWatchdogTimeout() {return lockWatchdogTimeout;}//省略无关代码}

从internalLockLeaseTime这个单词也可以看出,这个加的分布式锁的超时时间默认是30秒.但是还有一个问题,那就是这个看门狗,多久来延长一次有效期呢?我们往下看

lock

从我图中框起来的地方我们就知道了,获取锁成功就会开启一个定时任务,也就是watchdog,定时任务会定期检查去续期renewExpirationAsync(threadId).
这里定时用的是netty-common包中的HashedWheelTimer,肥朝公众号已经和各大搜索引擎建立了密切的合作关系,你只需要把这个类在任何搜索引擎一搜,都能知道相关API参数的意义.
从图中我们明白,该定时调度每次调用的时间差是internalLockLeaseTime / 3.也就10秒.

真相大白

通过源码分析我们知道,默认情况下,加锁的时间是30秒.如果加锁的业务没有执行完,那么到 30-10 = 20秒的时候,就会进行一次续期,把锁重置成30秒.那这个时候可能又有同学问了,那业务的机器万一宕机了呢?宕机了定时任务跑不了,就续不了期,那自然30秒之后锁就解开了呗.

Redis分布式锁的5个坑

一、锁未被释放

这种情况是一种低级错误,就是我上边犯的错,由于当前线程 获取到redis 锁,处理完业务后未及时释放锁,导致其它线程会一直尝试获取锁阻塞,例如:用Jedis客户端会报如下的错误信息

redis线程池已经没有空闲线程来处理客户端命令。

解决的方法也很简单,只要我们细心一点,拿到锁的线程处理完业务及时释放锁,如果是重入锁未拿到锁后,线程可以释放当前连接并且sleep一段时间。

public void lock() {    while (true) {        boolean flag = this.getLock(key);        if (flag) {              TODO .........        } else {              // 释放当前redis连接              redis.close();              // 休眠1000毫秒             sleep(1000);       }     } }

二、B的锁被A给释放了

我们知道Redis实现锁的原理在于 SETNX命令。当 key不存在时将 key的值设为 value ,返回值为 1;若给定的 key已经存在,则 SETNX不做任何动作,返回值为 0 。

SETNX key value

我们来设想一下这个场景:A、B两个线程来尝试给key myLock加锁,A线程先拿到锁(假如锁3秒后过期),B线程就在等待尝试获取锁,到这一点毛病没有。

那如果此时业务逻辑比较耗时,执行时间已经超过redis锁过期时间,这时A线程的锁自动释放(删除key),B线程检测到myLock这个key不存在,执行 SETNX命令也拿到了锁。

然而,即使A线程已经完成了业务逻辑,仍然会释放锁(即删除key),因此B线程的锁也会被A线程释放。

为避免上边的情况,一般我们在每个线程加锁时要带上自己独有的value值来标识,只释放指定value的key,否则就会出现释放锁混乱的场景。

三、数据库事务超时

emm~ 聊redis锁咋还扯到数据库事务上来了?别着急往下看,看下边这段代码:

 @Transaction public void lock() {      while (true) {          boolean flag = this.getLock(key);          if (flag) {              insert();          }      } }

给这个方法添加一个@Transaction注解开启事务,如代码中抛出异常进行回滚,要知道数据库事务可是有超时时间限制的,并不会无条件的一直等一个耗时的数据库操作。

比如:我们解析一个大文件,再将数据存入到数据库,如果执行时间太长,就会导致事务超时自动回滚。

一旦你的key长时间获取不到锁,获取锁等待的时间远超过数据库事务超时时间,程序就会报异常。

一般为解决这种问题,我们就需要将数据库事务改为手动提交、回滚事务。

  @Autowired  DataSourceTransactionManager dataSourceTransactionManager;  @Transaction  public void lock() {      //手动开启事务      TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);      try {          while (true) {             boolean flag = this.getLock(key);             if (flag) {                 insert();                 //手动提交事务                 dataSourceTransactionManager.commit(transactionStatus);             }         }     } catch (Exception e) {         //手动回滚事务         dataSourceTransactionManager.rollback(transactionStatus);     } }

四、锁过期了,业务还没执行完

这种情况和我们上边提到的第二种比较类似,但解决思路上略有不同。

同样是redis分布式锁过期,而业务逻辑没执行完的场景,不过,这里换一种思路想问题,把redis锁的过期时间再弄长点不就解决了吗?

那还是有问题,我们可以在加锁的时候,手动调长redis锁的过期时间,可这个时间多长合适?业务逻辑的执行时间是不可控的,调的过长又会影响操作性能。

要是redis锁的过期时间能够自动续期就好了。

为了解决这个问题我们使用redis客户端redisson,redisson很好的解决了redis在分布式环境下的一些棘手问题,它的宗旨就是让使用者减少对Redis的关注,将更多精力用在处理业务逻辑上。

redisson对分布式锁做了很好封装,只需调用API即可。

RLock lock = redissonClient.getLock("stockLock");

redisson在加锁成功后,会注册一个定时任务监听这个锁,每隔10秒就去查看这个锁,如果还持有锁,就对过期时间进行续期。默认过期时间30秒。这个机制也被叫做:“看门狗”,这名字。。。

举例子:假如加锁的时间是30秒,过10秒检查一次,一旦加锁的业务没有执行完,就会进行一次续期,把锁的过期时间再次重置成30秒。

通过分析下边redisson的源码实现可以发现,不管是加锁、解锁、续约都是客户端把一些复杂的业务逻辑,通过封装在Lua脚本中发送给redis,保证这段复杂业务逻辑执行的原子性。

@Slf4j@Servicepublic class RedisDistributionLockPlus {   /**    * 加锁超时时间,单位毫秒, 即:加锁时间内执行完操作,如果未完成会有并发现象    */   private static final long DEFAULT_LOCK_TIMEOUT = 30;  private static final long TIME_SECONDS_FIVE = 5 ;  /**   * 每个key的过期时间 {@link LockContent}   */  private Map<String, LockContent> lockContentMap = new ConcurrentHashMap<>(512);  /**   * redis执行成功的返回   */  private static final Long exec_SUCCESS = 1L;  /**   * 获取锁lua脚本, k1:获锁key, k2:续约耗时key, arg1:requestId,arg2:超时时间   */  private static final String LOCK_SCRIPT = "if redis.call(&#39;exists&#39;, KEYS[2]) == 1 then ARGV[2] = math.floor(redis.call(&#39;get&#39;, KEYS[2]) + 10) end " +          "if redis.call(&#39;exists&#39;, KEYS[1]) == 0 then " +             "local t = redis.call(&#39;set&#39;, KEYS[1], ARGV[1], &#39;EX&#39;, ARGV[2]) " +             "for k, v in pairs(t) do " +               "if v == &#39;OK&#39; then return tonumber(ARGV[2]) end " +             "end " +          "return 0 end";  /**   * 释放锁lua脚本, k1:获锁key, k2:续约耗时key, arg1:requestId,arg2:业务耗时 arg3: 业务开始设置的timeout   */  private static final String UNLOCK_SCRIPT = "if redis.call(&#39;get&#39;, KEYS[1]) == ARGV[1] then " +          "local ctime = tonumber(ARGV[2]) " +          "local biz_timeout = tonumber(ARGV[3]) " +          "if ctime > 0 then  " +             "if redis.call(&#39;exists&#39;, KEYS[2]) == 1 then " +                 "local avg_time = redis.call(&#39;get&#39;, KEYS[2]) " +                 "avg_time = (tonumber(avg_time) * 8 + ctime * 2)/10 " +                 "if avg_time >= biz_timeout - 5 then redis.call(&#39;set&#39;, KEYS[2], avg_time, &#39;EX&#39;, 24*60*60) " +                 "else redis.call(&#39;del&#39;, KEYS[2]) end " +             "elseif ctime > biz_timeout -5 then redis.call(&#39;set&#39;, KEYS[2], ARGV[2], &#39;EX&#39;, 24*60*60) end " +          "end " +          "return redis.call(&#39;del&#39;, KEYS[1]) " +          "else return 0 end";  /**   * 续约lua脚本   */  private static final String RENEW_SCRIPT = "if redis.call(&#39;get&#39;, KEYS[1]) == ARGV[1] then return redis.call(&#39;expire&#39;, KEYS[1], ARGV[2]) else return 0 end";  private final StringRedisTemplate redisTemplate;  public RedisDistributionLockPlus(StringRedisTemplate redisTemplate) {      this.redisTemplate = redisTemplate;      ScheduleTask task = new ScheduleTask(this, lockContentMap);      // 启动定时任务      ScheduleExecutor.schedule(task, 1, 1, TimeUnit.SECONDS);  }  /**   * 加锁   * 取到锁加锁,取不到锁一直等待知道获得锁   *   * @param lockKey   * @param requestId 全局唯一   * @param expire   锁过期时间, 单位秒   * @return   */  public boolean lock(String lockKey, String requestId, long expire) {      log.info("开始执行加锁, lockKey ={}, requestId={}", lockKey, requestId);      for (; ; ) {          // 判断是否已经有线程持有锁,减少redis的压力          LockContent lockContentOld = lockContentMap.get(lockKey);          boolean unLocked = null == lockContentOld;          // 如果没有被锁,就获取锁          if (unLocked) {              long startTime = System.currentTimeMillis();              // 计算超时时间              long bizExpire = expire == 0L ? DEFAULT_LOCK_TIMEOUT : expire;              String lockKeyRenew = lockKey + "_renew";              RedisScript<Long> script = RedisScript.of(LOCK_SCRIPT, Long.class);              List<String> keys = new ArrayList<>();              keys.add(lockKey);              keys.add(lockKeyRenew);              Long lockExpire = redisTemplate.execute(script, keys, requestId, Long.toString(bizExpire));              if (null != lockExpire && lockExpire > 0) {                  // 将锁放入map                  LockContent lockContent = new LockContent();                  lockContent.setStartTime(startTime);                  lockContent.setLockExpire(lockExpire);                  lockContent.setExpireTime(startTime + lockExpire * 1000);                  lockContent.setRequestId(requestId);                  lockContent.setThread(Thread.currentThread());                  lockContent.setBizExpire(bizExpire);                 lockContent.setLockCount(1);                 lockContentMap.put(lockKey, lockContent);                 log.info("加锁成功, lockKey ={}, requestId={}", lockKey, requestId);                 return true;             }         }         // 重复获取锁,在线程池中由于线程复用,线程相等并不能确定是该线程的锁         if (Thread.currentThread() == lockContentOld.getThread()                   && requestId.equals(lockContentOld.getRequestId())){             // 计数 +1             lockContentOld.setLockCount(lockContentOld.getLockCount()+1);             return true;         }         // 如果被锁或获取锁失败,则等待100毫秒         try {             TimeUnit.MILLISECONDS.sleep(100);         } catch (InterruptedException e) {             // 这里用lombok 有问题             log.error("获取redis 锁失败, lockKey ={}, requestId={}", lockKey, requestId, e);             return false;         }     } } /**  * 解锁  *  * @param lockKey  * @param lockValue  */ public boolean unlock(String lockKey, String lockValue) {     String lockKeyRenew = lockKey + "_renew";     LockContent lockContent = lockContentMap.get(lockKey);     long consumeTime;     if (null == lockContent) {         consumeTime = 0L;     } else if (lockValue.equals(lockContent.getRequestId())) {         int lockCount = lockContent.getLockCount();         // 每次释放锁, 计数 -1,减到0时删除redis上的key         if (--lockCount > 0) {             lockContent.setLockCount(lockCount);             return false;         }         consumeTime = (System.currentTimeMillis() - lockContent.getStartTime()) / 1000;     } else {         log.info("释放锁失败,不是自己的锁。");         return false;     }     // 删除已完成key,先删除本地缓存,减少redis压力, 分布式锁,只有一个,所以这里不加锁     lockContentMap.remove(lockKey);     RedisScript<Long> script = RedisScript.of(UNLOCK_SCRIPT, Long.class);     List<String> keys = new ArrayList<>();     keys.add(lockKey);     keys.add(lockKeyRenew);     Long result = redisTemplate.execute(script, keys, lockValue, Long.toString(consumeTime),             Long.toString(lockContent.getBizExpire()));     return EXEC_SUCCESS.equals(result); } /**  * 续约  *  * @param lockKey  * @param lockContent  * @return true:续约成功,false:续约失败(1、续约期间执行完成,锁被释放 2、不是自己的锁,3、续约期间锁过期了(未解决))  */ public boolean renew(String lockKey, LockContent lockContent) {     // 检测执行业务线程的状态     Thread.State state = lockContent.getThread().getState();     if (Thread.State.TERMINATED == state) {         log.info("执行业务的线程已终止,不再续约 lockKey ={}, lockContent={}", lockKey, lockContent);         return false;     }     String requestId = lockContent.getRequestId();     long timeOut = (lockContent.getExpireTime() - lockContent.getStartTime()) / 1000;     RedisScript<Long> script = RedisScript.of(RENEW_SCRIPT, Long.class);     List<String> keys = new ArrayList<>();     keys.add(lockKey);     Long result = redisTemplate.execute(script, keys, requestId, Long.toString(timeOut));     log.info("续约结果,True成功,False失败 lockKey ={}, result={}", lockKey, EXEC_SUCCESS.equals(result));     return EXEC_SUCCESS.equals(result); } static class ScheduleExecutor {     public static void schedule(ScheduleTask task, long initialDelay, long period, TimeUnit unit) {         long delay = unit.toMillis(initialDelay);         long period_ = unit.toMillis(period);         // 定时执行         new Timer("Lock-Renew-Task").schedule(task, delay, period_);     } } static class ScheduleTask extends TimerTask {     private final RedisDistributionLockPlus redisDistributionLock;     private final Map<String, LockContent> lockContentMap;     public ScheduleTask(RedisDistributionLockPlus redisDistributionLock, Map<String, LockContent> lockContentMap) {         this.redisDistributionLock = redisDistributionLock;         this.lockContentMap = lockContentMap;     }     @Override     public void run() {         if (lockContentMap.isEmpty()) {             return;         }         Set<Map.Entry<String, LockContent>> entries = lockContentMap.entrySet();         for (Map.Entry<String, LockContent> entry : entries) {             String lockKey = entry.getKey();             LockContent lockContent = entry.getValue();             long expireTime = lockContent.getExpireTime();             // 减少线程池中任务数量             if ((expireTime - System.currentTimeMillis())/ 1000 < TIME_SECONDS_FIVE) {                 //线程池异步续约                 ThreadPool.submit(() -> {                     boolean renew = redisDistributionLock.renew(lockKey, lockContent);                     if (renew) {                         long expireTimeNew = lockContent.getStartTime() + (expireTime - lockContent.getStartTime()) * 2 - TIME_SECONDS_FIVE * 1000;                         lockContent.setExpireTime(expireTimeNew);                     } else {                         // 续约失败,说明已经执行完 OR redis 出现问题                         lockContentMap.remove(lockKey);                    }                 });             }         }     } }}

五、redis主从复制的坑

redis高可用最常见的方案就是主从复制(master-slave),这种模式也给redis分布式锁挖了一坑。

redis cluster集群环境下,假如现在A客户端想要加锁,它会根据路由规则选择一台master节点写入key mylock,在加锁成功后,master节点会把key异步复制给对应的slave节点。

如果此时redis master节点宕机,为保证集群可用性,会进行主备切换,slave变为了redis master。A客户端错误地认为它在旧的master节点上成功加锁,但实际上锁已经被B客户端在新的master节点上加上了。

此时就会导致同一时间内多个客户端对一个分布式锁完成了加锁,导致各种脏数据的产生。

至于解决办法嘛,目前看还没有什么根治的方法,只能尽量保证机器的稳定性,减少发生此事件的概率。

小结一下:上面就是我在使用Redis 分布式锁时遇到的一些坑,有点小感慨,经常用一个方法填上这个坑,没多久就发现另一个坑又出来了,其实根本没有什么十全十美的解决方案,哪有什么银弹,只不过是在权衡利弊后,选一个在接受范围内的折中方案而已。

redis分布式锁