How to renew Redis distributed lock

How to renew Redis distributed lock

The correct posture of Redis distributed lock

According to Fei Chao’s understanding, when many students use distributed locks, they directly search on Baidu to find one The Redis distributed lock tool class is used directly. The key is that this tool class is also filled with many System.out.println(); and other statements. In fact, the more correct approach to Redis distributed lock is to use the redisson client tool. Specifically Introducing the largest gay dating website github that can be searched.

How to answer

First of all, if you have used Redis’s distributed lock correctly before and read the corresponding official documents, this question So easy. Let’s take a look

Frankly speaking, if your English is great, it may be better to understand by reading the English document

By default lock watchdog timeout is 30 seconds and can be changed through Config.lockWatchdogTimeout setting.

But if you are looking at the Chinese document

The default timeout for the watchdog check lock is 30 seconds

This sentence Fei Chao is an ambiguous sentence from a Chinese perspective. It has two meanings

1. The watchdog defaults to checking every 30 seconds Lock timeout time

2. The watchdog will check the lock timeout time. The default lock timeout time is 30 seconds

After seeing this, I hope everyone will not hack me. A primary school physical education teacher, although he and the Chinese teacher are the same person. If the Chinese language is not good, we can make up the source code!

Source code analysis

We wrote a final example based on the example given in the official document Simple demo, the example is based on the Ctr C and Ctr V wave operations in the screenshot above, as follows

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

From here We know that the two parameters internalLockLeaseTime and lockWatchdogTimeout are equal.

The default value of lockWatchdogTimeout is as follows

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

It can also be seen from the word internalLockLeaseTime that the default timeout time of this added distributed lock It is 30 seconds. But there is another question, that is, how often does this watchdog extend its validity period? Let’s look down

lock

##We know from the frame in my picture that successfully acquiring the lock will start a scheduled task, that is, watchdog. The scheduled task will periodically check to renew renewExpirationAsync(threadId).

Here The HashedWheelTimer in the netty-common package is used for timing. The Feichao official account has established close cooperative relationships with major search engines. You only need to search this class in any search engine to know the meaning of the relevant API parameters. .
We understand from the picture that the time difference between each call of the scheduled schedule is internalLockLeaseTime / 3. That is 10 seconds.

The truth is revealed

Through source code analysis, we know that by default Next, the locking time is 30 seconds. If the locked business has not been completed, then when 30-10 = 20 seconds, a renewal will be performed and the lock will be reset to 30 seconds. At this time, it may happen again Some students asked, what if the business machine goes down? If the machine goes down, the scheduled task cannot run, and the period cannot be renewed. Naturally, the lock will be unlocked after 30 seconds.

Redis Distributed 5 pitfalls of locks

1. The lock is not released

This situation is a low-level mistake, which is the mistake I made above. Since the current thread acquires the redis lock and completes the business The lock is not released in time, causing other threads to keep trying to obtain the lock and blocking. For example: using the Jedis client will report the following error message

redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool

The redis thread pool has no idle threads to process client commands.

The solution is also very simple. As long as we are careful, the thread that got the lock releases the lock in time after processing the business. If it is a reentrant lock but does not get the lock, the thread can release the current connection and sleep for a period of time. .

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

2. B’s lock is released by A

We know that the principle of Redis’ lock implementation lies in the SETNX command. When the key does not exist, the value of the key is set to value and the return value is 1; if the given key already exists, SETNX does not take any action and the return value is 0.

SETNX key value

Let’s imagine this scenario: Two threads A and B try to lock key myLock. Thread A gets the lock first (if the lock expires after 3 seconds), and thread B is waiting to try to acquire it. Lock, there is nothing wrong with it at this point.

If the business logic is time-consuming at this time and the execution time has exceeded the redis lock expiration time, then the lock of thread A is automatically released (key is deleted), and thread B detects that the key myLock does not exist and executes SETNX The command also got the lock.

However, even if thread A has completed the business logic, the lock will still be released (that is, the key will be deleted), so the lock of thread B will also be released by thread A.

In order to avoid the above situation, we generally need to bring its own unique value to identify each thread when locking, and only release the key with the specified value, otherwise there will be a chaotic scene of lock release.

三、数据库事务超时

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
@Service
public 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 分布式锁时遇到的一些坑，有点小感慨，经常用一个方法填上这个坑，没多久就发现另一个坑又出来了，其实根本没有什么十全十美的解决方案，哪有什么银弹，只不过是在权衡利弊后，选一个在接受范围内的折中方案而已。

The above is the detailed content of How to renew Redis distributed lock. For more information, please follow other related articles on the PHP Chinese website!