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

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

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

1.使用場(chǎng)景

想直接獲取加鎖和解鎖代碼,請(qǐng)直接到代碼處

在下單場(chǎng)景減庫(kù)存時(shí)我們一般會(huì)將庫(kù)存查詢出來(lái),進(jìn)行庫(kù)存的扣除

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

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

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

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

Thread-1 Set庫(kù)存為99

Thread-2 Set庫(kù)存為99

這樣就出問(wèn)題了,明天扣了兩次庫(kù)存，但是庫(kù)存僅僅減了1次

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

多線程調(diào)試

2.加鎖解決

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

3.分布式鎖

上述兩個(gè)線程同時(shí)進(jìn)行的時(shí)候沒(méi)有正確扣除庫(kù)存正是因?yàn)椤静樵儙?kù)存】和【扣除庫(kù)存】不是一個(gè)原子操作,我們?cè)黾右粋€(gè)鎖的機(jī)制,當(dāng)線程持有鎖的時(shí)候才允許進(jìn)行【查詢庫(kù)存】和【扣除庫(kù)存】，redis有一個(gè)sexNx命令允許當(dāng)指定的key不存在時(shí)才進(jìn)行set操作,在java中為RedisTemplate的setIfAbsent方法,這個(gè)方法保證了同時(shí)只能有一個(gè)線程set成功,set成功時(shí)就表明我們拿到了鎖,可以進(jìn)行原子操作了，當(dāng)我們執(zhí)行完原子操作時(shí)我們也需要將鎖釋放掉，在redis實(shí)現(xiàn)中也就是將key刪除，允許下一個(gè)線程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;
}

那么我們將代碼稍微修改一下,來(lái)利用鎖來(lái)完成接口的改進(jìn)

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

此時(shí),我們?cè)賹帱c(diǎn)放在獲取庫(kù)存之后,并先用一個(gè)終端請(qǐng)求接口

終端1

然后,我們?cè)購(gòu)慕K端2發(fā)起請(qǐng)求，可以看到我們終端1沒(méi)有結(jié)束自己的原子操作時(shí)，終端2是無(wú)法進(jìn)行庫(kù)存的扣除的

終端2

4.增加失效時(shí)間

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

/**
     * 加鎖
     *
     * @param key   redis主鍵
     * @param value 值
     * @param time  過(guò)期時(shí)間
     */
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:{} ========緩存時(shí)間為{}秒", key, time);
    }
    return result;
}

5.增加線程唯一值

還有一種情況會(huì)導(dǎo)致我們可能誤刪除別人的鎖,比如當(dāng)線程1執(zhí)行完流程之后準(zhǔn)備釋放鎖之時(shí),這時(shí)候鎖正好失效了,線程2此時(shí)獲取到鎖,線程1釋放鎖時(shí)并不知道鎖失效了,那么線程1執(zhí)行釋放操作就會(huì)將線程2擁有的鎖釋放掉，這肯定是不對(duì)的，那么我們?cè)賹?duì)unlock方法改進(jìn)一下

/**
     * 解鎖
     *
     * @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 {
        // 在此釋放鎖時(shí),判斷鎖是為自己持有才進(jìn)行釋放
        RedisUtil.unlock("stock", uuid);
    }
    return R.ok(stock);
}

6.Lua腳本

上面我們說(shuō)了為了防止誤刪別人的鎖,我們需要在刪除鎖時(shí)判斷一下鎖是否為自己持有,那么問(wèn)題來(lái)了,我們這個(gè)查詢鎖值和刪除鎖的操作也并不是一個(gè)原子操作,也就是說(shuō)可能你在獲取鎖值時(shí)鎖還為自己持有,但是執(zhí)行刪除時(shí)鎖已經(jīng)不為自己持有了,還是會(huì)可能誤刪別人的鎖,想要保證釋放鎖的原子性,我們可以通過(guò)redis原生支持的lua腳本來(lái)實(shí)現(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是如何實(shí)現(xiàn)原子性的

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

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

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

8.代碼演示

代碼演示

/**
     * 加鎖
     *
     * @param key   redis主鍵
     * @param value 值
     * @param time  過(guò)期時(shí)間
     */
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:{} ========緩存時(shí)間為{}秒", 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é)

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

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

您可能感興趣的文章: