歡迎您光臨本站 註冊首頁

基於redis實現分散式鎖的原理與方法

←手機掃碼閱讀     retouched @ 2020-06-03 , reply:0

前言

系統的不斷擴大,分散式鎖是最基本的保障。與單機的多執行緒不一樣的是,分散式跨多個機器。執行緒的共享變數無法跨機器。

為了保證一個在高併發存場景下只能被同一個執行緒操作,java併發處理提供ReentrantLock或Synchronized進行互斥控制。但是這僅僅對單機環境有效。我們實現分散式鎖大概透過三種方式。

  • redis實現分散式鎖

  • 資料庫實現分散式鎖

  • zk實現分散式鎖

今天我們介紹透過redis實現分散式鎖。實際上這三種和java對比看屬於一類。都是屬於程式外部鎖。

原理剖析

  • 上述三種分散式鎖都是透過各自為依據對各個請求進行上鎖,解鎖從而控制放行還是拒絕。redis鎖是基於其提供的setnx命令。

  • setnx當且僅當key不存在。若給定key已經存在,則setnx不做任何動作。setnx是一個原子性操作。

  • 和資料庫分散式相比,因為redis記憶體輕量。所以redis分散式鎖效能更好

實現

原理很簡單。結合springboot專案我們實現一套透過註解形式對介面進行庫存上鎖案例進行理解

編寫註解

我們編寫註解。方便我們在介面上新增註解提供攔截資訊

 /** * @author 張新華 * @version V1.0 * @Package com.ay.framework.order.redis.product * @date 2020年03月26日, 0026 10:29 * @Copyright © 2020 安元科技有限公司 */ @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited public @interface StockLock { /** * @author zxhtom * @Description 鎖key的字首 * @Date 15:25 2020年03月25日, 0025 * @Param [] * @return java.lang.String */ String prefix() default ""; /** * @author zxhtom * @Description key的分隔符 * @Date 15:27 2020年03月25日, 0025 * @Param [] * @return java.lang.String */ String delimiter() default ":"; }


 /** * @author 張新華 * @version V1.0 * @Package com.ay.framework.order.redis.product * @date 2020年03月26日, 0026 11:09 * @Copyright © 2020 安元科技有限公司 */ @Target({ElementType.PARAMETER , ElementType.METHOD , ElementType.FIELD}) @Retention(RetentionPolicy.RUNTIME) @Documented @Inherited public @interface StockParam { /** * @author zxhtom * @Description 組成key * @Date 11:11 2020年03月26日, 0026 * @Param [] * @return java.lang.String[] */ String[] names() default {""}; }


攔截器攔截

redis分散式鎖實現的關鍵就是攔截器的編寫。上面的註解只是為了實現攔截的一個輔助。

 @Around("execution(public * *(..)) && @annotation(com.ay.framework.order.redis.product.StockLock)")


透過springboot的Around進行鍼對StockLock註解的攔截。透過攔截我們可以獲取到攔截的方法、引數、及需要的鎖的引數。

我們獲取到需要鎖的名稱這裡叫做【a】之後透過redis的原子性操作對該key進行遞減操作。

為了方便我們在削減庫存的時候可以對庫存進行更新操作。我們在遞減庫存前還需要藉助於另一把鎖。 這一把鎖我們叫做【a_key】

換句話說我們介面想訪問就必須獲取【a】鎖,拿到【a】鎖需要減少庫存。減少庫存之前需要獲取【a_key】鎖。

拿到鎖之後處理完邏輯之後我們需要釋放對應鎖。

 RedisAtomicLong entityIdCounter = new RedisAtomicLong(lockKey, redisTemplate.getConnectionFactory()); if (redisTemplate.hasKey(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey)) { //表示lockKey的庫存資訊有變動。此時無法進行交易 throw new BusinessException("庫存變動。暫無法交易"); } Long increment = entityIdCounter.decrementAndGet(); if (increment >= 0) { try { Object proceed = pjp.proceed(); } catch (Throwable throwable) { //所佔資源需要釋放回資源池 while (!redisLock.tryGetLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey, "")) { } //表示lockKey的庫存資訊有變動。此時無法進行交易 long l = entityIdCounter.incrementAndGet(); if (l<1) { redisTemplate.opsForValue().set(lockKey,1); } redisLock.unLock(CoreConstants.UPDATEPRODUCTREDISLOCKKEY + lockKey); throwable.printStackTrace(); } } else { redisTemplate.opsForValue().set(lockKey,0); throw new BusinessException("庫存不足!無法操作"); }

因為我們上鎖就需要釋放鎖。但是程式在中途處理業務是發生異常導致沒有走到釋放鎖的步驟。這個時候就導致我們的分散式鎖一直被鎖。俗稱【死鎖】。為了避免這種場景的發生。我們常常在上鎖的時候給一個有效期。有效期已過自動釋放鎖。這個特性恰好和redis的過期策略不摩爾和。

上述提及工具

RedisLock

 public Boolean tryGetLock(String key , String value) { return tryGetLock(key, value, -1, TimeUnit.DAYS); } public Boolean tryGetLock(String key , String value, Integer expire) { return tryGetLock(key, value, expire, TimeUnit.SECONDS); } public Boolean tryGetLock(String key , String value, Integer expire , TimeUnit timeUnit) { ValueOperations operatiOns= redisTemplate.opsForValue(); if (operations.setIfAbsent(key, value)) { //說明 redis沒有該key , 換言之 加鎖成功 設定過期時間防止死鎖 if (expire > 0) { redisTemplate.expire(key, expire, timeUnit); } return true; } return false; } public Boolean unLock(String key) { return redisTemplate.delete(key); }


StockKeyGenerator

 @Component() @Primary public class StockKeyGenerator implements CacheKeyGenerator { @Override public String getLockKey(ProceedingJoinPoint pjp) { //獲取方法簽名 MethodSignature signature = (MethodSignature) pjp.getSignature(); Method method = signature.getMethod(); //獲取方法cacheLock註解 StockLock stockLock = method.getAnnotation(StockLock.class); //獲取方法引數 Object[] args = pjp.getArgs(); Parameter[] parameters = method.getParameters(); StringBuilder builder = new StringBuilder(); for (int i = 0; i 0) { builder.append(stockLock.delimiter()); } builder.append(temArgMap.get(name)); } } } return builder.toString(); } }


問題分析

上面分析了一個死鎖的場景,理論上出了死鎖我們redis分佈鎖很好的解決了分散式問題。但是還是會出現問題。下面列舉寫小編遇到的問題。

業務處理時間>上鎖過期時間

a執行緒獲取到鎖,開始進行業務處理需要8S,

在8S內,鎖的有效期是5S,在鎖過期後也就是第6S , b執行緒進入開始獲取鎖這個時候b是可以獲取到新鎖的。這個時候就是有問題的。

假設b執行緒業務處理只需要3S , 但是因為a執行緒釋放了鎖,所以在第8S的時候雖然b執行緒沒有釋放鎖,b的鎖也沒有過期但是這時候也沒有了鎖。從而導致C執行緒也可以進入


[retouched ] 基於redis實現分散式鎖的原理與方法已經有324次圍觀

http://coctec.com/docs/developer/show-post-236823.html