千帆競發-Redis分布式鎖

好久沒有寫博客了，這是在公司進行技術分享時寫的一篇科普性質的文章，在公司CRUD久了人容易失去對技術的向往，從而失去在如今大環境惡劣中的競爭力，大家一定要保持對技術的熱愛，對生活和工作亦是如此！

1.前言

小立，剛入職某電商商城，第一周分到的需求就是公司商品秒殺，在了解完需求后就開始著手設計方案，整體的方案是Redis 緩存+異步同步數據到數據庫。

實現思路

1秒殺前 將商品庫存信息從數據庫同步到redis

2.依靠redis來保證原子性

3.根據對應的返回結果，將訂單數據放入redis訂閱中 或者MQ中進行投遞 ，防止服務器等問題還可以 開一個定時器去掃描這個庫存信息和訂單信息 進行一個補償

4.消費者收到數據后，持久到數據庫中

2.開始編碼

2.1寫第一版代碼：

/***weng@*/ @RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@GetMapping("/buy_goods")public String buy_Goods(){String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;

寫完發現沒有加鎖，在生產環境下發生了超賣問題，a b用戶同時搶到了同一個商品，直接挨屌。

2.2寫第二版代碼：

想到學過JUC 加上鎖就能防止2個用戶同時獲取同一個商品問題，這個時候就考慮是使用 synchronized 還是 ReetranLock了 2者都能實現鎖功能但要選擇哪種這個就犯難了 ，精細控制下還是 選擇r更好 synchronized 必須要釋放鎖 或者出現異常被動釋放鎖 在高并發情況下應該等得到就等 等不到就不等（1不見不散 ， 2過時不候 ）

synchronized執行完同步方法或者代碼塊，才會釋放鎖 并發性下降。

reetranLock if (lock.tryLock()) 或者 if (lock.tryLock(2L,TimeUnit.SECONDS)) 拿得到就執行 ，拿不到就走 提高并發。

對比之下使用ReetranLock會更好。

/***weng@*/ @RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;private final Lock lock = new ReentrantLock();@GetMapping("/buy_goods")public String buyGoods() throws InterruptedException{/*synchronized (this){String number = stringRedisTemplate.opsForValue().get("goods:001");int realNumber = number == null ? 0 : Integer.parseInt(number);if(realNumber > 0){realNumber = realNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",String.valueOf(realNumber));return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件";}}*///if (lock.tryLock(2L,TimeUnit.SECONDS))if (lock.tryLock()){try{String number = stringRedisTemplate.opsForValue().get("goods:001");int realNumber = number == null ? 0 : Integer.parseInt(number);if(realNumber > 0){realNumber = realNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",String.valueOf(realNumber));return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件";}}finally {lock.unlock();}}return "商品售罄/活動結束，歡迎下次光臨";} }

這樣編寫完成后，上線之后沒有發生類似 a b 同時搶同一件商品的情況了，不久后公司技術架構從傳統單體項目升級到微服務架構，從單個JVM虛擬機變成了分布式 不同虛擬機內，這也導致單機的線程鎖機制不在起作用 ，資源在不同的服務器之間共享，那么這個時候就要引出“分布式鎖”。

3.分布式鎖

分布式鎖本質上要實現的目標就是在占一個“茅坑”，當別的進程也要來占 時，發現已經有人蹲在那里了，就只好放棄或者稍后再試。 占坑一般是只允許被一個客戶端占坑。先來先占， 用完了，再調用 del 指令釋放茅坑。

常見實現分布式鎖的方式有1通過redis , 2.通過MySQL（基本沒有使用） 3.通過zk ， 3.mysql 通過 悲觀鎖和樂觀鎖，悲觀鎖 select where for update 鎖表 樂觀鎖 cas 思想 update version zk 通過不斷創建臨時節點實現 性能 沒有redis 性能強 目前最主流的redis 來實現 分布式鎖

分布式鎖需要具備的條件和剛需:

獨占性

OnlyOne，任何時刻只能有且僅有一個線程持有

2.高可用

若redis集群環境下，不能因為某一個節點掛了而出現獲取鎖和釋放鎖失敗的情況

3.防死鎖

杜絕死鎖，必須有超時控制機制或者撤銷操作，有個兜底終止跳出方案

4.不亂搶

防止張冠李戴，不能私下unlock別人的鎖，只能自己加鎖自己釋放

5.重入性

同一個節點的同一個線程如果獲得鎖之后，它也可以再次獲取這個鎖

分布式鎖：

set key value [ex seconds] [px milliseconds] [nx|xx]

ex key在多少秒之后過期

px key在多少毫秒之后過期

nx 當key 不存在時才創建key 效果等同于setnx

xx 當key 存在時覆蓋 key

setnx key value + expire 存在不安全 2條命令非原子性操作

小立了解后開始對原來的代碼進行升級支持分布式環境下使用

3.1 第一版代碼

@RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@GetMapping("/buy_goods")public String buy_Goods(){String key = "wengRedisLock";String value = UUID.randomUUID().toString()+Thread.currentThread().getName();Boolean flagLock = stringRedisTemplate.opsForValue().setIfAbsent(key, value);if(!flagLock){return "搶奪鎖失敗";}String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");stringRedisTemplate.delete(key);System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" +realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;} }

突然有一天咋線上加鎖之后的業務代碼發生異常，但分布式鎖沒有解放導致遲遲不能執行其他的業務方法

3.2 第二版代碼

@RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@GetMapping("/buy_goods")public String buy_Goods(){String key = "wengRedisLock";String value = UUID.randomUUID().toString()+Thread.currentThread().getName();try {Boolean flagLock = stringRedisTemplate.opsForValue().setIfAbsent(key, value);if(!flagLock){return "搶鎖失敗";}String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;} finally {stringRedisTemplate.delete(key);}} }

我靠突然有一天公司的服務器上的微服務jar 包 突然掛了代碼層面根本沒有走到finally這塊，沒辦法保證解鎖，這個key沒有被刪除

3.3第三版代碼

要解決這個問題必須要在redis分布式鎖加入對應的過期時間，放在出現類似這種情況 沒辦法保證解鎖，這個key沒有被刪除，需要加入一個過期時間限定key

@RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@GetMapping("/buy_goods")public String buy_Goods(){String key = "wengRedisLock";String value = UUID.randomUUID().toString()+Thread.currentThread().getName();try {Boolean flagLock = stringRedisTemplate.opsForValue().setIfAbsent(key, value);stringRedisTemplate.expire(key,10L,TimeUnit.SECONDS);if(!flagLock){return "搶鎖失敗";}String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;} finally {stringRedisTemplate.delete(key);}} }

突然在公司周年慶時流量QPS 提高很多，在購買商品時發生在一個時刻商品明明有貨但用戶不能購買被別人占有有，結果發現是設置key +過期不是在同一個原子操作中

3.4 第四版代碼

解決上面的這個問題，將設置key和過期時間放到一個命令，保證原子性

原子操作就是: 不可中斷的一個或者一系列操作, 也就是不會被線程調度機制打斷的操作, 運行期間不會有任何的上下文切換(context switch)

@RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@GetMapping("/buy_goods")public String buy_Goods(){String key = "wymRedisLock";String value = UUID.randomUUID().toString()+Thread.currentThread().getName();try {Boolean flagLock = stringRedisTemplate.opsForValue().setIfAbsent(key,value,10L,TimeUnit.SECONDS);if(!flagLock){return "搶鎖失敗,o(╥﹏╥)o";}String result =stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;} finally {stringRedisTemplate.delete(key);}} }

上線一段時間后發現 可能在設置的過期時間業務還未完成，鎖已經被刪了，然后finally塊中就可能會刪除別的服務創建的鎖，張冠李戴

3.5 第五版代碼

解決上面的問題 只能刪除自己創建的鎖，不能動別人的

@RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@GetMapping("/buy_goods")public String buy_Goods(){String key = "wengRedisLock";String value = UUID.randomUUID().toString()+Thread.currentThread().getName();try {Boolean flagLock = stringRedisTemplate.opsForValue().setIfAbsent(key,value,10L,TimeUnit.SECONDS);if(!flagLock){return "搶鎖失敗,o(╥﹏╥)o";}String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;} finally {if (stringRedisTemplate.opsForValue().get(key).equals(value)) {stringRedisTemplate.delete(key);}}} }

在高qps 情況下 finally塊的判斷+del刪除操作不是原子性的，會發現商品明明在卻不能購買的情況

3.6第六版代碼

使用Lua腳本Redis調用Lua腳本通過eval命令保證代碼執行的原子性

使用jdeis@RestController public class GoodController {public static final String REDIS_LOCK_KEY = "redisLockPay";@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@GetMapping("/buy_goods")public String buy_Goods(){String value = UUID.randomUUID().toString()+Thread.currentThread().getName();try {Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(REDIS_LOCK_KEY, value,30L,TimeUnit.SECONDS);if(!flag){return "搶奪鎖失敗，請下次嘗試";}String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;} finally {Jedis jedis = RedisUtils.getJedis();String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +"then " +"return redis.call('del', KEYS[1]) " +== ARGV[1] " +"then " +"return redis.call('del', KEYS[1]) " +"else " +" return 0 " +"end";try {Object result = jedis.eval(script, Collections.singletonList(REDIS_LOCK_KEY), Collections.singletonList(value));if ("1".equals(result.toString())) {System.out.println("------del REDIS_LOCK_KEY success");}else{System.out.println("------del REDIS_LOCK_KEY error");}} finally {if(null != jedis) {jedis.close();}}}} }

終于一系列的迭代之后基于Redis實現分布式鎖達到了分布式要具備的獨占性，防死鎖，不亂搶。

4.集群環境

? 隨著公司業務的發展，技術是服務業務的，redis 架構也演進到了集群環境 ，隨著而來在之前的單節點實現的redis 分布式鎖也暴露出了redisLock過期時間小于業務執行時間的問題 ，redis分布式鎖續費的問題，在集群環境下redis 異步復制造成鎖丟失問題， 比如：主節點沒來的及把剛剛set進來這條數據給從節點，master就掛了，從機上位但從機上無該數據 … redis 環境下 自己寫的也不ok 要考慮的東西太多了，直接上 RedLock 的Redisson落地 。

4.1第一版代碼

@RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@Autowiredprivate Redisson redisson;@GetMapping("/buy_goods")public String buy_Goods(){String key = "wengRedisLock";RLock redissonLock = redisson.getLock(key);redissonLock.lock();try{String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0) {int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;}finally {redissonLock.unlock();}} }

按照之前單機版的 可能解鎖了別服務創建的鎖，所以解鎖時需要判斷當前鎖是否是自己創建的那個，避免張冠李戴

4.2 第二版代碼

@RestController public class GoodController {@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String serverPort;@Autowiredprivate Redisson redisson;@GetMapping("/buy_goods")public String buy_Goods(){String key = "wengRedisLock";RLock redissonLock = redisson.getLock(key);redissonLock.lock();try{String result = stringRedisTemplate.opsForValue().get("goods:001");int goodsNumber = result == null ? 0 : Integer.parseInt(result);if(goodsNumber > 0){int realNumber = goodsNumber - 1;stringRedisTemplate.opsForValue().set("goods:001",realNumber + "");System.out.println("你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort);return "你已經成功秒殺商品，此時還剩余：" + realNumber + "件"+"\t 服務器端口："+serverPort;}else{System.out.println("商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort);}return "商品已經售罄/活動結束/調用超時，歡迎下次光臨"+"\t 服務器端口："+serverPort;}finally {if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()){redissonLock.unlock();}}} }

1setnx分布式鎖缺點
嚴禁出現2個以上的請求線程拿到鎖。危險的

? 使用到的 就是Redlock (紅鎖)算法 大概就是客戶端發送setnx指令，同時向多個redis節點發信息，超過半數redis節點加鎖成功，才會返回成功 具體的落地實現是redisson客戶端工具。鎖變量由多個實例維護，即使有實例發生了故障，鎖變量仍然是存在的，客戶端還是可以完成鎖操作。Redlock算法是實現高可靠分布式鎖的一種有效解決方案，可以在實際開發中使用。

在使用 這套理論時 該方案為了解決數據不一致的問題，直接舍棄了異步復制只使用 master 節點，同時由于舍棄了 slave，為了保證可用性，引入了 N 個節點

N = 2X + 1 (N是最終部署機器數，X是容錯機器數)

失敗了多少個機器實例后我還是可以容忍的，所謂的容忍就是數據一致性還是可以Ok的，CP數據一致性還是可以滿足
加入在集群環境中，redis失敗1臺，可接受。2X+1 = 2 * 1+1 =3，部署3臺，死了1個剩下2個可以正常工作，那就部署3臺。
加入在集群環境中，redis失敗2臺，可接受。2X+1 = 2 * 2+1 =5，部署5臺，死了2個剩下3個可以正常工作，那就部署5臺。

4.3 第三版代碼

@RestController @Slf4j public class RedLockController {public static final String CACHE_KEY_REDLOCK = "ZZYY_REDLOCK";@AutowiredRedissonClient redissonClient1;@AutowiredRedissonClient redissonClient2;@AutowiredRedissonClient redissonClient3;@GetMapping(value = "/redlock")public void getlock() {//CACHE_KEY_REDLOCK為redis 分布式鎖的keyRLock lock1 = redissonClient1.getLock(CACHE_KEY_REDLOCK);RLock lock2 = redissonClient2.getLock(CACHE_KEY_REDLOCK);RLock lock3 = redissonClient3.getLock(CACHE_KEY_REDLOCK);RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);boolean isLock;try {//waitTime 鎖的等待時間處理,正常情況下 等5s//leaseTime就是redis key的過期時間,正常情況下等5分鐘。isLock = redLock.tryLock(5, 300, TimeUnit.SECONDS);log.info("線程{}，是否拿到鎖：{} ",Thread.currentThread().getName(),isLock);if (isLock) {//TODO if get lock success, do something;//暫停20秒鐘線程try { TimeUnit.SECONDS.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); }}} catch (Exception e) {log.error("redlock exception ",e);} finally {// 無論如何, 最后都要解鎖redLock.unlock();System.out.println(Thread.currentThread().getName()+"\t"+"redLock.unlock()");}}}

這樣就通過調用redisson-api實現高可用 集群分布式鎖

5.Redisson小結

看門狗 守護線程“緩存續命” 額外起一個線程，定期檢查線程是否還持有鎖，如果有則延長過期時間，Redisson 里面就實現了這個方案使用“看門狗”定期檢查（每1/3的鎖時間檢查1次），如果線程還持有鎖，則刷新過期時間；在獲取鎖成功后，給鎖加一個 watchdog，watchdog 會起一個定時任務，在鎖沒有被釋放且快要過期的時候會續期，具體的源碼可以 百度下周陽老師的redis課程。

總結

以上是生活随笔為你收集整理的千帆竞发-Redis分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。