真实的国产乱ⅩXXX66竹夫人,五月香六月婷婷激情综合,亚洲日本VA一区二区三区,亚洲精品一区二区三区麻豆

成都創(chuàng)新互聯(lián)網(wǎng)站制作重慶分公司

如何用Redis實(shí)現(xiàn)分布式鎖

這篇文章將為大家詳細(xì)講解有關(guān)如何用redis實(shí)現(xiàn)分布式鎖，小編覺(jué)得挺實(shí)用的，因此分享給大家做個(gè)參考，希望大家閱讀完這篇文章后可以有所收獲。

創(chuàng)新互聯(lián)專業(yè)為企業(yè)提供安順網(wǎng)站建設(shè)、安順做網(wǎng)站、安順網(wǎng)站設(shè)計(jì)、安順網(wǎng)站制作等企業(yè)網(wǎng)站建設(shè)、網(wǎng)頁(yè)設(shè)計(jì)與制作、安順企業(yè)網(wǎng)站模板建站服務(wù)，十年安順做網(wǎng)站經(jīng)驗(yàn)，不只是建網(wǎng)站，更提供有價(jià)值的思路和整體網(wǎng)絡(luò)服務(wù)。

啥是分布式鎖？

我們學(xué)習(xí) Java 都知道鎖的概念，例如基于 JVM 實(shí)現(xiàn)的同步鎖 synchronized，以及 jdk 提供的一套代碼級(jí)別的鎖機(jī)制 lock，我們?cè)诓l(fā)編程中會(huì)經(jīng)常用這兩種鎖去保證代碼在多線程環(huán)境下運(yùn)行的正確性。但是這些鎖機(jī)制在分布式場(chǎng)景下是不適用的，原因是在分布式業(yè)務(wù)場(chǎng)景下，我們的代碼都是跑在不同的JVM甚至是不同的機(jī)器上，synchronized 和 lock 只能在同一個(gè) JVM 環(huán)境下起作用。所以這時(shí)候就需要用到分布式鎖了。

例如，現(xiàn)在有個(gè)場(chǎng)景就是整點(diǎn)搶消費(fèi)券（疫情的原因，支付寶最近在8點(diǎn)、12點(diǎn)整點(diǎn)開(kāi)放搶消費(fèi)券），消費(fèi)券有一個(gè)固定的量，先到先得，搶完就沒(méi)了，線上的服務(wù)都是部署多個(gè)的，大致架構(gòu)如下：

 如何用Redis實(shí)現(xiàn)分布式鎖

 所以這個(gè)時(shí)候我們就得用分布式鎖來(lái)保證共享資源的訪問(wèn)的正確性。

回到頂部

為什么要用分布式鎖嗯？

假設(shè)不使用分布式鎖，我們看看 synchronized 能不能保證？其實(shí)是不能的，我們來(lái)演示一下。

下面我寫(xiě)了一個(gè)簡(jiǎn)單的 springboot 項(xiàng)目來(lái)模擬這個(gè)搶消費(fèi)券的場(chǎng)景，代碼很簡(jiǎn)單，大致意思是先從 Redis 獲取剩余消費(fèi)券數(shù)，然后判斷大于0，則減一模擬被某個(gè)用戶搶到一個(gè)，然后減一后再修改 Redis 的剩余消費(fèi)券數(shù)量，打印扣減成功，剩余還有多少，否則扣減失敗，就沒(méi)搶到。整塊代碼被 synchronized 包裹，Redis 設(shè)置的庫(kù)存數(shù)量為50。

//假設(shè)庫(kù)存編號(hào)是00001private String key = "stock:00001";
@Autowiredprivate StringRedisTemplate stringRedisTemplate;/**
 * 扣減庫(kù)存 synchronized同步鎖*/@RequestMapping("/deductStock")public String deductStock(){    synchronized (this){        //獲取當(dāng)前庫(kù)存
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));        if(stock>0){            int afterStock = stock-1;
            stringRedisTemplate.opsForValue().set(key,afterStock+"");//修改庫(kù)存
            System.out.println("扣減庫(kù)存成功，剩余庫(kù)存"+afterStock);
        }else {
            System.out.println("扣減庫(kù)存失敗");
        }
    }    return "ok";
}

然后啟動(dòng)兩個(gè)springboot項(xiàng)目，端口分別為8080,8081，然后在nginx里配置負(fù)載均衡

upstream redislock{
    server 127.0.0.1:8080;
    server 127.0.0.1:8081;
}
server {
    listen       80;
    server_name  127.0.0.1;
    location / {
        root   html;
        index  index.html index.htm;
        proxy_pass http://redislock;    }
}

 然后用jmeter壓測(cè)工具進(jìn)行測(cè)試

 如何用Redis實(shí)現(xiàn)分布式鎖

 如何用Redis實(shí)現(xiàn)分布式鎖

然后我們看一下控制臺(tái)輸出，可以看到我們運(yùn)行的兩個(gè)web實(shí)例，很多同樣的消費(fèi)券被不同的線程搶到，證明synchronized在這樣的情況下是不起作用的，所以就需要使用分布式鎖來(lái)保證資源的正確性。

 如何用Redis實(shí)現(xiàn)分布式鎖

回到頂部

如何用Redis實(shí)現(xiàn)分布式鎖？

在實(shí)現(xiàn)分布式鎖之前，我們先考慮如何實(shí)現(xiàn)，以及都要實(shí)現(xiàn)鎖的哪些功能。

1、分布式特性（部署在多個(gè)機(jī)器上的實(shí)例都能夠訪問(wèn)這把鎖）

2、排他性（同一時(shí)間只能有一個(gè)線程持有鎖）

3、超時(shí)自動(dòng)釋放的特性（持有鎖的線程需要給定一定的持有鎖的最大時(shí)間，防止線程死掉無(wú)法釋放鎖而造成死鎖）

4、...

基于以上列出的分布式鎖需要擁有的基本特性，我們思考一下使用Redis該如何實(shí)現(xiàn)？

1、第一個(gè)分布式的特性Redis已經(jīng)支持，多個(gè)實(shí)例連同一個(gè)Redis即可

2、第二個(gè)排他性，也就是要實(shí)現(xiàn)一個(gè)獨(dú)占鎖，可以使用Redis的setnx命令實(shí)現(xiàn)

3、第三個(gè)超時(shí)自動(dòng)釋放特性，Redis可以針對(duì)某個(gè)key設(shè)置過(guò)期時(shí)間

4、執(zhí)行完畢釋放分布式鎖

科普時(shí)間

Redis Setnx 命令

Redis Setnx（SET if Not eXists） 命令在指定的 key 不存在時(shí)，為 key 設(shè)置指定的值

語(yǔ)法

redis Setnx 命令基本語(yǔ)法如下：

redis 127.0.0.1:6379> SETNX KEY_NAME VALUE

可用版本：>= 1.0.0

返回值：設(shè)置成功，返回1， 設(shè)置失敗，返回0

@RequestMapping("/stock_redis_lock")public String stock_redis_lock(){    //底層使用setnx命令
    Boolean aTrue = stringRedisTemplate.opsForValue().setIfAbsent(lock_key, "true");
    stringRedisTemplate.expire(lock_key,10, TimeUnit.SECONDS);//設(shè)置過(guò)期時(shí)間10秒
    if (!aTrue) {//設(shè)置失敗則表示沒(méi)有拿到分布式鎖
        return "error";//這里可以給用戶一個(gè)友好的提示    }    //獲取當(dāng)前庫(kù)存
    int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));    if(stock>0){        int afterStock = stock-1;
        stringRedisTemplate.opsForValue().set(key,afterStock+"");
        System.out.println("扣減庫(kù)存成功，剩余庫(kù)存"+afterStock);
    }else {
        System.out.println("扣減庫(kù)存失敗");
    }
    stringRedisTemplate.delete(lock_key);//執(zhí)行完畢釋放分布式鎖
    return "ok";
}

仍然設(shè)置庫(kù)存數(shù)量為50，我們?cè)儆胘meter測(cè)試一下，把jmeter的測(cè)試地址改為127.0.0.1/stock_redis_lock，同樣的設(shè)置再來(lái)測(cè)一次。

 如何用Redis實(shí)現(xiàn)分布式鎖

測(cè)試了5次沒(méi)有出現(xiàn)臟數(shù)據(jù)，把發(fā)送時(shí)間改為0，測(cè)了5次也沒(méi)問(wèn)題，然后又把線程數(shù)改為600，時(shí)間為0 ，循環(huán)4次，測(cè)了幾次也是正常的。

上面實(shí)現(xiàn)分布式鎖的代碼已經(jīng)是一個(gè)較為成熟的分布式鎖的實(shí)現(xiàn)了，對(duì)大多數(shù)軟件公司來(lái)說(shuō)都已經(jīng)滿足需求了。但是上面代碼還是有優(yōu)化的空間，例如:

1）上面的代碼我們是沒(méi)有考慮異常情況的，實(shí)際情況下代碼沒(méi)有這么簡(jiǎn)單，可能還會(huì)有別的很多復(fù)雜的操作，都有可能會(huì)出現(xiàn)異常，所以我們釋放鎖的代碼需要放在finally塊里來(lái)保證即使是代碼拋異常了釋放鎖的代碼他依然會(huì)被執(zhí)行。

2）還有，你有沒(méi)有注意到，上面我們的分布式鎖的代碼的獲取和設(shè)置過(guò)期時(shí)間的代碼是兩步操作第4行和第5行，即非原子操作，就有可能剛執(zhí)行了第4行還沒(méi)來(lái)得及執(zhí)行第5行這臺(tái)機(jī)器掛了，那么這個(gè)鎖就沒(méi)有設(shè)置超時(shí)時(shí)間，其他線程就一直無(wú)法獲取，除非人工干預(yù)，所以這是一步優(yōu)化的地方，Redis也提供了原子操作，那就是SET key value EX seconds  NX

科普時(shí)間

SET key value [EX seconds] [PX milliseconds] [NX|XX]  將字符串值 value 關(guān)聯(lián)到 key

可選參數(shù)

從 Redis 2.6.12 版本開(kāi)始， SET 命令的行為可以通過(guò)一系列參數(shù)來(lái)修改：

  • EX second ：設(shè)置鍵的過(guò)期時(shí)間為 second 秒。SET key value EX second 效果等同于 SETEX key second value

  • PX millisecond ：設(shè)置鍵的過(guò)期時(shí)間為 millisecond 毫秒。SET key value PX millisecond 效果等同于 PSETEX key millisecond value

  • NX ：只在鍵不存在時(shí)，才對(duì)鍵進(jìn)行設(shè)置操作。SET key value NX 效果等同于 SETNX key value

  • XX ：只在鍵已經(jīng)存在時(shí)，才對(duì)鍵進(jìn)行設(shè)置操作

SpringBoot的StringRedisTemplate也有對(duì)應(yīng)的方法實(shí)現(xiàn)，如下代碼： 

//假設(shè)庫(kù)存編號(hào)是00001private String key = "stock:00001";private String lock_key = "lock_key:00001";
@Autowiredprivate StringRedisTemplate stringRedisTemplate;
@RequestMapping("/stock_redis_lock")public String stock_redis_lock() {
    String uuid = UUID.randomUUID().toString();    try {        //原子的設(shè)置key及超時(shí)時(shí)間
        Boolean aTrue = stringRedisTemplate.opsForValue().setIfAbsent(lock_key, "true", 30, TimeUnit.SECONDS);        if (!aTrue) {            return "error";
        }        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));        if (stock > 0) {            int afterStock = stock - 1;
            stringRedisTemplate.opsForValue().set(key, afterStock + "");
            System.out.println("扣減庫(kù)存成功，剩余庫(kù)存" + afterStock);
        } else {
            System.out.println("扣減庫(kù)存失敗");
        }
    } catch (NumberFormatException e) {
        e.printStackTrace();
    } finally {        //避免死鎖
        if (uuid.equals(stringRedisTemplate.opsForValue().get(lock_key))) {
            stringRedisTemplate.delete(lock_key);
        }
    }    return "ok";
}

