??點(diǎn)擊上方?好好學(xué)java?，選擇?星標(biāo)?公眾號(hào)

重磅資訊、干貨，第一時(shí)間送達(dá) 今日推薦：我的大學(xué)到研究生自學(xué) Java 之路，過(guò)程艱辛，不放棄，保持熱情，最終發(fā)現(xiàn)我是這樣拿到大廠 offer 的！作者：你的笑像一條狗 鏈接：https://segmentfault.com/a/1190000022533998

基于Redis實(shí)現(xiàn)的分布式鎖

Spring Cloud 分布式環(huán)境下，同一個(gè)服務(wù)都是部署在不同的機(jī)器上，這種情況無(wú)法像單體架構(gòu)下數(shù)據(jù)一致性問(wèn)題采用加鎖就實(shí)現(xiàn)數(shù)據(jù)一致性問(wèn)題，在高并發(fā)情況下，對(duì)于分布式架構(gòu)顯然是不合適的，針對(duì)這種情況我們就需要用到分布式鎖了。

哪些場(chǎng)景需要用分布式鎖

場(chǎng)景一:比較敏感的數(shù)據(jù)比如金額修改，同一時(shí)間只能有一個(gè)人操作，想象下2個(gè)人同時(shí)修改金額，一個(gè)加金額一個(gè)減金額，為了防止同時(shí)操作造成數(shù)據(jù)不一致，需要鎖，如果是數(shù)據(jù)庫(kù)需要的就是行鎖或表鎖，如果是在集群里，多個(gè)客戶端同時(shí)修改一個(gè)共享的數(shù)據(jù)就需要分布式鎖。

場(chǎng)景二:比如多臺(tái)機(jī)器都可以定時(shí)執(zhí)行某個(gè)任務(wù)，如果限制任務(wù)每次只能被一臺(tái)機(jī)器執(zhí)行，不能重復(fù)執(zhí)行，就可以用分布式鎖來(lái)做標(biāo)記。

場(chǎng)景三:比如秒殺場(chǎng)景，要求并發(fā)量很高，那么同一件商品只能被一個(gè)用戶搶到，那么就可以使用分布式鎖實(shí)現(xiàn)。

分布式鎖實(shí)現(xiàn)方式：

• 1、基于數(shù)據(jù)庫(kù)實(shí)現(xiàn)分布式鎖

• 2、基于緩存（redis，memcached，tair）實(shí)現(xiàn)分布式鎖

• 3、基于Zookeeper實(shí)現(xiàn)分布式鎖

為什么不使用數(shù)據(jù)庫(kù)？

數(shù)據(jù)庫(kù)是單點(diǎn)？搞兩個(gè)數(shù)據(jù)庫(kù)，數(shù)據(jù)之前雙向同步。一旦掛掉快速切換到備庫(kù)上。

沒(méi)有失效時(shí)間？只要做一個(gè)定時(shí)任務(wù)，每隔一定時(shí)間把數(shù)據(jù)庫(kù)中的超時(shí)數(shù)據(jù)清理一遍。

非阻塞的？搞一個(gè)while循環(huán)，直到insert成功再返回成功。

非重入的？在數(shù)據(jù)庫(kù)表中加個(gè)字段，記錄當(dāng)前獲得鎖的機(jī)器的主機(jī)信息和線程信息，那么下次再獲取鎖的時(shí)候先查詢數(shù)據(jù)庫(kù)，如果當(dāng)前機(jī)器的主機(jī)信息和線程信息在數(shù)據(jù)庫(kù)可以查到的話，直接把鎖分配給他就可以了。

大量請(qǐng)求下數(shù)據(jù)庫(kù)往往是系統(tǒng)的瓶頸，大量連接，然后sql查詢，幾乎所有時(shí)間都浪費(fèi)到這些上面，所以往往情況下能內(nèi)存操作就在內(nèi)存操作，使用基于內(nèi)存操作的Redis實(shí)現(xiàn)分布式鎖，也可以根據(jù)需求選擇ZooKeeper 來(lái)實(shí)現(xiàn)。

通過(guò) Redis 的 Redlock 和 ZooKeeper 來(lái)加鎖，性能有了比較大的提升，一般情況我們根據(jù)實(shí)際場(chǎng)景選擇使用。

