SpringBoot RedisTemplate分布式鎖的項目實戰(zhàn)

更新時間：2022年05月15日 09:43:15 作者：原味酸牛奶丶

本文主要介紹了SpringBoot RedisTemplate分布式鎖的項目實戰(zhàn)，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

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);
}

上述的操作看起來很正常,但是其實是有問題的，試想一下當(dāng)我們有兩個線程同時訪問這個接口會發(fā)生什么

Thread-1 查詢庫存結(jié)果為100

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

Thread-1 Set庫存為99

Thread-2 Set庫存為99

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

使用Idea時,我們可以使在斷點處右鍵將Suspend調(diào)整為Thread,僅阻斷線程,并使用多個客戶端同時請求接口,即可復(fù)現(xiàn)上述過程

多線程調(diào)試

2.加鎖解決

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

3.分布式鎖

上述兩個線程同時進行的時候沒有正確扣除庫存正是因為【查詢庫存】和【扣除庫存】不是一個原子操作,我們增加一個鎖的機制,當(dāng)線程持有鎖的時候才允許進行【查詢庫存】和【扣除庫存】，redis有一個sexNx命令允許當(dāng)指定的key不存在時才進行set操作,在java中為RedisTemplate的setIfAbsent方法,這個方法保證了同時只能有一個線程set成功,set成功時就表明我們拿到了鎖,可以進行原子操作了，當(dāng)我們執(zhí)行完原子操作時我們也需要將鎖釋放掉，在redis實現(xiàn)中也就是將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]設(shè)置鎖緩存 緩存  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("服務(wù)繁忙,稍后再試");
        }
        stock = RedisUtil.getObject("stock", Integer.class);
        if (stock > 0) {
            RedisUtil.set("stock", --stock);
        }
    } finally {
        RedisUtil.unlock("stock");
    }
    return R.ok(stock);
}

此時,我們再將斷點放在獲取庫存之后,并先用一個終端請求接口

終端1

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

終端2

4.增加失效時間

在上一步中,我們仿佛已經(jīng)完成了需求,同時進行扣除庫存的只有一個線程,但是試想一下,當(dāng)線程獲取到鎖之后,服務(wù)突然宕機了,這時候就算及時重啟機器,那么鎖也一直得不到釋放,那么扣除庫存接口始終無法獲取到鎖,這肯定不是我們想要的效果,那么我們改進一下我們加鎖的方法,增加一下失效時間,即使服務(wù)宕機了,我們重啟機器之后,鎖也能正常釋放掉不會影響一下個線程獲取到鎖

/**
     * 加鎖
     *
     * @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]設(shè)置鎖緩存 緩存  url:{} ========緩存時間為{}秒", key, time);
    }
    return result;
}

5.增加線程唯一值

還有一種情況會導(dǎo)致我們可能誤刪除別人的鎖,比如當(dāng)線程1執(zhí)行完流程之后準備釋放鎖之時,這時候鎖正好失效了,線程2此時獲取到鎖,線程1釋放鎖時并不知道鎖失效了,那么線程1執(zhí)行釋放操作就會將線程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("服務(wù)繁忙,稍后再試");
        }
        stock = RedisUtil.getObject("stock", Integer.class);
        if (stock > 0) {
            RedisUtil.set("stock", --stock);
        }
    } finally {
        // 在此釋放鎖時,判斷鎖是為自己持有才進行釋放
        RedisUtil.unlock("stock", uuid);
    }
    return R.ok(stock);
}

6.Lua腳本

上面我們說了為了防止誤刪別人的鎖,我們需要在刪除鎖時判斷一下鎖是否為自己持有,那么問題來了,我們這個查詢鎖值和刪除鎖的操作也并不是一個原子操作,也就是說可能你在獲取鎖值時鎖還為自己持有,但是執(zhí)行刪除時鎖已經(jīng)不為自己持有了,還是會可能誤刪別人的鎖,想要保證釋放鎖的原子性,我們可以通過redis原生支持的lua腳本來實現(xiàn)

/**
     * 解鎖
     *
     * @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是如何實現(xiàn)原子性的

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

Redis使用同一個Lua解釋器來執(zhí)行所有命令，同時，Redis保證以一種原子性的方式來執(zhí)行腳本：當(dāng)lua腳本在執(zhí)行的時候，不會有其他腳本和命令同時執(zhí)行，這種語義類似于 MULTI/EXEC。從別的客戶端的視角來看，一個lua腳本要么不可見，要么已經(jīng)執(zhí)行完。

然而這也意味著，執(zhí)行一個較慢的lua腳本是不建議的，由于腳本的開銷非常低，構(gòu)造一個快速執(zhí)行的腳本并非難事。但是你要注意到，當(dāng)你正在執(zhí)行一個比較慢的腳本時，所以其他的客戶端都無法執(zhí)行命令。

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]設(shè)置鎖緩存 緩存  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("服務(wù)繁忙,稍后再試");
        }
        stock = RedisUtil.getObject("stock", Integer.class);
        if (stock > 0) {
            RedisUtil.set("stock", --stock);
        }
    } finally {
        RedisUtil.unlock("stock", uuid);
    }
    return R.ok(stock);
}

9. 總結(jié)

分布式鎖在使用的過程中還是有挺多的講究的,主要看應(yīng)用場景例如還需要保證上述流程中可能碰到的鎖失效時間小于代碼執(zhí)行時間，鎖提前失效的問題，鎖如何保證重入性的問題,歡迎大家討論

到此這篇關(guān)于SpringBoot RedisTemplate分布式鎖的項目實戰(zhàn)的文章就介紹到這了,更多相關(guān)SpringBoot RedisTemplate分布式鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: