快捷導(dǎo)航

Redis實現(xiàn)分布式鎖(setnx、getset、incr)以及如何處理超時情況

更新時間：2021年11月11日 10:46:09 作者：九十三大人

本文主要介紹了Redis實現(xiàn)分布式鎖(setnx、getset、incr)以及如何處理超時情況，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下

如果你通過網(wǎng)絡(luò)搜索分布式鎖，最多的就是基于redis的了?；趓edis的分布式鎖得益于redis的單線程執(zhí)行機制，單線程在執(zhí)行上就保證了指令的順序化，所以很大程度上降低了開發(fā)人員的思考設(shè)計成本。

一、通過setnx實現(xiàn)

1、setnx key value

當(dāng)且僅當(dāng)key不存在，將key的值設(shè)置為value，并且返回1；若是給定的key已經(jīng)存在，則setnx不做任何動作，返回0。

public static Boolean setnx(final String key, final String value, final long seconds) {
        return getShardedJedisClient().execute(new ShardedJedisAction<Boolean>() {
            public Boolean doAction(ShardedJedis shardedJedis) {
                Jedis jedis = (Jedis) shardedJedis.getShard(key);
                String result = jedis.set(key, value, "NX", "EX", seconds);
                return "OK".equalsIgnoreCase(result);
            }
        });
    }

2、get key

獲取key對應(yīng)的value值，如果不存在該key，返回0。

public String get(final String key) {
        this.checkIsInMulti();
        return (String)this.execute(new SmartJedis.Action<String>() {
            public String doAction(Jedis jedis) {
                return jedis.get(key);
            }
        }, SmartJedis.RW.R, key);
    }

3、getset key value

獲取key的舊值，將新value放入

public static String getset(final String key, final String value) {
        return getShardedJedisClient().execute(new ShardedJedisAction<String>() {
            @Override
            public String doAction(ShardedJedis shardedJedis) {
                return shardedJedis.getSet(key, value);
            }
        });
    }

至此，我們先舉個手機三要素驗證的列子：(A渠道系統(tǒng)，業(yè)務(wù)B系統(tǒng)，外部廠商C系統(tǒng))
(1)B業(yè)務(wù)系統(tǒng)調(diào)用A渠道系統(tǒng)，驗證傳入的手機、身份證、號碼三要素是否一一致。
(2)A渠道系統(tǒng)再調(diào)用外部廠商C系統(tǒng)。
(3)A渠道系統(tǒng)將結(jié)果返回給B業(yè)務(wù)系統(tǒng)。
這3個過程中，(2)過程，外部廠商的調(diào)用時是需要計費的。
當(dāng)B業(yè)務(wù)系統(tǒng)并發(fā)量很高時，有100筆相同的三要素校驗，由于是相同的三要素，A渠道只要調(diào)用一次廠商即可知道結(jié)果。那么A渠道系統(tǒng)如何控制不讓100筆請求全部去訪問外部廠商C系統(tǒng)呢？

小明提出了方案一：

在A系統(tǒng)中，
當(dāng)100個線程同時請求過來，進行redis.setnx(“LOCK_KEY_phone&idNo&name”,”demo”)，這樣第一筆線程率先拿到鎖，其他的線程等待，當(dāng)thread(0)處理結(jié)束后，thread(0)進行delete(“LOCK_KEY_phone&idNo&name”),把鎖放開，thread(i)進行g(shù)et(“LOCK_KEY_phone&idNo&name”)拿到0，說明上一筆已經(jīng)處理完成，這個時候，我們可以去查詢上一筆的記錄。

RedisUtils.setnx("LOCK_KEY_phone&idNo&name","demo");
JSONObject result = A.request(B);
AssetUtils.notNull(result,ResponseCodeEnum.Success,"拿到結(jié)果");
ResultDmo resultDmo = (ResultDmo)BeanUtils.maptoBean(result);
resultDao.insert(resultDmo);
if(result!=0){
  //上一筆同樣的請求還未處理完成，輪訓(xùn)等待(具體如何輪訓(xùn)在此不展開)
}else{
  //上一筆同樣的請求處理完成，進行查庫操作
  resultDao.select("參數(shù)");
}

小宏說：小明的思想不嚴(yán)謹(jǐn)

問題：當(dāng)100筆線程中一些線程超時或者系統(tǒng)宕機等意外情況發(fā)現(xiàn)，鎖會一直被某些線程持有，造成死鎖狀態(tài)。
應(yīng)該給緩存key設(shè)置一個超時時間。比如:200ms

RedisUtils.setnx("LOCK_KEY_phone&idNo&name","demo",200);

這種情況是，大致判斷了外部廠商C系統(tǒng)業(yè)務(wù)處理時間大概為200ms，

網(wǎng)上看還有一種方式（B）：

RedisUtils.setnx("LOCK_KEY_phone&idNo&name",currentTime,200);
Long old = RedisUtils.get("LOCK_KEY_phone&idNo&name");
Long new = System.currentTimeMillis();
Long time = new - old;
if(time>0){
//處理已經(jīng)超時
RedisUtils.delete("LOCK_KEY_phone&idNo&name");
}

（B）這種情況不嚴(yán)謹(jǐn)：當(dāng)a獲取setnx鎖，a線程崩潰或超時，b、c線程同時get到old，且判斷超時，可能出現(xiàn)b線程delete a線程的鎖，并且setnx后；c線程又將b線程的鎖delete,并且setnx。這種情況完全鎖不住線程了。

（B）方案的升級版—->>（C）方案：

當(dāng)a獲取setnx鎖，a線程崩潰或超時，b線程getset，獲取old且判斷超時，c線程getset，獲取old（此時這個值是b剛剛set進去的），判斷未超時，c繼續(xù)等待。b線程delete a線程的鎖，并且setnx后。這種情況是安全的。

需要注意的地方：
①不要輕易將get和getset混用，筆者認(rèn)為getset單獨使用比較好。
有一種情況，a、b、c、三個線程，a、b同時get，a立即返回了old，突然來了個c，卡在b之前getset了，且刪除鎖，那么b的get只能返回nil了。此時再根據(jù)時間戳對比：
a.get != (a.set)
b.get ! = (b.set)
這樣a、b都沒拿到鎖，但是a其實已經(jīng)獲取到了鎖。
②多個服務(wù)器時間的同步問題。

總結(jié)：鎖超時了該如何處理，通過getset方式判斷時間戳差的方式，多比同時getset都得到超時，同時去setnx。總會有一個更快地去setnx。

二、通過incr搶占資源實現(xiàn)

1、incr

將 key 中儲存的數(shù)字值增一。如果 key 不存在，那么 key 的值會先被初始化為 0 ，然后再執(zhí)行 INCR 操作。如果值包含錯誤的類型，或字符串類型的值不能表示為數(shù)字，那么返回一個錯誤。

public static Long incr(final String key) {
        return shardedClient.execute(new ShardedJedisAction<Long>() {
            @Override
            public Long doAction(ShardedJedis shardedJedis) {
                shardedJedis.expire(key, 200);
                return shardedJedis.incr(key);
            }
        });
    }

還是上面的三要素的例子

Long result = RedisUtils.incr("LOCK_KEY_phone&idNo&name");
        if (result > 1) {
            //如果計數(shù)器>1,說明已經(jīng)有請求進來
            throw new AppException(ResponseCode.FAIL.getCode(), "操作頻繁");
        }


 JSONObject result = A.request(B);
 Long endTime = System.currentTimeMillis();
 Long time = endTime - startTime;
  //如果處理時間大于incr的key存活時間,說明該筆請求已經(jīng)超時
  if (time > 200) {
      //全局ID,統(tǒng)計超時次數(shù)
      String key = "LOCK_KEY_phone&idNo&name" + source;
      RedisUtils.incr(key);
      int total = Integer.valueOf(RedisUtils.get(key));
      //斷言若超時10次,進行報警(報警不在次展開)
      AssertUtils.isTrue(total < 10, ResponseCode.FAIL, "調(diào)用" + source + "渠道超時");
  }

這里設(shè)置了計數(shù)器的超時時間為200ms，如果請求超時，會有大量的線程同時訪問，筆者這里有10筆同時過來，就啟動報警。人為排查渠道。和setnx的不同是，某個線程超時，setnx的方式需要手動去判斷，再去加鎖，防止大量線程進入(這里可以通過輪訓(xùn)實現(xiàn))；而incr的方式超時了，大量線程進來，我不做處理，但是這里的time>200是具有誤差的。

到此這篇關(guān)于Redis實現(xiàn)分布式鎖(setnx、getset、incr)以及如何處理超時情況的文章就介紹到這了,更多相關(guān)Redis setnx、getset、incr內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: