SpringBoot RedisTemplate分散式鎖的專案實戰

2022-05-15 13:00:01

1.使用場景

想直接獲取加鎖和解鎖程式碼,請直接到程式碼處

在下單場景減庫存時我們一般會將庫存查詢出來,進行庫存的扣除

@GetMapping(value = "order")
public R order() {
    int stock = RedisUtil.getObject("stock", Integer.class);
    if (stock > 0) {
        RedisUtil.set("stock", --stock);
    }
    return R.ok(stock);
}

上述的操作看起來很正常,但是其實是有問題的，試想一下當我們有兩個執行緒同時存取這個介面會發生什麼

Thread-1 查詢庫存結果為100

Thread-2 也來查詢庫存，此時Thread-1還沒有執行減少庫存操作,Thread-2 查詢庫存的結果也是100

Thread-1 Set庫存為99

Thread-2 Set庫存為99

這樣就出問題了,明天扣了兩次庫存，但是庫存僅僅減了1次

使用Idea時,我們可以使在斷點處右鍵將Suspend調整為Thread,僅阻斷執行緒,並使用多個使用者端同時請求介面,即可復現上述過程

2.加鎖解決

synchronized 我們可以用Java提供的synchronized關鍵字將方法分散式鎖,分散式鎖的實現方案有很多種, zookeeper，redis,db，這邊我們使用redis來實現以下分散式鎖

3.分散式鎖

上述兩個執行緒同時進行的時候沒有正確扣除庫存正是因為【查詢庫存】和【扣除庫存】不是一個原子操作,我們增加一個鎖的機制,當執行緒持有鎖的時候才允許進行【查詢庫存】和【扣除庫存】，redis有一個sexNx命令允許當指定的key不存在時才進行set操作,在java中為RedisTemplate的setIfAbsent方法,這個方法保證了同時只能有一個執行緒set成功,set成功時就表明我們拿到了鎖,可以進行原子操作了，當我們執行完原子操作時我們也需要將鎖釋放掉，在redis實現中也就是將key刪除，允許下一個執行緒set值,加鎖和釋放鎖的程式碼如下

/**
     * 加鎖
     *
     * @param key   redis主鍵
     * @param value 值
     */
public static boolean lock(String key, String value) {
    final boolean result = Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(CacheConstant.LOCK_KEY + key, value));
    if (result) {
        log.info("[redisTemplate redis]設定鎖快取 快取  url:{} ", key);
    }
    return result;
}

/**
     * 解鎖
     *
     * @param key redis主鍵
     */
public static boolean unlock(String key) {
    final boolean result = Boolean.TRUE.equals(redisTemplate.delete(CacheConstant.LOCK_KEY + key));
    if (result) {
        log.info("[redisTemplate redis]釋放鎖 快取  url:{}", key);
    }
    return result;
}

那麼我們將程式碼稍微修改一下,來利用鎖來完成介面的改進

@GetMapping(value = "order")
public R order() {
    boolean lock;
    int stock;
    try {
        lock = RedisUtil.lock("stock", "");
        if (!lock) {
            return R.failed("服務繁忙,稍後再試");
        }
        stock = RedisUtil.getObject("stock", Integer.class);
        if (stock > 0) {
            RedisUtil.set("stock", --stock);
        }
    } finally {
        RedisUtil.unlock("stock");
    }
    return R.ok(stock);
}

此時,我們再將斷點放在獲取庫存之後,並先用一個終端請求介面

然後,我們再從終端2發起請求，可以看到我們終端1沒有結束自己的原子操作時，終端2是無法進行庫存的扣除的

4.增加失效時間

在上一步中,我們彷彿已經完成了需求,同時進行扣除庫存的只有一個執行緒,但是試想一下,當執行緒獲取到鎖之後,服務突然宕機了,這時候就算及時重啟機器,那麼鎖也一直得不到釋放,那麼扣除庫存介面始終無法獲取到鎖,這肯定不是我們想要的效果,那麼我們改進一下我們加鎖的方法,增加一下失效時間,即使服務宕機了,我們重啟機器之後,鎖也能正常釋放掉不會影響一下個執行緒獲取到鎖

/**
     * 加鎖
     *
     * @param key   redis主鍵
     * @param value 值
     * @param time  過期時間
     */
public static boolean lock(String key, String value, long time) {
    final boolean result = Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(CacheConstant.LOCK_KEY + key, value, time, TimeUnit.SECONDS));
    if (result) {
        log.info("[redisTemplate redis]設定鎖快取 快取  url:{} ========快取時間為{}秒", key, time);
    }
    return result;
}

5.增加執行緒唯一值