這樣實(shí)現(xiàn)是否就完美了呢？嗯，對(duì)于并發(fā)量要求不高或者非大并發(fā)的場(chǎng)景的話這樣實(shí)現(xiàn)已經(jīng)可以了。但是對(duì)于搶購(gòu) ，秒殺這樣的場(chǎng)景，當(dāng)流量很大，這時(shí)候服務(wù)器網(wǎng)卡、磁盤(pán)IO、CPU負(fù)載都可能會(huì)達(dá)到極限，那么服務(wù)器對(duì)于一個(gè)請(qǐng)求的的響應(yīng)時(shí)間勢(shì)必變得比正常情況下慢很多，那么假設(shè)就剛才設(shè)置的鎖的超時(shí)時(shí)間為10秒，如果某一個(gè)線程拿到鎖之后因?yàn)槟承┰驔](méi)能在10秒內(nèi)執(zhí)行完畢鎖就失效了，這時(shí)候其他線程就會(huì)搶占到分布式鎖去執(zhí)行業(yè)務(wù)邏輯，然后之前的線程執(zhí)行完了，會(huì)去執(zhí)行 finally 里的釋放鎖的代碼就會(huì)把正在占有分布式鎖的線程的鎖給釋放掉，實(shí)際上剛剛正在占有鎖的線程還沒(méi)執(zhí)行完，那么其他線程就又有機(jī)會(huì)獲得鎖了...這樣整個(gè)分布式鎖就失效了，將會(huì)產(chǎn)生意想不到的后果。如下圖模擬了這個(gè)場(chǎng)景。

如何用Redis實(shí)現(xiàn)分布式鎖

所以這個(gè)問(wèn)題總結(jié)一下，就是因?yàn)殒i的過(guò)期時(shí)間設(shè)置的不合適或因?yàn)槟承┰驅(qū)е麓a執(zhí)行時(shí)間大于鎖過(guò)期時(shí)間而導(dǎo)致并發(fā)問(wèn)題以及鎖被別的線程釋放，以至于分布式鎖混亂。在簡(jiǎn)單的說(shuō)就是兩個(gè)問(wèn)題，1）自己的鎖被別人釋放 2）鎖超時(shí)無(wú)法續(xù)時(shí)間。

第一個(gè)問(wèn)題很好解決，在設(shè)置分布式鎖時(shí)，我們?cè)诋?dāng)前線程中生產(chǎn)一個(gè)唯一串將value設(shè)置為這個(gè)唯一值，然后在finally塊里判斷當(dāng)前鎖的value和自己設(shè)置的一樣時(shí)再去執(zhí)行delete，如下：

String uuid = UUID.randomUUID().toString();try {    //原子的設(shè)置key及超時(shí)時(shí)間，鎖唯一值
    Boolean aTrue = stringRedisTemplate.opsForValue().setIfAbsent(lock_key,uuid,30,TimeUnit.SECONDS);    //...} finally {    //是自己設(shè)置的鎖再執(zhí)行delete
    if(uuid.equals(stringRedisTemplate.opsForValue().get(lock_key))){
        stringRedisTemplate.delete(lock_key);//避免死鎖    }
}

問(wèn)題一解決了（設(shè)想一下上述代碼還有什么問(wèn)題，一會(huì)兒講），那鎖的超時(shí)時(shí)間就很關(guān)鍵了，不能太大也不能太小，這就需要評(píng)估業(yè)務(wù)代碼的執(zhí)行時(shí)間，比如設(shè)置個(gè)10秒，20秒。即使是你的鎖設(shè)置了合適的超時(shí)時(shí)間，也避免不了可能會(huì)發(fā)生上述分析的因?yàn)槟承┰虼a沒(méi)在正常評(píng)估的時(shí)間內(nèi)執(zhí)行完畢，所以這時(shí)候的解決方案就是給鎖續(xù)超時(shí)時(shí)間。大致思路就是，業(yè)務(wù)線程單獨(dú)起一個(gè)分線程，定時(shí)去監(jiān)聽(tīng)業(yè)務(wù)線程設(shè)置的分布式鎖是否還存在，存在就說(shuō)明業(yè)務(wù)線程還沒(méi)執(zhí)行完，那么就延長(zhǎng)鎖的超時(shí)時(shí)間，若鎖已不存在則業(yè)務(wù)線程執(zhí)行完畢，然后就結(jié)束自己。

