引言

最近項目上線的頻率頗高，連著幾天加班熬夜，身體有點吃不消精神也有些萎靡，無奈業務方催的緊，工期就在眼前只能硬著頭皮上了。腦子渾渾噩噩的時候，寫的就不能叫代碼，可以直接叫做Bug。我就熬夜寫了一個bug被罵慘了。

由于是做商城業務，要頻繁的對商品庫存進行扣減，應用是集群部署，為避免并發造成庫存超買超賣等問題，采用?redis?分布式鎖加以控制。本以為給扣庫存的代碼加上鎖lock.tryLock就萬事大吉了

1????/**2?????*?@author?xiaofu3?????*?@description?扣減庫存4?????*?@date?2020/4/21?12:105?????*/6???public?String?stockLock()?{7????????RLock?lock?=?redissonClient.getLock("stockLock");8????????try?{9????????????/** 10?????????????*?獲取鎖 11?????????????*/ 12????????????if?(lock.tryLock(10,?TimeUnit.SECONDS))?{ 13????????????????/** 14?????????????????*?查詢庫存數 15?????????????????*/ 16????????????????Integer?stock?=?Integer.valueOf(stringRedisTemplate.opsForValue().get("stockCount")); 17????????????????/** 18?????????????????*?扣減庫存 19?????????????????*/ 20????????????????if?(stock?>?0)?{ 21????????????????????stock?=?stock?-?1; 22????????????????????stringRedisTemplate.opsForValue().set("stockCount",?stock.toString()); 23????????????????????LOGGER.info("庫存扣減成功，剩余庫存數量：{}",?stock); 24????????????????}?else?{ 25????????????????????LOGGER.info("庫存不足~"); 26????????????????} 27????????????}?else?{ 28????????????????LOGGER.info("未獲取到鎖業務結束.."); 29????????????} 30????????}?catch?(Exception?e)?{ 31????????????LOGGER.info("處理異常",?e); 32????????}?finally?{ 33????????????lock.unlock(); 34????????} 35????????return?"ok"; 36??}

結果業務代碼執行完以后我忘了釋放鎖lock.unlock()，導致redis線程池被打滿，redis服務大面積故障，造成庫存數據扣減混亂，被領導一頓臭罵，這個月績效~ 哎·~。

隨著 使用redis?鎖的時間越長，我發現?redis?鎖的坑遠比想象中要多。就算在面試題當中redis分布式鎖的出鏡率也比較高，比如：“用鎖遇到過哪些問題？” ，“又是如何解決的？” 基本都是一套連招問出來的。

今天就分享一下我用redis?分布式鎖的踩坑日記，以及一些解決方案，和大家一起共勉。

一、鎖未被釋放

這種情況是一種低級錯誤，就是我上邊犯的錯，由于當前線程 獲取到redis?鎖，處理完業務后未及時釋放鎖，導致其它線程會一直嘗試獲取鎖阻塞，例如：用Jedis客戶端會報如下的錯誤信息

1redis.clients.jedis.exceptions.JedisConnectionException:?Could?not?get?a?resource?from?the?pool

redis線程池已經沒有空閑線程來處理客戶端命令。

解決的方法也很簡單，只要我們細心一點，拿到鎖的線程處理完業務及時釋放鎖，如果是重入鎖未拿到鎖后，線程可以釋放當前連接并且sleep一段時間。

