Comment renouveler le verrouillage distribué Redis

Comment renouveler le verrouillage distribué Redis

La posture correcte du verrouillage distribué Redis

Selon la compréhension de Fei Chao, de nombreux étudiants utilisent simplement des verrous distribués sur Baidu. pour trouver une classe d'outils de verrouillage distribué Redis et l'utiliser directement. La clé est que cette classe d'outils est également remplie de nombreuses instructions System.out.println(); utilisez les outils client de Redisson. Pour une introduction détaillée, vous pouvez rechercher le plus grand site de rencontres homosexuelles github

Comment répondre

Tout d'abord, si vous avez utilisé Redis. le verrou distribué correctement auparavant et je l'ai vu S'il existe des documents officiels correspondants, ce problème est si simple Jetons un coup d'œil à

Franchement, si. votre anglais est excellent, alors il serait peut-être préférable de lire des documents en anglais. Comprenez

Par défaut, le délai d'expiration du chien de garde du verrouillage est de 30 secondes et peut être modifié via le paramètre Config.lockWatchdogTimeout.

# 🎜🎜#

Mais si vous lisez un document chinois

Le délai d'expiration par défaut du verrouillage de vérification du chien de garde est de 30 secondes

Cette phrase, Fei Chao l'analyse d'un point de vue chinois et c'est une phrase ambiguë, il a deux significations

1 Le chien de garde par défaut est de 30 secondes pour vérifier le délai d'attente du verrouillage

.

2. Le chien de garde vérifiera le verrouillage Le délai d'attente et le temps de verrouillage par défaut sont de 30 secondes

Voyant cela, j'espère que tout le monde ne critiquera pas mon professeur d'éducation physique à l'école primaire , bien que lui et le professeur de chinois soient la même personne. Le chinois n'est pas bon , nous pouvons inventer le code source !

Analyse du code source

Nous avons écrit la démo la plus simple basée sur sur l'exemple donné dans le document officiel L'exemple est basé sur Ctr+C dans la capture d'écran ci-dessus Une vague d'opérations avec Ctr+V, comme suit

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.

De là, nous savons que les deux paramètres internalLockLeaseTime et lockWatchdogTimeout sont égaux

La valeur par défaut de lockWatchdogTimeout est la suivante.

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

On peut également voir à partir du mot internalLockLeaseTime que ce verrou distribué ajouté Le délai d'expiration par défaut est de 30 secondes. Mais il y a une autre question, c'est-à-dire à quelle fréquence le chien de garde prolonge-t-il sa validité ? vers le bas.

lock#🎜 🎜#

Nous savons d'après le cadre de ma photo que si le Le verrouillage est acquis avec succès, une tâche planifiée, c'est-à-dire un chien de garde, sera démarrée. La tâche planifiée sera vérifiée régulièrement. Accédez au renouvellement renouvelerExpirationAsync(threadId).

Le HashedWheelTimer dans le package netty-common est utilisé ici pour. timing. Le compte officiel Feichao a établi une relation de coopération étroite avec les principaux moteurs de recherche. Il vous suffit de mettre cette classe dans n'importe quel moteur de recherche. Une recherche sur le moteur de recherche révélera la signification des paramètres API pertinents. comprenez que le décalage horaire entre chaque appel du planning programmé est internalLockLeaseTime / 3. Cela fait 10 secondes

La vérité est révélée

Grâce à l'analyse du code source, nous savons que. par défaut, le temps de verrouillage est de 30 secondes. Si l'activité verrouillée n'est pas terminée, alors lorsque 30-10 = 20 secondes, il effectuera un renouvellement et réinitialisera le verrouillage à 30 secondes. À ce moment, certains étudiants peuvent demander à nouveau. , que se passe-t-il si la machine de l'entreprise tombe en panne, la tâche planifiée ne peut pas s'exécuter et la période ne peut pas être renouvelée, alors naturellement 30 secondes Le verrou sera déverrouillé après quelques secondes. C'est une erreur de bas niveau. est l'erreur que j'ai commise ci-dessus. Étant donné que le thread actuel a acquis le verrou Redis et n'a pas libéré le verrou à temps après le traitement de l'affaire, d'autres threads continueront d'essayer d'acquérir le verrou et de le bloquer. Par exemple : l'utilisation du client Jedis signalera le. Message d'erreur suivant

redis.clients.jedis.exceptions.JedisConnectionException : impossible d'obtenir une ressource du pool

redis le pool de threads n'est pas fil libre plus long pour gérer les commandes client.

La solution est également très simple. Tant qu'on fait attention, le thread qui a obtenu le verrou libère le verrou à temps après le traitement de l'affaire s'il s'agit d'un verrou réentrant et n'obtient pas le verrou. , le thread peut libérer la connexion actuelle et dormir un certain temps.

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

2. Le verrou de B est libéré par A

Nous savons que le principe de l'implémentation du verrou de Redis réside dans la commande SETNX. Lorsque la clé n'existe pas, la valeur de la clé est définie sur value et la valeur de retour est 1 ; si la clé donnée existe déjà, SETNX n'effectue aucune action et la valeur de retour est 0.

SETNX key value

Imaginons ce scénario : deux threads A et B tentent de verrouiller la clé myLock. Le thread A obtient le verrou en premier (si le verrou expire après 3 secondes), et le thread B attend d'essayer d'acquérir. la serrure, et il n'y a aucun problème à ce stade.

Si la logique métier prend du temps à ce moment-là et que le temps d'exécution a dépassé le délai d'expiration du verrouillage Redis, alors le verrou du thread A est automatiquement libéré (la clé est supprimée) et le thread B détecte que la clé myLock n'existe pas. L'exécution de la commande SETNX a également obtenu le verrou.

Cependant, même si le thread A a terminé la logique métier, le verrou sera toujours libéré (c'est-à-dire que la clé sera supprimée), donc le verrou du thread B sera également libéré par le thread A .

Afin d'éviter la situation ci-dessus, généralement lorsque nous verrouillons chaque thread, nous devons apporter sa propre valeur unique pour l'identifier, et relâcher uniquement la clé avec la valeur spécifiée, sinon il y aura confusion en libérant la scène de verrouillage.

三、数据库事务超时

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

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!