“鎖續(xù)命”的這套邏輯屬實(shí)有點(diǎn)復(fù)雜啊，要考慮的問(wèn)題太多了，稍不注意就會(huì)有bug。不要看上面實(shí)現(xiàn)分布式鎖的代碼沒(méi)有幾行，就認(rèn)為實(shí)現(xiàn)起來(lái)很簡(jiǎn)單，如果說(shuō)自己去實(shí)現(xiàn)的時(shí)候沒(méi)有實(shí)際高并發(fā)的經(jīng)驗(yàn)，肯定也會(huì)踩很多坑，例如，

1）鎖的設(shè)置和過(guò)期時(shí)間的設(shè)置是非原子操作的，就可能會(huì)導(dǎo)致死鎖。

2）還有上面遺留的一個(gè)，在finally塊里判斷鎖是否是自己設(shè)置的，是的話再刪除鎖，這兩步操作也不是原子的，假設(shè)剛判斷完為true服務(wù)就掛了，那么刪除鎖的代碼不會(huì)執(zhí)行，就會(huì)造成死鎖，即使是設(shè)置了過(guò)期時(shí)間，在沒(méi)過(guò)期這段時(shí)間也會(huì)死鎖。所以這里也是一個(gè)注意的點(diǎn)，要保證原子操作的話，Redis提供了執(zhí)行Lua腳本的功能來(lái)保證操作的原子性，具體怎么使用不再展開(kāi)。

所以，“鎖續(xù)命”的這套邏輯實(shí)現(xiàn)起來(lái)還是有點(diǎn)復(fù)雜的，好在市面上已經(jīng)有現(xiàn)成的開(kāi)源框架幫我們實(shí)現(xiàn)了，那就是Redisson。

回到頂部

Redisson分布式鎖的實(shí)現(xiàn)原理

如何用Redis實(shí)現(xiàn)分布式鎖

實(shí)現(xiàn)原理：

1、首先Redisson會(huì)嘗試進(jìn)行加鎖，加鎖的原理也是使用類似Redis的setnx命令原子的加鎖，加鎖成功的話其內(nèi)部會(huì)開(kāi)啟一個(gè)子線程

2、子線程主要負(fù)責(zé)監(jiān)聽(tīng)，其實(shí)就是一個(gè)定時(shí)器，定時(shí)監(jiān)聽(tīng)主線程是否還持有鎖，持有則將鎖的時(shí)間延時(shí)，否則結(jié)束線程

3、如果加鎖失敗則自旋不斷嘗試加鎖

4、執(zhí)行完代碼主線程主動(dòng)釋放鎖

那我們看一下使用后Redisson后的代碼是什么樣的。

1、首先在pom.xml文件添加Redisson的maven坐標(biāo)

如何用Redis實(shí)現(xiàn)分布式鎖


    org.redisson
    redisson
    3.12.5

如何用Redis實(shí)現(xiàn)分布式鎖

2、我們要拿到Redisson的這個(gè)對(duì)象，如下配置Bean

如何用Redis實(shí)現(xiàn)分布式鎖

@SpringBootApplicationpublic class RedisLockApplication {    public static void main(String[] args) {
        SpringApplication.run(RedisLockApplication.class, args);
    }
    @Bean    public Redisson redisson(){
        Config config = new Config();
        config.useSingleServer().setAddress("redis://localhost:6379")
                .setDatabase(0);        return (Redisson) Redisson.create(config);
    }
}

如何用Redis實(shí)現(xiàn)分布式鎖