還有一種情況會導致我們可能誤刪除別人的鎖,比如當執行緒1執行完流程之後準備釋放鎖之時,這時候鎖正好失效了,執行緒2此時獲取到鎖,執行緒1釋放鎖時並不知道鎖失效了,那麼執行緒1執行釋放操作就會將執行緒2擁有的鎖釋放掉，這肯定是不對的，那麼我們再對unlock方法改進一下

/**
     * 解鎖
     *
     * @param key redis主鍵
     */
public static boolean unlock(String key, String value) {
    if (Objects.equals(value, redisTemplate.opsForValue().get(CacheConstant.LOCK_KEY))) {
        final boolean result = Boolean.TRUE.equals(redisTemplate.delete(CacheConstant.LOCK_KEY + key));
        if (result) {
            log.info("[redisTemplate redis]釋放鎖 快取  url:{}", key);
        }
        return result;
    }
    return false;
}

@GetMapping(value = "order")
public R order() {
    boolean lock;
    int stock;
    String uuid = IdUtil.fastUUID();
    try {
        lock = RedisUtil.lock("stock", uuid, 60L);
        if (!lock) {
            return R.failed("服務繁忙,稍後再試");
        }
        stock = RedisUtil.getObject("stock", Integer.class);
        if (stock > 0) {
            RedisUtil.set("stock", --stock);
        }
    } finally {
        // 在此釋放鎖時,判斷鎖是為自己持有才進行釋放
        RedisUtil.unlock("stock", uuid);
    }
    return R.ok(stock);
}

6.Lua指令碼

上面我們說了為了防止誤刪別人的鎖,我們需要在刪除鎖時判斷一下鎖是否為自己持有,那麼問題來了,我們這個查詢鎖值和刪除鎖的操作也並不是一個原子操作,也就是說可能你在獲取鎖值時鎖還為自己持有,但是執行刪除時鎖已經不為自己持有了,還是會可能誤刪別人的鎖,想要保證釋放鎖的原子性,我們可以通過redis原生支援的lua指令碼來實現

/**
     * 解鎖
     *
     * @param key redis主鍵
     * @param value 值
     */
public static boolean unlock(String key, String value) {
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    RedisScript<Long> redisScript = new DefaultRedisScript<>(script, Long.class);
    Long result = redisTemplate.execute(redisScript, Collections.singletonList(CacheConstant.LOCK_KEY + key), value);
    if (Objects.equals(1L, result)) {
        log.info("[redisTemplate redis]釋放鎖 快取  url:{}", key);
        return true;
    }
    return false;
}

7.Lua是如何實現原子性的

可以看到Lua指令碼的大致意思也是跟我們自己寫的程式碼差不多，判斷是否為自己持有如果是才進行刪除，那為什麼Lua指令碼可以保證原子性呢

Redis使用同一個Lua直譯器來執行所有命令，同時，Redis保證以一種原子性的方式來執行指令碼：當lua指令碼在執行的時候，不會有其他指令碼和命令同時執行，這種語意類似於 MULTI/EXEC。從別的使用者端的視角來看，一個lua指令碼要麼不可見，要麼已經執行完。

然而這也意味著，執行一個較慢的lua指令碼是不建議的，由於指令碼的開銷非常低，構造一個快速執行的指令碼並非難事。但是你要注意到，當你正在執行一個比較慢的指令碼時，所以其他的使用者端都無法執行命令。

8.程式碼演示

程式碼演示

/**
     * 加鎖
     *
     * @param key   redis主鍵
     * @param value 值
     * @param time  過期時間
     */
public static boolean lock(String key, String value, long time) {
    final boolean result = Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(CacheConstant.LOCK_KEY + key, value, time, TimeUnit.SECONDS));
    if (result) {
        log.info("[redisTemplate redis]設定鎖快取 快取  url:{} ========快取時間為{}秒", key, time);
    }
    return result;
}

/**
     * 解鎖
     *
     * @param key redis主鍵
     * @param value 值
     */
public static boolean unlock(String key, String value) {
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    RedisScript<Long> redisScript = new DefaultRedisScript<>(script, Long.class);
    Long result = redisTemplate.execute(redisScript, Collections.singletonList(CacheConstant.LOCK_KEY + key), value);
    if (Objects.equals(1L, result)) {
        log.info("[redisTemplate redis]釋放鎖 快取  url:{}", key);
        return true;
    }
    return false;
}

@GetMapping(value = "order")
public R order() {
    boolean lock;
    int stock;
    String uuid = IdUtil.fastUUID();
    try {
        lock = RedisUtil.lock("stock", uuid,6000L);
        if (!lock) {
            return R.failed("服務繁忙,稍後再試");
        }
        stock = RedisUtil.getObject("stock", Integer.class);
        if (stock > 0) {
            RedisUtil.set("stock", --stock);
        }
    } finally {
        RedisUtil.unlock("stock", uuid);
    }
    return R.ok(stock);
}