Redis 分布式鎖

一、分布式鎖概念

隨著業務發展的需要，原單機部署的系統被演化成分布式集群系統後，由於分布式系統多線程、多進程並且分布在不同機器上，這將使原單機部署情况下的並發控制鎖策略失效，單純的 Java API 並不能提供分布式鎖的能力。為了解决這個問題就需要一種跨 JVM 的互斥機制來控制共享資源的訪問，這就是分布式鎖要解决的問題！

說得通俗些，集群中上了鎖後，無論當前操作在哪臺機器，所有的機器都會識別並且等待，鎖釋放後其他操作才能進行，這就是分布式鎖，對所有集群裏都有效

分布式鎖主流的實現方案：

1. 基於數據庫實現分布式鎖
2. 基於緩存（Redis 等）
3. 基於 Zookeeper

每一種分布式鎖解决方案都有各自的優缺點，其中redis性能最高zookeeper可靠性最高

二、使用setnx實現鎖

set stu:1:info “OK” nx px 10000

• EX second ：設置鍵的過期時間為 second 秒,，SET key value EX second 效果等同於 SETEX key second value

• PX millisecond ：設置鍵的過期時間為 millisecond 毫秒，SET key value PX millisecond 效果等同於 PSETEX key millisecond value

• NX ：只在鍵不存在時，才對鍵進行設置操作，SET key value NX 效果等同於 SETNX key value

• XX ：只在鍵已經存在時，才對鍵進行設置操作

• 多個客戶端同時獲取鎖（setnx）
• 獲取成功，執行業務邏輯（從 db 獲取數據，放入緩存），執行完成釋放鎖（del）
• 獲取失敗的客戶端則等待重試

用setnx和del添加以及釋放鎖

一般地，我們需要給鎖設置過期時間防止鎖被長期占用

這裏有個問題：加鎖和設置過期時間是兩個操作，而不是同時進行操作的，如果上鎖後發生异常情况，就無法設置過期時間了。我們可以上鎖的同時設置過期時間

三、編寫代碼測試分布式鎖

1. 使用Java代碼測試分布式鎖

首先在redis中設置num的值為0，編寫Java代碼進行測試

下方代碼做的就是：獲取到鎖則num++，並釋放鎖；沒獲取到則0.1秒後重新獲取

重啟，服務集群，通過網關壓力測試：ab -n 5000 -c 100 http://192.168.140.1:8080/test/testLock

查看 redis 中 num 的值

問題： setnx 剛好獲取到鎖，業務邏輯出現异常，導致鎖無法釋放
解决： 設置過期時間，自動釋放鎖

2. 優化之設置鎖的過期時間

設置過期時間有兩種方式：

• 首先想到通過 expire 設置過期時間（缺乏原子性：如果在 setnx 和 expire 之 間出現异常，鎖也無法釋放）
• 在 set 的同時指定過期時間（推薦）

代碼中設置過期時間：

問題： 可能會釋放其他服務器的鎖

如果業務邏輯的執行時間是 7s，執行流程如下：

• index1 業務邏輯沒執行完，3 秒後鎖被自動釋放
• index2 獲取到鎖，執行業務邏輯，3 秒後鎖被自動釋放
• index3 獲取到鎖，執行業務邏輯
• index1 業務邏輯執行完成，開始調用 del 釋放鎖，這時釋放的是 index3 的鎖， 導致 index3 的業務只執行 1s 就被別人釋放。
  最終等於沒鎖的情况

a在操作時卡頓了，導致鎖超時後自動釋放；釋放後，b搶到鎖進行操作；此時a操作完成，手動釋放鎖，這就把b的鎖給釋放了，b再釋放鎖則會報錯

解决： setnx 獲取鎖時，設置一個指定的唯一值（例如：uuid）；釋放前獲取這 個值，判斷是否自己的鎖

四、優化之給lock設置UUID防誤删

五、使用LUA脚本保證删除的原子性

使用lock的uuid可以一定程度上緩解線程釋放其他鎖，但並不能完全解决這種情况。因為比較uuid和删除lock並不是原子性的

問題： a比較uuid通過後，鎖到期了自動釋放，b重新加鎖，a此時會手動釋放b的鎖，這還是出現問題

解决： 使用LUA 脚本保證删除的原子性

LUA脚本：

• 將複雜的或者多步的 redis 操作，寫為一個脚本，一次提交給 redis 執行，减少反複連接 redis 的次數，提昇性能

• LUA 脚本是類似 redis 事務，有一定的原子性，不會被其他命令插隊，可以完成一些redis 事務性的

@GetMapping("testLockLua")
public void testLockLua() {
    
    //1 聲明一個 uuid ,將做為一個 value 放入我們的 key 所對應的值中
    String uuid = UUID.randomUUID().toString();
    //2 定義一個鎖：lua 脚本可以使用同一把鎖，來實現删除！
    String skuId = "25"; // 訪問 skuId 為 25 號的商品 100008348542
    String locKey = "lock:" + skuId; // 鎖住的是每個商品的數據
    // 3 獲取鎖
    Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS);
    // 第一種： lock 與過期時間中間不寫任何的代碼。
    // redisTemplate.expire("lock",10, TimeUnit.SECONDS);//設置過期時間
    // 如果 true
    if (lock) {
    
        // 執行的業務邏輯開始
        // 獲取緩存中的 num 數據
        Object value = redisTemplate.opsForValue().get("num");
        // 如果是空直接返回
        if (StringUtils.isEmpty(value)) {
    
            return;
        }
        // 不是空 如果說在這出現了异常！ 那麼 delete 就删除失敗！ 也就是說鎖永遠存在！
        int num = Integer.parseInt(value + "");
        // 使 num 每次+1 放入緩存
        redisTemplate.opsForValue().set("num", String.valueOf(++num));
        /*使用 lua 脚本來鎖*/
        // 定義 lua 脚本：將判斷和删除操作同時進行
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        // 使用 redis 執行 lua 執行
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(script);
        // 設置一下返回值類型 為 Long
        // 因為删除判斷的時候，返回的 0,給其封裝為數據類型。如果不封裝那麼默認返回 String 類型，
        // 那麼返回字符串與 0 會有發生錯誤。
        redisScript.setResultType(Long.class);
        // 第一個是執行的 script 脚本 ，第二個需要判斷的 key，第三個就是 key 所對應的值。
        redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
        } else {
    
        // 其他線程等待
        try {
    
            // 睡眠
            Thread.sleep(1000);
            // 睡醒了之後，調用方法。
            testLockLua();
        } catch (InterruptedException e) {
    
            e.printStackTrace();
        }
    }
}

為了確保分布式鎖可用，我們至少要確保鎖的實現同時滿足以下四個條件：

• 互斥性；在任意時刻，只有一個客戶端能持有鎖
• 不會發生死鎖；即使有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證後續其他客戶端能加鎖（設置lock的過期時間）
• 解鈴還須系鈴人；加鎖和解鎖必須是同一個客戶端，客戶端自己不能把別人加的鎖給解了（使用LUA脚本和uuid）
• 加鎖和解鎖必須具有原子性（使用LUA脚本）