3、然后我們獲取Redisson的實(shí)例，使用其API進(jìn)行加鎖釋放鎖操作

如何用Redis實(shí)現(xiàn)分布式鎖

//假設(shè)庫(kù)存編號(hào)是00001private String key = "stock:00001";private String lock_key = "lock_key:00001";
@Autowiredprivate StringRedisTemplate stringRedisTemplate;/**
 * 使用Redisson實(shí)現(xiàn)分布式鎖
 * @return
 */@RequestMapping("/stock_redisson_lock")public String stock_redisson_lock() {
    RLock redissonLock = redisson.getLock(lock_key);    try {
        redissonLock.lock();        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get(key));        if (stock > 0) {            int afterStock = stock - 1;
            stringRedisTemplate.opsForValue().set(key, afterStock + "");
            System.out.println("扣減庫(kù)存成功，剩余庫(kù)存" + afterStock);
        } else {
            System.out.println("扣減庫(kù)存失敗");
        }
    } catch (NumberFormatException e) {
        e.printStackTrace();
    } finally {
        redissonLock.unlock();
    }    return "ok";
}

如何用Redis實(shí)現(xiàn)分布式鎖

看這個(gè)Redisson的分布式鎖提供的API是不是非常的簡(jiǎn)單？就像Java并發(fā)變成里AQS那套Lock機(jī)制一樣，如下獲取一把RedissonLock

RLock redissonLock = redisson.getLock(lock_key);

默認(rèn)返回的是RedissonLock的對(duì)象，該對(duì)象實(shí)現(xiàn)了RLock接口，而RLock接口繼承了JDK并發(fā)編程報(bào)包里的Lock接口

如何用Redis實(shí)現(xiàn)分布式鎖

在使用Redisson加鎖時(shí)，它也提供了很多API，如下

如何用Redis實(shí)現(xiàn)分布式鎖

現(xiàn)在我們選擇使用的是最簡(jiǎn)單的無(wú)參lock方法，簡(jiǎn)單的點(diǎn)進(jìn)去跟一下看看他的源碼，我們找到最終的執(zhí)行加鎖的代碼如下：

如何用Redis實(shí)現(xiàn)分布式鎖

我們可以看到其底層使用了Lua腳本來(lái)保證原子性，使用Redis的hash結(jié)構(gòu)實(shí)現(xiàn)的加鎖，以及可重入鎖。

比我們自己實(shí)現(xiàn)分布式鎖看起來(lái)還要簡(jiǎn)單，但是我們自己寫(xiě)的鎖功能他都有，我們沒(méi)有的他也有。比如，他實(shí)現(xiàn)的分布式鎖是支持可重入的，也支持可等待，即嘗試等待一定時(shí)間，沒(méi)拿到鎖就返回false。上述代碼中的redissonLock.lock();是一直等待，內(nèi)部自旋嘗試加鎖。

Distributed Java locks and synchronizers 

Lock 

FairLock 

MultiLock 

RedLock 

ReadWriteLock 

Semaphore 

PermitExpirableSemaphore 

CountDownLatch

redisson.org

Redisson提供了豐富的API，內(nèi)部運(yùn)用了大量的Lua腳本保證原子操作，篇幅原因redisson實(shí)現(xiàn)鎖的代碼暫不分析了。

注意：在上述示例代碼中，為了方便演示，查詢r(jià)edis庫(kù)存、修改庫(kù)存并非原子操作，實(shí)際這兩部操作也得保證原子行，可以用redis自帶的Lua腳本功能去實(shí)現(xiàn)

關(guān)于“如何用Redis實(shí)現(xiàn)分布式鎖”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，使各位可以學(xué)到更多知識(shí)，如果覺(jué)得文章不錯(cuò)，請(qǐng)把它分享出去讓更多的人看到。


網(wǎng)站題目：如何用Redis實(shí)現(xiàn)分布式鎖
地址分享：http://weahome.cn/article/pjgpes.html

其他資訊

在線咨詢

微信咨詢

電話咨詢

028-86922220（工作日）

18980820575（7×24）

提交需求

返回頂部