深入理解redis分布式鎖

哈嘍，大家好，我是指北君。
本篇文件我們來介紹如何Redis實現分布式鎖的演進過程，以及為什么不能直接用Setnx實現分布式鎖。

1、分布式鎖簡介

分布式鎖是控制分布式系統不同進程共同訪問共享資源的一種鎖的實現。如果不同的系統或同一個系統的不同主機之間共享了某個臨界資源，往往需要互斥來防止彼此干擾，以保證一致性。

業界流行的分布式鎖實現，一般有這3種方式：

• 基于數據庫實現的分布式鎖
• 基于Redis實現的分布式鎖
• 基于Zookeeper實現的分布式鎖

這里主要介紹如何通過 Redis 來實現分布式鎖。在介紹 Redis 分布式鎖之前，我們首先介紹一下實現Redis 分布式鎖的關鍵命令。

2、setnx

setnx key value

Setnx（SET if Not eXists） 命令在指定的 key 不存在時，為 key 設置指定的值。

設置成功，返回 1 。設置失敗，返回 0 。

PS：Redis 官方是不推薦基于 setnx 命令來實現分布式鎖的，因為會存在很多問題，

①、單點問題。比如：

• 1、客戶端A 從master拿到鎖lock01
• 2、master正要把lock01同步(Redis的主從同步通常是異步的）給slave時，突然宕機了，導致lock01沒同步給slave
• 3、主從切換，slave節點被晉級為master節點
• 4、客戶端B到master拿lock01照樣能拿到。這樣必將導致同一把鎖被多人使用。

②、鎖的高級用法，比如讀寫鎖、可重入鎖等等，setnx 都比較難實現。

這里先介紹基于 sentnx 實現的分布式鎖，后面會介紹官方推薦的基于 redisson 來實現分布式鎖。

3、Redis-分布式鎖-階段1

接到上文，查詢三級分類數據，如果我們部署了多個商品服務，然后多個線程同時去獲取三級分類數據，如果不加分布式鎖，就會導致，每一個部署的商品服務第一次查詢都會走 DB。

复制public Map< String, List< Catelog2Vo >> getCatelogJsonWithRedisLock() throws InterruptedException {
    // 一、獲取分布式鎖
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111");

    if(lock){
        // true 表示加鎖成功,執行相關業務
        Map< String, List< Catelog2Vo >> dataFromDb = getDataFromDb();
        stringRedisTemplate.delete("lock");
        return dataFromDb;
    }else{
        System.out.println("獲取分布式鎖失敗...等待重試...");
        //加鎖失敗...重試機制
        //休眠一百毫秒
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //自旋的方式
        return getCatelogJsonWithRedisLock();
    }
}

4、Redis-分布式鎖-階段2

設置鎖自動過期

复制public Map< String, List< Catelog2Vo >> getCatelogJsonWithRedisLock() throws InterruptedException {
    // 一、獲取分布式鎖
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111");

    if(lock){
        // true 表示加鎖成功,執行相關業務
        // 設置過期時間
        stringRedisTemplate.expire("lock",30,TimeUnit.SECONDS);
        Map< String, List< Catelog2Vo >> dataFromDb = getDataFromDb();
        stringRedisTemplate.delete("lock");
        return dataFromDb;
    }else{
        System.out.println("獲取分布式鎖失敗...等待重試...");
        //加鎖失敗...重試機制
        //休眠一百毫秒
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //自旋的方式
        return getCatelogJsonWithRedisLock();
    }
}

5、Redis-分布式鎖-階段3

setnx 命令和過期時間保證原子性。

复制public Map< String, List< Catelog2Vo >> getCatelogJsonWithRedisLock() throws InterruptedException {
    // 一、獲取分布式鎖
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111",30,TimeUnit.SECONDS);

    if(lock){
        // true 表示加鎖成功,執行相關業務
        // 設置過期時間
        //stringRedisTemplate.expire("lock",30,TimeUnit.SECONDS);
        Map< String, List< Catelog2Vo >> dataFromDb = getDataFromDb();
        stringRedisTemplate.delete("lock");
        return dataFromDb;
    }else{
        System.out.println("獲取分布式鎖失敗...等待重試...");
        //加鎖失敗...重試機制
        //休眠一百毫秒
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //自旋的方式
        return getCatelogJsonWithRedisLock();
    }
}

6、Redis-分布式鎖-階段4

保證刪除的是自己的鎖。

复制public Map< String, List< Catelog2Vo >> getCatelogJsonWithRedisLock() throws InterruptedException {
    // 一、獲取分布式鎖
    String uuid = UUID.randomUUID().toString();
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,30,TimeUnit.SECONDS);

    if(lock){
        // true 表示加鎖成功,執行相關業務 
        // 設置過期時間
        //stringRedisTemplate.expire("lock",30,TimeUnit.SECONDS);
        Map< String, List< Catelog2Vo >> dataFromDb = getDataFromDb();
        String lockValue = stringRedisTemplate.opsForValue().get("lock");
        if(uuid.equals(lockValue)){
            stringRedisTemplate.delete("lock");
        }
        return dataFromDb;
    }else{
        System.out.println("獲取分布式鎖失敗...等待重試...");
        //加鎖失敗...重試機制
        //休眠一百毫秒
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //自旋的方式
        return getCatelogJsonWithRedisLock();
    }
}

7、Redis-分布式鎖-階段5

通過Lua腳本保證刪除鎖和判斷鎖兩個操作原子性

复制public Map< String, List< Catelog2Vo >> getCatelogJsonWithRedisLock(){
    // 一、獲取分布式鎖
    String uuid = UUID.randomUUID().toString();
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,30,TimeUnit.SECONDS);

    if (lock) {
        System.out.println("獲取分布式鎖成功...");
        Map< String, List< Catelog2Vo >> dataFromDb = null;
        try {
            //加鎖成功...執行業務
            dataFromDb = getDataFromDb();
        } finally {
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

            //刪除鎖
            stringRedisTemplate.execute(new DefaultRedisScript< Long >(script, Long.class), Arrays.asList("lock"), uuid);

        }
        //先去redis查詢下保證當前的鎖是自己的
        //獲取值對比，對比成功刪除=原子性 lua腳本解鎖
        // String lockValue = stringRedisTemplate.opsForValue().get("lock");
        // if (uuid.equals(lockValue)) {
        //     //刪除我自己的鎖
        //     stringRedisTemplate.delete("lock");
        // }

        return dataFromDb;
    }else{
        System.out.println("獲取分布式鎖失敗...等待重試...");
        //加鎖失敗...重試機制
        //休眠一百毫秒
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //自旋的方式
        return getCatelogJsonWithRedisLock();
    }
}

這也是分布式鎖的最終模式，需要保證兩個點：加鎖【設置鎖+過期時間】和刪除鎖【判斷+刪除】原子性。