快捷導(dǎo)航

基于redis實現(xiàn)分布式鎖的原理與方法

更新時間：2020年06月02日 08:32:57 作者：煙花散盡13141

這篇文章主要給大家介紹了基于redis實現(xiàn)分布式鎖的原理與方法，文中通過示例代碼介紹的非常詳細，對大家學(xué)習(xí)或者使用redis具有一定的參考學(xué)習(xí)價值，需要的朋友們下面來一起學(xué)習(xí)學(xué)習(xí)吧

前言

系統(tǒng)的不斷擴大，分布式鎖是最基本的保障。與單機的多線程不一樣的是，分布式跨多個機器。線程的共享變量無法跨機器。

為了保證一個在高并發(fā)存場景下只能被同一個線程操作，java并發(fā)處理提供ReentrantLock或Synchronized進行互斥控制。但是這僅僅對單機環(huán)境有效。我們實現(xiàn)分布式鎖大概通過三種方式。

redis實現(xiàn)分布式鎖
數(shù)據(jù)庫實現(xiàn)分布式鎖
zk實現(xiàn)分布式鎖

今天我們介紹通過redis實現(xiàn)分布式鎖。實際上這三種和java對比看屬于一類。都是屬于程序外部鎖。

原理剖析

上述三種分布式鎖都是通過各自為依據(jù)對各個請求進行上鎖，解鎖從而控制放行還是拒絕。redis鎖是基于其提供的setnx命令。
setnx當(dāng)且僅當(dāng)key不存在。若給定key已經(jīng)存在，則setnx不做任何動作。setnx是一個原子性操作。
和數(shù)據(jù)庫分布式相比，因為redis內(nèi)存輕量。所以redis分布式鎖性能更好

實現(xiàn)

原理很簡單。結(jié)合springboot項目我們實現(xiàn)一套通過注解形式對接口進行庫存上鎖案例進行理解

編寫注解

我們編寫注解。方便我們在接口上添加注解提供攔截信息

/**
 * @author 張新華
 * @version V1.0
 * @Package com.ay.framework.order.redis.product
 * @date 2020年03月26日, 0026 10:29
 * @Copyright © 2020 安元科技有限公司
 */
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface StockLock {

  /**
   * @author zxhtom
   * @Description 鎖key的前綴
   * @Date 15:25 2020年03月25日, 0025
   * @Param []
   * @return java.lang.String
   */
  String prefix() default "";
  /**
   * @author zxhtom
   * @Description key的分隔符
   * @Date 15:27 2020年03月25日, 0025
   * @Param []
   * @return java.lang.String
   */
  String delimiter() default ":";
}

/**
 * @author 張新華
 * @version V1.0
 * @Package com.ay.framework.order.redis.product
 * @date 2020年03月26日, 0026 11:09
 * @Copyright © 2020 安元科技有限公司
 */
@Target({ElementType.PARAMETER , ElementType.METHOD , ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface StockParam {
  /**
  * @author zxhtom
  * @Description 組成key
  * @Date 11:11 2020年03月26日, 0026
  * @Param []
  * @return java.lang.String[]
  */
  String[] names() default {""};
}

攔截器攔截

redis分布式鎖實現(xiàn)的關(guān)鍵就是攔截器的編寫。上面的注解只是為了實現(xiàn)攔截的一個輔助。

@Around("execution(public * *(..)) && @annotation(com.ay.framework.order.redis.product.StockLock)")

通過springboot的Around進行針對StockLock注解的攔截。通過攔截我們可以獲取到攔截的方法、參數(shù)、及需要的鎖的參數(shù)。

我們獲取到需要鎖的名稱這里叫做【a】之后通過redis的原子性操作對該key進行遞減操作。

為了方便我們在削減庫存的時候可以對庫存進行更新操作。我們在遞減庫存前還需要借助于另一把鎖。這一把鎖我們叫做【a_key】

換句話說我們接口想訪問就必須獲取【a】鎖，拿到【a】鎖需要減少庫存。減少庫存之前需要獲取【a_key】鎖。

拿到鎖之后處理完邏輯之后我們需要釋放對應(yīng)鎖。

RedisAtomicLong entityIdCounter = new RedisAtomicLong(lockKey, redisTemplate.getConnectionFactory());
  if (redisTemplate.hasKey(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey)) {
    //表示lockKey的庫存信息有變動。此時無法進行交易
    throw new BusinessException("庫存變動。暫無法交易");
  }
  Long increment = entityIdCounter.decrementAndGet();
  if (increment >= 0) {
    try {
      Object proceed = pjp.proceed();
    } catch (Throwable throwable) {
      //所占資源需要釋放回資源池
      while (!redisLock.tryGetLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey, "")) {

      }
      //表示lockKey的庫存信息有變動。此時無法進行交易
      long l = entityIdCounter.incrementAndGet();
      if (l < 1) {
        redisTemplate.opsForValue().set(lockKey,1);
      }
      redisLock.unLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey);
      throwable.printStackTrace();
    }
  } else {
    redisTemplate.opsForValue().set(lockKey,0);
    throw new BusinessException("庫存不足！無法操作");
  }

因為我們上鎖就需要釋放鎖。但是程序在中途處理業(yè)務(wù)是發(fā)生異常導(dǎo)致沒有走到釋放鎖的步驟。這個時候就導(dǎo)致我們的分布式鎖一直被鎖。俗稱【死鎖】。為了避免這種場景的發(fā)生。我們常常在上鎖的時候給一個有效期。有效期已過自動釋放鎖。這個特性恰好和redis的過期策略不摩爾和。

上述提及工具

RedisLock

public Boolean tryGetLock(String key , String value) {
  return tryGetLock(key, value, -1, TimeUnit.DAYS);
}
public Boolean tryGetLock(String key , String value, Integer expire) {
  return tryGetLock(key, value, expire, TimeUnit.SECONDS);
}
public Boolean tryGetLock(String key , String value, Integer expire , TimeUnit timeUnit) {
  ValueOperations operations = redisTemplate.opsForValue();
  if (operations.setIfAbsent(key, value)) {
    //說明 redis沒有該key , 換言之 加鎖成功 設(shè)置過期時間防止死鎖
    if (expire > 0) {
      redisTemplate.expire(key, expire, timeUnit);
    }
    return true;
  }
  return false;
}

public Boolean unLock(String key) {
  return redisTemplate.delete(key);
}

StockKeyGenerator

@Component()
@Primary
public class StockKeyGenerator implements CacheKeyGenerator {
  @Override
  public String getLockKey(ProceedingJoinPoint pjp) {
    //獲取方法簽名
    MethodSignature signature = (MethodSignature) pjp.getSignature();
    Method method = signature.getMethod();
    //獲取方法cacheLock注解
    StockLock stockLock = method.getAnnotation(StockLock.class);
    //獲取方法參數(shù)
    Object[] args = pjp.getArgs();
    Parameter[] parameters = method.getParameters();
    StringBuilder builder = new StringBuilder();
    for (int i = 0; i < parameters.length; i++) {
      StockParam stockParam = parameters[i].getAnnotation(StockParam.class);
      Object arg = args[i];
      if (arg instanceof Map) {
        Map<String, Object> temArgMap = (Map<String, Object>) arg;
        String[] names = stockParam.names();
        for (String name : names) {
          if (builder.length() > 0) {
            builder.append(stockLock.delimiter());
          }
          builder.append(temArgMap.get(name));
        }
      }

    }
    return builder.toString();
  }
}

問題分析

上面分析了一個死鎖的場景，理論上出了死鎖我們redis分布鎖很好的解決了分布式問題。但是還是會出現(xiàn)問題。下面列舉寫小編遇到的問題。

業(yè)務(wù)處理時間>上鎖過期時間

a線程獲取到鎖，開始進行業(yè)務(wù)處理需要8S，

在8S內(nèi)，鎖的有效期是5S，在鎖過期后也就是第6S ， b線程進入開始獲取鎖這個時候b是可以獲取到新鎖的。這個時候就是有問題的。

假設(shè)b線程業(yè)務(wù)處理只需要3S ，但是因為a線程釋放了鎖，所以在第8S的時候雖然b線程沒有釋放鎖，b的鎖也沒有過期但是這時候也沒有了鎖。從而導(dǎo)致C線程也可以進入

總結(jié)

到此這篇基于redis實現(xiàn)分布式鎖的原理與方法的文章就介紹到這了,更多相關(guān)redis分布式鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: