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

一、通過(guò)setnx實(shí)現(xiàn)

1、setnx key value

當(dāng)且僅當(dāng)key不存在，將key的值設(shè)置為value，并且返回1；若是給定的key已經(jīng)存在，則setnx不做任何動(dòng)作，返回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對(duì)應(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);
            }
        });
    }

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

小明提出了方案一：

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

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){
  //上一筆同樣的請(qǐng)求還未處理完成，輪訓(xùn)等待(具體如何輪訓(xùn)在此不展開)
}else{
  //上一筆同樣的請(qǐng)求處理完成，進(jìn)行查庫(kù)操作
  resultDao.select("參數(shù)");
}

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

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

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

這種情況是，大致判斷了外部廠商C系統(tǒng)業(yè)務(wù)處理時(shí)間大概為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)超時(shí)
RedisUtils.delete("LOCK_KEY_phone&idNo&name");
}

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

（B）方案的升級(jí)版—->>（C）方案：

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

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

總結(jié)： 鎖超時(shí)了該如何處理，通過(guò)getset方式判斷時(shí)間戳差的方式，多比同時(shí)getset都得到超時(shí)，同時(shí)去setnx?？倳?huì)有一個(gè)更快地去setnx。

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

1、incr

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

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) {
            //如果計(jì)數(shù)器>1,說(shuō)明已經(jīng)有請(qǐng)求進(jìn)來(lái)
            throw new AppException(ResponseCode.FAIL.getCode(), "操作頻繁");
        }


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

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

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

最新評(píng)論