1??public?void?lock()?{2??????while?(true)?{3??????????boolean?flag?=?this.getLock(key);4??????????if?(flag)?{5????????????????TODO?.........6??????????}?else?{7????????????????//?釋放當前redis連接8????????????????redis.close();9????????????????//?休眠1000毫秒 10????????????????sleep(1000); 11??????????} 12????????} 13????}

二、B的鎖被A給釋放了

我們知道Redis實現鎖的原理在于?SETNX命令。當?key不存在時將?key的值設為?value?，返回值為?1；若給定的?key已經存在，則?SETNX不做任何動作，返回值為?0?。

1SETNX?key?value

我們來設想一下這個場景：A、B兩個線程來嘗試給key?myLock加鎖，A線程先拿到鎖（假如鎖3秒后過期），B線程就在等待嘗試獲取鎖，到這一點毛病沒有。

那如果此時業務邏輯比較耗時，執行時間已經超過redis鎖過期時間，這時A線程的鎖自動釋放（刪除key），B線程檢測到myLock這個key不存在，執行?SETNX命令也拿到了鎖。

但是，此時A線程執行完業務邏輯之后，還是會去釋放鎖（刪除key），這就導致B線程的鎖被A線程給釋放了。

為避免上邊的情況，一般我們在每個線程加鎖時要帶上自己獨有的value值來標識，只釋放指定value的key，否則就會出現釋放鎖混亂的場景。

三、數據庫事務超時

emm~ 聊redis鎖咋還扯到數據庫事務上來了？別著急往下看，看下邊這段代碼：

1???@Transaction2???public?void?lock()?{34????????while?(true)?{5????????????boolean?flag?=?this.getLock(key);6????????????if?(flag)?{7????????????????insert();8????????????}9????????} 10????}

給這個方法添加一個@Transaction注解開啟事務，如代碼中拋出異常進行回滾，要知道數據庫事務可是有超時時間限制的，并不會無條件的一直等一個耗時的數據庫操作。

比如：我們解析一個大文件，再將數據存入到數據庫，如果執行時間太長，就會導致事務超時自動回滾。

一旦你的key長時間獲取不到鎖，獲取鎖等待的時間遠超過數據庫事務超時時間，程序就會報異常。

一般為解決這種問題，我們就需要將數據庫事務改為手動提交、回滾事務。

1????@Autowired2????DataSourceTransactionManager?dataSourceTransactionManager;34????@Transaction5????public?void?lock()?{6????????//手動開啟事務7????????TransactionStatus?transactionStatus?=?dataSourceTransactionManager.getTransaction(transactionDefinition);8????????try?{9????????????while?(true)?{ 10????????????????boolean?flag?=?this.getLock(key); 11????????????????if?(flag)?{ 12????????????????????insert(); 13????????????????????//手動提交事務 14????????????????????dataSourceTransactionManager.commit(transactionStatus); 15????????????????} 16????????????} 17????????}?catch?(Exception?e)?{ 18????????????//手動回滾事務 19????????????dataSourceTransactionManager.rollback(transactionStatus); 20????????} 21????}

四、鎖過期了，業務還沒執行完

這種情況和我們上邊提到的第二種比較類似，但解決思路上略有不同。

同樣是redis分布式鎖過期，而業務邏輯沒執行完的場景，不過，這里換一種思路想問題，把redis鎖的過期時間再弄長點不就解決了嗎？

那還是有問題，我們可以在加鎖的時候，手動調長redis鎖的過期時間，可這個時間多長合適？業務邏輯的執行時間是不可控的，調的過長又會影響操作性能。

要是redis鎖的過期時間能夠自動續期就好了。

為了解決這個問題我們使用redis客戶端redisson，redisson很好的解決了redis在分布式環境下的一些棘手問題，它的宗旨就是讓使用者減少對Redis的關注，將更多精力用在處理業務邏輯上。

redisson對分布式鎖做了很好封裝，只需調用API即可。

1??RLock?lock?=?redissonClient.getLock("stockLock");

redisson在加鎖成功后，會注冊一個定時任務監聽這個鎖，每隔10秒就去查看這個鎖，如果還持有鎖，就對過期時間進行續期。默認過期時間30秒。這個機制也被叫做：“看門狗”，這名字。。。

舉例子：假如加鎖的時間是30秒，過10秒檢查一次，一旦加鎖的業務沒有執行完，就會進行一次續期，把鎖的過期時間再次重置成30秒。

通過分析下邊redisson的源碼實現可以發現，不管是加鎖、解鎖、續約都是客戶端把一些復雜的業務邏輯，通過封裝在Lua腳本中發送給redis，保證這段復雜業務邏輯執行的原子性。

1@Slf4j2@Service3public?class?RedisDistributionLockPlus?{45????/**6?????*?加鎖超時時間，單位毫秒，?即：加鎖時間內執行完操作，如果未完成會有并發現象7?????*/8????private?static?final?long?DEFAULT_LOCK_TIMEOUT?=?30;910????private?static?final?long?TIME_SECONDS_FIVE?=?5?;1112????/**13?????*?每個key的過期時間?{@link?LockContent}14?????*/15????private?Map<String,?LockContent>?lockContentMap?=?new?ConcurrentHashMap<>(512);1617????/**18?????*?redis執行成功的返回19?????*/20????private?static?final?Long?EXEC_SUCCESS?=?1L;2122????/**23?????*?獲取鎖lua腳本， k1：獲鎖key, k2：續約耗時key, arg1:requestId，arg2：超時時間24?????*/25????private?static?final?String?LOCK_SCRIPT?=?"if?redis.call('exists',?KEYS[2])?==?1?then?ARGV[2]?=?math.floor(redis.call('get',?KEYS[2])?+?10)?end?"?+26????????????"if?redis.call('exists',?KEYS[1])?==?0?then?"?+27???????????????"local?t?=?redis.call('set',?KEYS[1],?ARGV[1],?'EX',?ARGV[2])?"?+28???????????????"for?k,?v?in?pairs(t)?do?"?+29?????????????????"if?v?==?'OK'?then?return?tonumber(ARGV[2])?end?"?+30???????????????"end?"?+31????????????"return?0?end";3233????/**34?????*?釋放鎖lua腳本, k1：獲鎖key, k2：續約耗時key, arg1:requestId，arg2：業務耗時 arg3:?業務開始設置的timeout35?????*/36????private?static?final?String?UNLOCK_SCRIPT?=?"if?redis.call('get',?KEYS[1])?==?ARGV[1]?then?"?+37????????????"local?ctime?=?tonumber(ARGV[2])?"?+38????????????"local?biz_timeout?=?tonumber(ARGV[3])?"?+39????????????"if?ctime?>?0?then??"?+40???????????????"if?redis.call('exists',?KEYS[2])?==?1?then?"?+41???????????????????"local?avg_time?=?redis.call('get',?KEYS[2])?"?+42???????????????????"avg_time?=?(tonumber(avg_time)?*?8?+?ctime?*?2)/10?"?+43???????????????????"if?avg_time?>=?biz_timeout?-?5?then?redis.call('set',?KEYS[2],?avg_time,?'EX',?24*60*60)?"?+44???????????????????"else?redis.call('del',?KEYS[2])?end?"?+45???????????????"elseif?ctime?>?biz_timeout?-5?then?redis.call('set',?KEYS[2],?ARGV[2],?'EX',?24*60*60)?end?"?+46????????????"end?"?+47????????????"return?redis.call('del',?KEYS[1])?"?+48????????????"else?return?0?end";49????/**50?????*?續約lua腳本51?????*/52????private?static?final?String?RENEW_SCRIPT?=?"if?redis.call('get',?KEYS[1])?==?ARGV[1]?then?return?redis.call('expire',?KEYS[1],?ARGV[2])?else?return?0?end";535455????private?final?StringRedisTemplate?redisTemplate;5657????public?RedisDistributionLockPlus(StringRedisTemplate?redisTemplate)?{58????????this.redisTemplate?=?redisTemplate;59????????ScheduleTask?task?=?new?ScheduleTask(this,?lockContentMap);60????????//?啟動定時任務61????????ScheduleExecutor.schedule(task,?1,?1,?TimeUnit.SECONDS);62????}6364????/**65?????*?加鎖66?????*?取到鎖加鎖，取不到鎖一直等待知道獲得鎖67?????*68?????*?@param?lockKey69?????*?@param?requestId?全局唯一70?????*?@param?expire???鎖過期時間,?單位秒71?????*?@return72?????*/73????public?boolean?lock(String?lockKey,?String?requestId,?long?expire)?{74????????log.info("開始執行加鎖,?lockKey?={},?requestId={}",?lockKey,?requestId);75????????for?(;?;?)?{76????????????//?判斷是否已經有線程持有鎖，減少redis的壓力77????????????LockContent?lockContentOld?=?lockContentMap.get(lockKey);78????????????boolean?unLocked?=?null?==?lockContentOld;79????????????//?如果沒有被鎖，就獲取鎖80????????????if?(unLocked)?{81????????????????long?startTime?=?System.currentTimeMillis();82????????????????//?計算超時時間83????????????????long?bizExpire?=?expire?==?0L???DEFAULT_LOCK_TIMEOUT?:?expire;84????????????????String?lockKeyRenew?=?lockKey?+?"_renew";8586????????????????RedisScript<Long>?script?=?RedisScript.of(LOCK_SCRIPT,?Long.class);87????????????????List<String>?keys?=?new?ArrayList<>();88????????????????keys.add(lockKey);89????????????????keys.add(lockKeyRenew);90????????????????Long?lockExpire?=?redisTemplate.execute(script,?keys,?requestId,?Long.toString(bizExpire));91????????????????if?(null?!=?lockExpire?&&?lockExpire?>?0)?{92????????????????????//?將鎖放入map93????????????????????LockContent?lockContent?=?new?LockContent();94????????????????????lockContent.setStartTime(startTime);95????????????????????lockContent.setLockExpire(lockExpire);96????????????????????lockContent.setExpireTime(startTime?+?lockExpire?*?1000);97????????????????????lockContent.setRequestId(requestId);98????????????????????lockContent.setThread(Thread.currentThread());99????????????????????lockContent.setBizExpire(bizExpire); 100????????????????????lockContent.setLockCount(1); 101????????????????????lockContentMap.put(lockKey,?lockContent); 102????????????????????log.info("加鎖成功,?lockKey?={},?requestId={}",?lockKey,?requestId); 103????????????????????return?true; 104????????????????} 105????????????} 106????????????//?重復獲取鎖，在線程池中由于線程復用，線程相等并不能確定是該線程的鎖 107????????????if?(Thread.currentThread()?==?lockContentOld.getThread() 108??????????????????????&&?requestId.equals(lockContentOld.getRequestId())){ 109????????????????//?計數?+1 110????????????????lockContentOld.setLockCount(lockContentOld.getLockCount()+1); 111????????????????return?true; 112????????????} 113 114????????????//?如果被鎖或獲取鎖失敗，則等待100毫秒 115????????????try?{ 116????????????????TimeUnit.MILLISECONDS.sleep(100); 117????????????}?catch?(InterruptedException?e)?{ 118????????????????//?這里用lombok?有問題 119????????????????log.error("獲取redis?鎖失敗,?lockKey?={},?requestId={}",?lockKey,?requestId,?e); 120????????????????return?false; 121????????????} 122????????} 123????} 124 125 126????/** 127?????*?解鎖 128?????* 129?????*?@param?lockKey 130?????*?@param?lockValue 131?????*/ 132????public?boolean?unlock(String?lockKey,?String?lockValue)?{ 133????????String?lockKeyRenew?=?lockKey?+?"_renew"; 134????????LockContent?lockContent?=?lockContentMap.get(lockKey); 135 136????????long?consumeTime; 137????????if?(null?==?lockContent)?{ 138????????????consumeTime?=?0L; 139????????}?else?if?(lockValue.equals(lockContent.getRequestId()))?{ 140????????????int?lockCount?=?lockContent.getLockCount(); 141????????????//?每次釋放鎖，?計數?-1，減到0時刪除redis上的key 142????????????if?(--lockCount?>?0)?{ 143????????????????lockContent.setLockCount(lockCount); 144????????????????return?false; 145????????????} 146????????????consumeTime?=?(System.currentTimeMillis()?-?lockContent.getStartTime())?/?1000; 147????????}?else?{ 148????????????log.info("釋放鎖失敗，不是自己的鎖。"); 149????????????return?false; 150????????} 151 152????????//?刪除已完成key，先刪除本地緩存，減少redis壓力,?分布式鎖，只有一個，所以這里不加鎖 153????????lockContentMap.remove(lockKey); 154 155????????RedisScript<Long>?script?=?RedisScript.of(UNLOCK_SCRIPT,?Long.class); 156????????List<String>?keys?=?new?ArrayList<>(); 157????????keys.add(lockKey); 158????????keys.add(lockKeyRenew); 159 160????????Long?result?=?redisTemplate.execute(script,?keys,?lockValue,?Long.toString(consumeTime), 161????????????????Long.toString(lockContent.getBizExpire())); 162????????return?EXEC_SUCCESS.equals(result); 163 164????} 165 166????/** 167?????*?續約 168?????* 169?????*?@param?lockKey 170?????*?@param?lockContent 171?????*?@return?true:續約成功，false:續約失敗（1、續約期間執行完成，鎖被釋放?2、不是自己的鎖，3、續約期間鎖過期了（未解決）） 172?????*/ 173????public?boolean?renew(String?lockKey,?LockContent?lockContent)?{ 174 175????????//?檢測執行業務線程的狀態 176????????Thread.State?state?=?lockContent.getThread().getState(); 177????????if?(Thread.State.TERMINATED?==?state)?{ 178????????????log.info("執行業務的線程已終止,不再續約?lockKey?={},?lockContent={}",?lockKey,?lockContent); 179????????????return?false; 180????????} 181 182????????String?requestId?=?lockContent.getRequestId(); 183????????long?timeOut?=?(lockContent.getExpireTime()?-?lockContent.getStartTime())?/?1000; 184 185????????RedisScript<Long>?script?=?RedisScript.of(RENEW_SCRIPT,?Long.class); 186????????List<String>?keys?=?new?ArrayList<>(); 187????????keys.add(lockKey); 188 189????????Long?result?=?redisTemplate.execute(script,?keys,?requestId,?Long.toString(timeOut)); 190????????log.info("續約結果，True成功，False失敗?lockKey?={},?result={}",?lockKey,?EXEC_SUCCESS.equals(result)); 191????????return?EXEC_SUCCESS.equals(result); 192????} 193 194 195????static?class?ScheduleExecutor?{ 196 197????????public?static?void?schedule(ScheduleTask?task,?long?initialDelay,?long?period,?TimeUnit?unit)?{ 198????????????long?delay?=?unit.toMillis(initialDelay); 199????????????long?period_?=?unit.toMillis(period); 200????????????//?定時執行 201????????????new?Timer("Lock-Renew-Task").schedule(task,?delay,?period_); 202????????} 203????} 204 205????static?class?ScheduleTask?extends?TimerTask?{ 206 207????????private?final?RedisDistributionLockPlus?redisDistributionLock; 208????????private?final?Map<String,?LockContent>?lockContentMap; 209 210????????public?ScheduleTask(RedisDistributionLockPlus?redisDistributionLock,?Map<String,?LockContent>?lockContentMap)?{ 211????????????this.redisDistributionLock?=?redisDistributionLock; 212????????????this.lockContentMap?=?lockContentMap; 213????????} 214 215????????@Override 216????????public?void?run()?{ 217????????????if?(lockContentMap.isEmpty())?{ 218????????????????return; 219????????????} 220????????????Set<Map.Entry<String,?LockContent>>?entries?=?lockContentMap.entrySet(); 221????????????for?(Map.Entry<String,?LockContent>?entry?:?entries)?{ 222????????????????String?lockKey?=?entry.getKey(); 223????????????????LockContent?lockContent?=?entry.getValue(); 224????????????????long?expireTime?=?lockContent.getExpireTime(); 225????????????????//?減少線程池中任務數量 226????????????????if?((expireTime?-?System.currentTimeMillis())/?1000?<?TIME_SECONDS_FIVE)?{ 227????????????????????//線程池異步續約 228????????????????????ThreadPool.submit(()?->?{ 229????????????????????????boolean?renew?=?redisDistributionLock.renew(lockKey,?lockContent); 230????????????????????????if?(renew)?{ 231????????????????????????????long?expireTimeNew?=?lockContent.getStartTime()?+?(expireTime?-?lockContent.getStartTime())?*?2?-?TIME_SECONDS_FIVE?*?1000; 232????????????????????????????lockContent.setExpireTime(expireTimeNew); 233????????????????????????}?else?{ 234????????????????????????????//?續約失敗，說明已經執行完?OR?redis?出現問題 235????????????????????????????lockContentMap.remove(lockKey); 236????????????????????????} 237????????????????????}); 238????????????????} 239????????????} 240????????} 241????} 242}

五、redis主從復制的坑

redis高可用最常見的方案就是主從復制（master-slave），這種模式也給redis分布式鎖挖了一坑。

redis cluster集群環境下，假如現在A客戶端想要加鎖，它會根據路由規則選擇一臺master節點寫入key?mylock，在加鎖成功后，master節點會把key異步復制給對應的slave節點。

如果此時redis master節點宕機，為保證集群可用性，會進行主備切換，slave變為了redis master。B客戶端在新的master節點上加鎖成功，而A客戶端也以為自己還是成功加了鎖的。

此時就會導致同一時間內多個客戶端對一個分布式鎖完成了加鎖，導致各種臟數據的產生。

至于解決辦法嘛，目前看還沒有什么根治的方法，只能盡量保證機器的穩定性，減少發生此事件的概率。

總結

上面就是我在使用Redis?分布式鎖時遇到的一些坑，有點小感慨，經常用一個方法填上這個坑，沒多久就發現另一個坑又出來了，其實根本沒有什么十全十美的解決方案，哪有什么銀彈，只不過是在權衡利弊后，選一個在接受范圍內的折中方案而已。

有道無術，術可成；有術無道，止于術

歡迎大家關注Java之道公眾號

好文章，我在看??

新人創作打卡挑戰賽發博客就能抽獎！定制產品紅包拿不停！

總結

以上是生活随笔為你收集整理的redis 分布式锁的 5个坑，真是又大又深的全部內容，希望文章能夠幫你解決所遇到的問題。

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