第七章【高階篇】分散式鎖之Redis6+Lua指令碼實現原生分散式鎖

阿新 • • 發佈：2021-06-21

第1集分散式核心技術-關於高併發下分散式鎖你知道多少？

簡介：分散式鎖核心知識介紹和注意事項

背景
- 就是保證同一時間只有一個客戶端可以對共享資源進行操作
- 案例：優惠券領劵限制張數、商品庫存超賣
- 核心
  - 為了防止分散式系統中的多個程序之間相互干擾，我們需要一種分散式協調技術來對這些程序進行排程
  - 利用互斥機制來控制共享資源的訪問，這就是分散式鎖要解決的問題
避免共享資源併發操作導致資料問題
- 加鎖
  - 本地鎖：synchronize、lock等，鎖在當前程序內，叢集部署下依舊存在問題
  - 分散式鎖：redis、zookeeper等實現，雖然還是鎖，但是多個程序共用的鎖標記，可以用Redis、Zookeeper、Mysql等都可以

設計分散式鎖應該考慮的東西
- 排他性
  - 在分散式應用叢集中，同一個方法在同一時間只能被一臺機器上的一個執行緒執行
- 容錯性
  - 分散式鎖一定能得到釋放，比如客戶端奔潰或者網路中斷
- 滿足可重入、高效能、高可用
- 注意分散式鎖的開銷、鎖粒度

第2集基於Redis實現分散式鎖的幾種坑你是否踩過《上》

簡介：基於Redis實現分散式鎖的幾種坑

實現分散式鎖可以用 Redis、Zookeeper、Mysql資料庫這幾種 , 效能最好的是Redis且是最容易理解
- 分散式鎖離不開 key - value 設定

key 是鎖的唯一標識，一般按業務來決定命名，比如想要給一種優惠券活動加鎖，key 命名為 “coupon:id” 。value就可以使用固定值，比如設定成1

基於redis實現分散式鎖，文件：http://www.redis.cn/commands.html#string

加鎖 SETNX key value

setnx 的含義就是 SET if Not Exists，有兩個引數 setnx(key, value)，該方法是原子性操作

如果 key 不存在，則設定當前 key 成功，返回 1；

如果當前 key 已經存在，則設定當前 key 失敗，返回 0

解鎖 del (key)

得到鎖的執行緒執行完任務，需要釋放鎖，以便其他執行緒可以進入,呼叫 del(key)

配置鎖超時 expire (key，30s）

客戶端奔潰或者網路中斷，資源將會永遠被鎖住,即死鎖，因此需要給key配置過期時間，以保證即使沒有被顯式釋放，這把鎖也要在一定時間後自動釋放

綜合虛擬碼

methodA(){

  String key = "coupon_66"

  if（setnx（key，1） == 1）{

    expire(key,30,TimeUnit.MILLISECONDS)

    try {

      //做對應的業務邏輯

      //查詢使用者是否已經領券

      //如果沒有則扣減庫存

      //新增領劵記錄

    } finally {

      del（key）

  }else{

   //睡眠100毫秒，然後自旋呼叫本方法

    methodA()

存在哪些問題，大家自行思考下

第3集基於Redis實現分散式鎖的幾種坑你是否踩過《下》

簡介：手把手教你徹底掌握分散式鎖+原生程式碼編寫

存在什麼問題？
- 多個命令之間不是原子性操作，如setnx和expire之間，如果setnx成功，但是expire失敗，且宕機了，則這個資源就是死鎖
```
使用原子命令：設定和配置過期時間  setnx / setex
```
```
如: set key 1 ex 30 nx
```
```
java裡面 redisTemplate.opsForValue().setIfAbsent("seckill_1","success",30,TimeUnit.MILLISECONDS)
```
- 業務超時，存在其他執行緒勿刪，key 30秒過期，假如執行緒A執行很慢超過30秒，則key就被釋放了，其他執行緒B就得到了鎖，這個時候執行緒A執行完成，而B還沒執行完成，結果就是執行緒A刪除了執行緒B加的鎖
```
可以在 del 釋放鎖之前做一個判斷，驗證當前的鎖是不是自己加的鎖, 那 value 應該是存當前執行緒的標識或者uuid
```
```
String key = "coupon_66"
```
```
String value = Thread.currentThread().getId()
```
```
if（setnx（key，value） == 1）{
```
```
   expire(key,30,TimeUnit.MILLISECONDS)
```
```
   try {
```
```
     //做對應的業務邏輯
```
```
   } finally {
```
```
     //刪除鎖，判斷是否是當前執行緒加的
```
```
     if(get(key).equals(value)){
```
```
          //還存在時間間隔
```
```
          del（key）
```
```
     }
```
```
   }
```
```
}else{
```
```
  //睡眠100毫秒，然後自旋呼叫本方法
```
```
}
```
- 進一步細化誤刪
  - 當執行緒A獲取到正常值時，返回帶程式碼中判斷期間鎖過期了，執行緒B剛好重新設定了新值，執行緒A那邊有判斷value是自己的標識，然後呼叫del方法，結果就是刪除了新設定的執行緒B的值
  - 核心還是判斷和刪除命令不是原子性操作導致
- 總結
  - 加鎖+配置過期時間：保證原子性操作
  - 解鎖: 防止誤刪除、也要保證原子性操作
- 那如何解決呢？下集講解

第4集手把手教你徹底掌握分散式鎖lua指令碼+redis原生程式碼編寫

簡介：手把手教你徹底掌握分散式鎖+原生程式碼編寫

前面說了redis做分散式鎖存在的問題

核心是保證多個指令原子性，加鎖使用setnx setex 可以保證原子性，那解鎖使用判斷和刪除怎麼保證原子性
文件：http://www.redis.cn/commands/set.html
多個命令的原子性：採用 lua指令碼+redis, 由於【判斷和刪除】是lua指令碼執行，所以要麼全成功，要麼全失敗

//獲取lock的值和傳遞的值一樣，呼叫刪除操作返回1，否則返回0

String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";

//Arrays.asList(lockKey)是key列表，uuid是引數

Integer result = redisTemplate.execute(new DefaultRedisScript<>(script, Integer.class), Arrays.asList(lockKey), uuid);

全部程式碼

/**

* 原生分散式鎖 開始

* 1、原子加鎖 設定過期時間，防止宕機死鎖

* 2、原子解鎖：需要判斷是不是自己的鎖

*/

@RestController

@RequestMapping("/api/v1/coupon")

public class CouponController {

   @Autowired

   private StringRedisTemplate redisTemplate;

   @GetMapping("add")

   public JsonData saveCoupon(@RequestParam(value = "coupon_id",required = true) int couponId){

     //防止其他執行緒誤刪

     String uuid = UUID.randomUUID().toString();

     String lockKey = "lock:coupon:"+couponId;

     lock(couponId,uuid,lockKey);

     return JsonData.buildSuccess();

   private void lock(int couponId,String uuid,String lockKey){

     //lua指令碼

     String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";

     Boolean nativeLock = redisTemplate.opsForValue().setIfAbsent(lockKey,uuid,Duration.ofSeconds(30));

     System.out.println(uuid+"加鎖狀態:"+nativeLock);

     if(nativeLock){

       //加鎖成功

       try{

         //TODO 做相關業務邏輯

         TimeUnit.SECONDS.sleep(10L);

       } catch (InterruptedException e) {

       } finally {

         //解鎖

         Long result = redisTemplate.execute( new DefaultRedisScript<>(script,Long.class),Arrays.asList(lockKey),uuid);

         System.out.println("解鎖狀態:"+result);

     }else {

       //自旋操作

       try {

         System.out.println("加鎖失敗，睡眠5秒 進行自旋");

         TimeUnit.MILLISECONDS.sleep(5000);

       } catch (InterruptedException e) { }

       //睡眠一會再嘗試獲取鎖

       lock(couponId,uuid,lockKey);

遺留一個問題，鎖的過期時間，如何實現鎖的自動續期或者避免業務執行時間過長，鎖過期了？
- 原生方式的話，一般把鎖的過期時間設定久一點，比如10分鐘時間
原生程式碼+redis實現分散式鎖使用比較複雜，且有些鎖續期問題更難處理
- 延伸出框架官方推薦方式：https://redis.io/topics/distlock
- 使用特別簡單: 看看公司做的工業級專案程式碼