分布式鎖應(yīng)該滿足要求

• 互斥性 可以保證在分布式部署的應(yīng)用集群中，同一個(gè)方法在同一時(shí)間只能被一臺(tái)機(jī)器上的一個(gè)線程執(zhí)行。

• 這把鎖要是一把可重入鎖（避免死鎖）

• 不會(huì)發(fā)生死鎖：有一個(gè)客戶端在持有鎖的過(guò)程中崩潰而沒(méi)有解鎖，也能保證其他客戶端能夠加鎖

• 這把鎖最好是一把阻塞鎖（根據(jù)業(yè)務(wù)需求考慮要不要這條）

• 有高可用的獲取鎖和釋放鎖功能

• 獲取鎖和釋放鎖的性能要好

Redis實(shí)現(xiàn)分布式鎖

Redis實(shí)現(xiàn)分布式鎖利用 SETNX 和 SETEX

基本命令主要有：

• SETNX(SET If Not Exists)：當(dāng)且僅當(dāng) Key 不存在時(shí)，則可以設(shè)置，否則不做任何動(dòng)作。

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

• SETEX：基于SETNX功能外,還可以設(shè)置超時(shí)時(shí)間，防止死鎖。

分布式鎖

分布式鎖其實(shí)大白話，本質(zhì)上要實(shí)現(xiàn)的目標(biāo)(客戶端)在redis中占一個(gè)位置，等到這個(gè)客戶試用，別的人進(jìn)來(lái)就必須得等著，等我試用完了，走了，你再來(lái)。 感覺(jué)跟多線程鎖一樣，意思大致是一樣的，多線程是針對(duì)單機(jī)的，在同一個(gè)Jvm中，但是分布式石鎖，是跨機(jī)器的，多個(gè)進(jìn)程不同機(jī)器上發(fā)來(lái)得請(qǐng)求，去對(duì)同一個(gè)數(shù)據(jù)進(jìn)行操作。

比如，分布式架構(gòu)下的秒殺系統(tǒng)，幾萬(wàn)人對(duì)10個(gè)商品進(jìn)行搶購(gòu)，10個(gè)商品存在redis中，就是表示10個(gè)位置，第一個(gè)人進(jìn)來(lái)了，商品就剩9個(gè)了，第二個(gè)人進(jìn)來(lái)就剩8個(gè)，在第一個(gè)人進(jìn)來(lái)的時(shí)候，其他人必須等到10個(gè)商品數(shù)量成功減去1之后你才能進(jìn)來(lái)。

這個(gè)過(guò)程中第一個(gè)人進(jìn)來(lái)的時(shí)候還沒(méi)操作減1然后異常了，沒(méi)有釋放鎖，然后后面人一直等待著，這就是死鎖。真對(duì)這種情況可以設(shè)置超時(shí)時(shí)間，如果超過(guò)10s中還是沒(méi)出來(lái)，就讓他超時(shí)失效。

redis中提供了 setnx(set if not exists) 指令

> setnx lock:codehole true -- 鎖定 OK ... do something xxxx... 數(shù)量減1 > del lock:codehole -- 釋放鎖 (integer) 1 --成功

如果在減1期間發(fā)生異常 del 指令沒(méi)有被調(diào)用 然后就一直等著，鎖永遠(yuǎn)不會(huì)釋放。

redis Redis 2.8 版本中提供了 setex(set if not exists) 指令 setnx 和 expire 兩個(gè)指令構(gòu)成一個(gè)原子操作 給鎖加上一個(gè)過(guò)期時(shí)間

> setex lock:codehole true OK > expire lock:codehole 5 ... do something xxxx ... > del lock:codehole (integer) 1

SETEX 實(shí)現(xiàn)原理

通過(guò) SETNX 設(shè)置 Key-Value 來(lái)獲得鎖，隨即進(jìn)入死循環(huán)，每次循環(huán)判斷，如果存在 Key 則繼續(xù)循環(huán)，如果不存在 Key，則跳出循環(huán)，當(dāng)前任務(wù)執(zhí)行完成后，刪除 Key 以釋放鎖。

實(shí)現(xiàn)步驟

pom.xml 導(dǎo)入Redis依賴

<!-- redis--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.16.10</version><scope>provided</scope></dependency>

添加配置文件 application.yml：

server:port: 8080spring:profiles: devdata:redis:# Redis數(shù)據(jù)庫(kù)索引（默認(rèn)為0）database: 0# Redis服務(wù)器地址host: 127.0.0.1# Redis服務(wù)器連接端口port: 6379# Redis服務(wù)器連接密碼（默認(rèn)為空）password:

全局鎖類

@Data public?class?Lock?{/***?key名*/private?String?name;/***?value值*/private?String?value;public?Lock(String?name,?String?value)?{this.name?=?name;this.value?=?value;}}

分布式鎖類

@Slf4j @Component public?class?DistributedLockConfig?{/***?單個(gè)業(yè)務(wù)持有鎖的時(shí)間30s，防止死鎖*/private?final?static?long?LOCK_EXPIRE?=?30?*?1000L;/***?默認(rèn)30ms嘗試一次*/private?final?static?long?LOCK_TRY_INTERVAL?=?30L;/***?默認(rèn)嘗試20s*/private?final?static?long?LOCK_TRY_TIMEOUT?=?20?*?1000L;private?RedisTemplate?template;public?void?setTemplate(RedisTemplate?template)?{this.template?=?template;}/***?嘗試獲取全局鎖**?@param?lock?鎖的名稱*?@return?true?獲取成功，false獲取失敗*/public?boolean?tryLock(Lock?lock)?{return?getLock(lock,?LOCK_TRY_TIMEOUT,?LOCK_TRY_INTERVAL,?LOCK_EXPIRE);}/***?嘗試獲取全局鎖*?SETEX：可以設(shè)置超時(shí)時(shí)間**?@param?lock????鎖的名稱*?@param?timeout?獲取超時(shí)時(shí)間?單位ms*?@return?true?獲取成功，false獲取失敗*/public?boolean?tryLock(Lock?lock,?long?timeout)?{return?getLock(lock,?timeout,?LOCK_TRY_INTERVAL,?LOCK_EXPIRE);}/***?嘗試獲取全局鎖**?@param?lock????????鎖的名稱*?@param?timeout?????獲取鎖的超時(shí)時(shí)間*?@param?tryInterval?多少毫秒嘗試獲取一次*?@return?true?獲取成功，false獲取失敗*/public?boolean?tryLock(Lock?lock,?long?timeout,?long?tryInterval)?{return?getLock(lock,?timeout,?tryInterval,?LOCK_EXPIRE);}/***?嘗試獲取全局鎖**?@param?lock???????????鎖的名稱*?@param?timeout????????獲取鎖的超時(shí)時(shí)間*?@param?tryInterval????多少毫秒嘗試獲取一次*?@param?lockExpireTime?鎖的過(guò)期*?@return?true?獲取成功，false獲取失敗*/public?boolean?tryLock(Lock?lock,?long?timeout,?long?tryInterval,?long?lockExpireTime)?{return?getLock(lock,?timeout,?tryInterval,?lockExpireTime);}/***?操作redis獲取全局鎖**?@param?lock???????????鎖的名稱*?@param?timeout????????獲取的超時(shí)時(shí)間*?@param?tryInterval????多少ms嘗試一次*?@param?lockExpireTime?獲取成功后鎖的過(guò)期時(shí)間*?@return?true?獲取成功，false獲取失敗*/public?boolean?getLock(Lock?lock,?long?timeout,?long?tryInterval,?long?lockExpireTime)?{try?{if?(StringUtils.isEmpty(lock.getName())?||?StringUtils.isEmpty(lock.getValue()))?{return?false;}long?startTime?=?System.currentTimeMillis();do?{if?(!template.hasKey(lock.getName()))?{ValueOperations<String,?String>?ops?=?template.opsForValue();ops.set(lock.getName(),?lock.getValue(),?lockExpireTime,?TimeUnit.MILLISECONDS);return?true;}?else?{//存在鎖log.debug("lock?is?exist!！！");}//嘗試超過(guò)了設(shè)定值之后直接跳出循環(huán)if?(System.currentTimeMillis()?-?startTime?>?timeout)?{return?false;}//每隔多長(zhǎng)時(shí)間嘗試獲取Thread.sleep(tryInterval);}while?(template.hasKey(lock.getName()));}?catch?(InterruptedException?e)?{log.error(e.getMessage());return?false;}return?false;}/***?獲取鎖*?SETNX(SET?If?Not?Exists)：當(dāng)且僅當(dāng)?Key?不存在時(shí)，則可以設(shè)置，否則不做任何動(dòng)作。*/public?Boolean?getLockNoTime(Lock?lock)?{if?(!StringUtils.isEmpty(lock.getName()))?{return?false;}//?setIfAbsent?底層封裝命令?是?setNX()boolean?falg?=?template.opsForValue().setIfAbsent(lock.getName(),?lock.getValue());return?false;}/***?釋放鎖*/public?void?releaseLock(Lock?lock)?{if?(!StringUtils.isEmpty(lock.getName()))?{template.delete(lock.getName());}}}

測(cè)試方法

@RequestMapping("test")public?String?index()?{distributedLockConfig.setTemplate(redisTemplate);Lock?lock?=?new?Lock("test",?"test");if?(distributedLockConfig.tryLock(lock))?{try?{//為了演示鎖的效果，這里睡眠5000毫秒System.out.println("執(zhí)行方法");Thread.sleep(5000);}?catch?(Exception?e)?{e.printStackTrace();}distributedLockConfig.releaseLock(lock);}return?"hello?world!";}

開啟兩個(gè)瀏覽器窗口，執(zhí)行方法，我們可以看到兩個(gè)瀏覽器在等待執(zhí)行，當(dāng)一個(gè)返回 hello world! 之后，如果沒(méi)超時(shí)執(zhí)行另一個(gè)也會(huì)返回hello world! 兩個(gè)方法彼此先后返回，說(shuō)明分布式鎖執(zhí)行成功。

但是存在一個(gè)問(wèn)題：

這段方法是先去查詢key是否存在redis中，如果存在走循環(huán)，然后根據(jù)間隔時(shí)間去等待嘗試獲取，如果不存在則進(jìn)行獲取鎖，如果等待時(shí)間超過(guò)超時(shí)時(shí)間返回false。

• 1 這種方式性能問(wèn)題很差，每次獲取鎖都要進(jìn)行等待，很是浪費(fèi)資源，

• 2 如果在判斷鎖是否存在這兒2個(gè)或者2個(gè)以上的線程都查到redis中存在key，同一時(shí)刻就無(wú)法保證一個(gè)客戶端持有鎖，不具有排他性。

如果在集群環(huán)境下也會(huì)存在問(wèn)題

假如在哨兵模式中 主節(jié)點(diǎn)獲取到鎖之后,數(shù)據(jù)沒(méi)有同步到從節(jié)點(diǎn)主節(jié)點(diǎn)掛掉了，這樣數(shù)據(jù)完整性不能保證，另一個(gè)客戶端請(qǐng)求過(guò)來(lái)，就會(huì)一把鎖被兩個(gè)客戶端持有，會(huì)導(dǎo)致數(shù)據(jù)一致性出問(wèn)題。

在這里插入圖片描述

對(duì)此Redis中還提供了另外一種實(shí)現(xiàn)分布式鎖的方法 Redlock

利用 Redlock

Redlock是redis官方提出的實(shí)現(xiàn)分布式鎖管理器的算法。這個(gè)算法會(huì)比一般的普通方法更加安全可靠。

為什么選擇紅鎖？ 在集群中需要半數(shù)以上的節(jié)點(diǎn)同意才能獲得鎖，保證了數(shù)據(jù)的完整性，不會(huì)因?yàn)橹鞴?jié)點(diǎn)數(shù)據(jù)存在，主節(jié)點(diǎn)掛了之后沒(méi)有同步到從節(jié)點(diǎn)，導(dǎo)致數(shù)據(jù)丟失。

Redlock 算法

使用場(chǎng)景

對(duì)于Redis集群模式盡量采用這種分布式鎖，保證高可用，數(shù)據(jù)一致性，就使用Redlock 分布式鎖。

pom.xml 增加依賴

<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.7.0</version> </dependency>

獲取鎖后需要處理的邏輯

/***?獲取鎖后需要處理的邏輯*/ public?interface?AquiredLockWorker<T>?{T?invokeAfterLockAquire()?throws?Exception; }

獲取鎖管理類

/***?獲取鎖管理類*/ public?interface?DistributedLocker?{/***?獲取鎖*?@param?resourceName??鎖的名稱*?@param?worker?獲取鎖后的處理類*?@param?<T>*?@return?處理完具體的業(yè)務(wù)邏輯要返回的數(shù)據(jù)*?@throws?UnableToAquireLockException*?@throws?Exception*/<T>?T?lock(String?resourceName,?AquiredLockWorker<T>?worker)?throws?UnableToAquireLockException,?Exception;<T>?T?lock(String?resourceName,?AquiredLockWorker<T>?worker,?int?lockTime)?throws?UnableToAquireLockException,?Exception;}

異常類

/***?異常類*/ public?class?UnableToAquireLockException?extends?RuntimeException?{public?UnableToAquireLockException()?{}public?UnableToAquireLockException(String?message)?{super(message);}public?UnableToAquireLockException(String?message,?Throwable?cause)?{super(message,?cause);} }

獲取RedissonClient連接類

/***?獲取RedissonClient連接類*/ @Component public?class?RedissonConnector?{RedissonClient?redisson;@PostConstructpublic?void?init(){redisson?=?Redisson.create();}public?RedissonClient?getClient(){return?redisson;}}

分布式鎖實(shí)現(xiàn)

@Component public?class?RedisLocker??implements?DistributedLocker{private?final?static?String?LOCKER_PREFIX?=?"lock:";@AutowiredRedissonConnector?redissonConnector;@Overridepublic?<T>?T?lock(String?resourceName,?AquiredLockWorker<T>?worker)?throws?InterruptedException,?UnableToAquireLockException,?Exception?{return?lock(resourceName,?worker,?100);}@Overridepublic?<T>?T?lock(String?resourceName,?AquiredLockWorker<T>?worker,?int?lockTime)?throws?UnableToAquireLockException,?Exception?{RedissonClient?redisson=?redissonConnector.getClient();RLock?lock?=?redisson.getLock(LOCKER_PREFIX?+?resourceName);//?Wait?for?100?seconds?seconds?and?automatically?unlock?it?after?lockTime?secondsboolean?success?=?lock.tryLock(100,?lockTime,?TimeUnit.SECONDS);if?(success)?{try?{return?worker.invokeAfterLockAquire();}?finally?{lock.unlock();}}throw?new?UnableToAquireLockException();} }

測(cè)試方法

ScheduledExecutorService?scheduledExecutorService?=?Executors.newScheduledThreadPool(10);for?(int?i?=?0;?i?<?50;?i++)?{scheduledExecutorService.execute(new?Worker());}scheduledExecutorService.shutdown();//任務(wù)class?Worker?implements?Runnable?{public?Worker()?{}@Overridepublic?void?run()?{try?{redisLocker.lock("tizz1100",?new?AquiredLockWorker<Object>()?{@Overridepublic?Object?invokeAfterLockAquire()?{doTask();return?null;}});}?catch?(Exception?e)?{}}void?doTask()?{System.out.println(Thread.currentThread().getName()?+?"?----------?"?+?LocalDateTime.now());System.out.println(Thread.currentThread().getName()?+?"?start");Random?random?=?new?Random();int?_int?=?random.nextInt(200);System.out.println(Thread.currentThread().getName()?+?"?sleep?"?+?_int?+?"millis");try?{Thread.sleep(_int);}?catch?(InterruptedException?e)?{e.printStackTrace();}System.out.println(Thread.currentThread().getName()?+?"?end");}}

參考資料:

https://blog.csdn.net/yue_201...

https://blog.csdn.net/weixin_...

https://github.com/pomestyle/SpringBoot/tree/master/Springboot-Redis-SETEX

最后，再附上我歷時(shí)三個(gè)月總結(jié)的?Java 面試 + Java 后端技術(shù)學(xué)習(xí)指南，這是本人這幾年及春招的總結(jié)，目前，已經(jīng)拿到了騰訊等大廠offer，拿去不謝，github 地址：https://github.com/OUYANGSIHAI/JavaInterview

這么辛苦總結(jié)，給個(gè)star好不好。?點(diǎn)擊閱讀原文，直達(dá)

總結(jié)

以上是生活随笔為你收集整理的基于 Redis 实现的分布式锁的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。