柚子快報激活碼778899分享：Redisson實現(xiàn)分布式鎖

http://yzkb.51969.com/

一、分布式鎖使用場景

隨著互聯(lián)網(wǎng)技術(shù)的不斷發(fā)展，數(shù)據(jù)量的不斷增加，業(yè)務(wù)邏輯日趨復(fù)雜，在這種背景下，傳統(tǒng)的集中式系統(tǒng)已經(jīng)無法滿足我們的業(yè)務(wù)需求，分布式系統(tǒng)被應(yīng)用在更多的場景，而在分布式系統(tǒng)中訪問共享資源就需要一種互斥機制，來防止彼此之間的互相干擾，以保證一致性，在這種情況下，我們就需要用到分布式鎖

首先我們先來看一個小例子：

　　假設(shè)某商城有一個商品庫存剩10個，用戶A想要買6個，用戶B想要買5個，在理想狀態(tài)下，用戶A先買走了6了，庫存減少6個還剩4個，此時用戶B應(yīng)該無法購買5個，給出數(shù)量不足的提示；而在真實情況下，用戶A和B同時獲取到商品剩10個，A買走6個，在A更新庫存之前，B又買走了5個，此時B更新庫存，商品還剩5個，這就是典型的電商“秒殺”活動。

案例：　

@RestController

public class HelloController {

//數(shù)據(jù)庫中的庫存

private Integer num=10;

@GetMapping("/")

public Integer secondsKill(int count){

synchronized (num){

//是否有庫存

if(num>0){

//減庫存

num=num-count;

}

}

return num;

}

}

?

思考：將該項目啟動兩個實例，8080實例購買6個，8081實例購買5個此時會有什么問題

　　從上述例子不難看出，在高并發(fā)情況下，如果不做處理將會出現(xiàn)各種不可預(yù)知的后果。那么在這種高并發(fā)多線程的情況下，解決問題最有效最普遍的方法就是給共享資源或?qū)蚕碣Y源的操作加一把鎖，來保證對資源的訪問互斥。在Java JDK已經(jīng)為我們提供了這樣的鎖，利用ReentrantLock或者synchronized，即可達到資源互斥訪問的目的。但是在分布式系統(tǒng)中，由于分布式系統(tǒng)的分布性，即多線程和多進程并且分布在不同機器中，也就是說一個服務(wù)可以同時啟動多個實例，不同用戶訪問不同的實例，那么這兩種鎖將失去原有鎖的效果，需要我們自己實現(xiàn)分布式鎖——分布式鎖?！?/p>

一般我們使用分布式鎖有兩個場景：

效率：使用分布式鎖可以避免不同節(jié)點重復(fù)相同的工作，這些工作會浪費資源。比如用戶付了錢之后有可能不同節(jié)點會發(fā)出多封短信。 正確性：加分布式鎖同樣可以避免破壞正確性的發(fā)生，如果兩個節(jié)點在同一條數(shù)據(jù)上面操作，比如多個節(jié)點機器對同一個訂單操作不同的流程有可能會導(dǎo)致該筆訂單最后狀態(tài)出現(xiàn)錯誤，造成損失。

Redis 因為其性能好，實現(xiàn)起來分布式鎖簡單，所以讓很多人都對基于 Redis 實現(xiàn)的分布式鎖十分青睞。

提示

除了能使用 Redis 實現(xiàn)分布式鎖之外，Zookeeper 也能實現(xiàn)分布式鎖。但是項目中不可能僅僅為了實現(xiàn)分布式鎖而專門引入 Zookeeper ，所以，除非你的項目體系中本來就有 Zookeeper（來實現(xiàn)其它功能），否則不會單獨因為分布式鎖而引入它

二、使用Redis的Setnx命令實現(xiàn)分布式鎖

步驟分析

1、每次用戶請求下單時，就在redis中設(shè)置一個鍵值對，如果設(shè)置成功，就執(zhí)行下單操作流程

2、如果下單失敗，就讓你等待，等到前面的人下單完成后將該鍵刪除，你再下單

1、SETNX 命令

早期，SETNX 是獨立于 SET 命令之外的另一條命令。它的意思是 SET if Not eXists，即，在鍵值對不存在的時候才能設(shè)值成功。

注意:SETNX 命令的價值在于：它將 判斷 和 設(shè)值 兩個操作合二為一，從而避免了 查查改改 的情況的出現(xiàn)。

后來，在 Redis 2013 年推出的 2.6.12 版本中，Redis 為 SET 命令官方提供了 NX 選項，使得 SET 命令也能實現(xiàn) SETNX 命令的功能。其語法如下：

?SET [EX seconds] [PX milliseconds] [NX | XX]

EX 值的是 key 的存活時間，單位為秒。PX 與 EX 作用一樣，唯一的不同就是后者的單位是微秒（使用較少）。

NX 和 XX 作用是相反的。NX 表示只有當(dāng) key『不存在時』才會設(shè)置其值；XX 表示當(dāng) key 存在時才設(shè)置 key 的值。

在 “升級” 了 SET 命令之后，Redis 官方說：“由于 SET 命令選項可以替換 SETNX，SETEX，因此在 Redis 的將來版本中，這二個命令可能會被棄用并最終刪除”。

所以，現(xiàn)在我們口頭所說的 SETNX 命令，并非單指 SETNX 命令，而是包括帶 NX 選項的 SET 命令（甚至以后就沒有 SETNX 命令了）

2、SETNX 的使用

在使用 SETNX 操作實現(xiàn)分布式鎖功能時，需要注意以下幾點：

這里的『鎖』指的是 Redis 中的一個約定的鍵值對。誰能創(chuàng)建這個鍵值對，就意味著誰擁有這整個『鎖』。 使用 SETNX 命令獲取『鎖』時，如果操作返回結(jié)果是 0（表示 key 已存在，設(shè)值失?。?，則意味著獲取『鎖』失?。ㄔ撴i被其它線程先獲?。?，反之，則設(shè)值成功，表示獲取『鎖』成功。

如果這個 key 不存在，SETNX 才會設(shè)置該 key 的值。此時 Redis 返回 1 。 如果這個 key 存在，SETNX 則不會設(shè)置該 key 的值。此時 Redis 返回 0 。 為了防止其它線程獲得『鎖』之后，有意或無意，長期持有『鎖』而不釋放（導(dǎo)致其它線程無法獲得該『鎖』）。因此，需要為 key 設(shè)置一個合理的過期時間。 當(dāng)成功獲得『鎖』并成功完成響應(yīng)操作之后，需要釋放『鎖』（可以執(zhí)行 DEL 命令將『鎖』刪除）。

在代碼層面，與 Setnx 命令對應(yīng)的接口是 ValueOperations 的 setIfAbsent 方法

示例代碼一：

@RestController

public class HelloController {

@Autowired

private StringRedisTemplate redisTemplate;

@SneakyThrows

@GetMapping("/add")

public String addGoods(String goodsname) {

String message="";

ValueOperations operations = redisTemplate.opsForValue();

//在redis中設(shè)置一個鍵，表示有人正在操作庫存

boolean flag = operations.setIfAbsent(goodsname, "y",30, TimeUnit.SECONDS);

if (flag) {

Thread.sleep(10000);

try {

message= "修改了商品信息";

} catch (Exception e) {

//刪除redis中的鍵

redisTemplate.delete(goodsname);

}

}else{

message= "其他人正在使用，請稍后重試";

}

return message;

}

}

示例代碼二：

@SneakyThrows

@GetMapping("/add2")

public String addGoods2(String goodsname) {

String message = "";

ValueOperations operations = redisTemplate.opsForValue();

//在redis中設(shè)置一個鍵，表示有人正在操作庫存，返回false表示正在有人操作此時會失敗

//如果失敗了隔兩秒重試一次

while (!operations.setIfAbsent(goodsname, "y", 30, TimeUnit.SECONDS)) {

//睡100 毫秒，繼續(xù)取set 看看是否成功

System.out.println(Thread.currentThread().getName() + ":獲取鎖失敗");

Thread.sleep(1000);

}

try {

message= "修改了商品信息"

} catch (Exception e) {

//刪除redis中的鍵

redisTemplate.delete(goodsname);

}

return message;

}

開啟兩個不同的瀏覽器發(fā)請求測試

3、SETNX命令的問題

a、死鎖問題

假設(shè)線程1通過SETNX獲取到鎖并且正常執(zhí)行然后釋放鎖那么一切ok，其它線程也能獲取到鎖。但是線程1現(xiàn)在"耍脾氣"了，線程1抱怨說"工作太久有點累需要休息一下，你們想要獲取鎖等著吧，等我把活干完你們再來獲取鎖"。此時其它線程就無法向下繼續(xù)執(zhí)行，因為鎖在線程1手中。這種長期不釋放鎖情況就有可能造成死鎖。 為了防止像線程1這種"耍脾氣"的現(xiàn)象發(fā)生，我們可以設(shè)置key的過期時間來解決。設(shè)置過期時間過后其它線程可不會慣著線程1,其它線程表示你要休息可以，休息了指定時間把鎖讓出來然后拍拍屁股走人，沒人慣著你。

上鎖時，設(shè)置的超時自動刪除時長（3 秒），設(shè)置多長合適？萬一設(shè)置短了怎么辦?如果設(shè)置短了，在業(yè)務(wù)邏輯執(zhí)行完之前時間到期，那么 Redis 自動就把鍵值對給刪除了，即，把鎖給釋放了，這不符合邏輯。

b、SETNX誤刪情況

情況一 設(shè)置過期時間線程1被治得服服帖帖，此時線程1又開始不當(dāng)人了。線程1想既然你搶我得鎖，等你獲得鎖后我就將鎖刪除畢竟我還要有備用鑰匙，讓你也鎖不住，讓其它線程也執(zhí)行。 線程1休息的時間超過了過期時間，此時鎖會自動釋放。線程2現(xiàn)在脫穎而出搶到了鎖然后開心的繼續(xù)執(zhí)行。但是現(xiàn)在線程1醒了，發(fā)現(xiàn)線程2搶走了鎖。線程1表示小子膽挺肥啊，敢搶我的鎖，等我執(zhí)行完了就將你鎖刪除，讓其它"哥們"也進來。此時就會發(fā)生蝴蝶效應(yīng)，線程1刪除了線程2的鎖，線程2刪除了線程3的鎖，直到最后一個"哥們：wc，我鎖了?"。當(dāng)然線程是無感知，其實線程1乃至其它線程都不知道刪除的是別人的鎖，全部線程都以為刪除的是自己的鎖。直到最后一個線程無鎖可刪。 這種誤刪鎖的情況讓鎖的存在蕩然無存，本來應(yīng)該串行執(zhí)行的線程，在一定程度上都開始并發(fā)執(zhí)行了。 那么誤刪情況該如何解決了? 我們可以給鎖加上線程標識，只有鎖是當(dāng)前線程的才能刪除，否則不能刪除。在添加key的時候，key的value存儲當(dāng)前線程的標識，這個標識只要保證唯一即可。可以使用UUID或者一個自增數(shù)據(jù)。在刪除鎖的時候，將線程標識取出來進行判斷，如果相同就表示鎖是自己的能夠刪除，否則不能刪除。 獲取鎖 //獲取線程前綴，同時也是線程表示。通過UUID唯一性

private static final String ID_PREFIX= UUID.randomUUID().toString(true)+"-";

//與線程id組合

public boolean tryLock(long timeOut) {

//獲取線程id

String id =ID_PREFIX+ Thread.currentThread().getId();

//獲取鎖

Boolean absent = redisTemplate.opsForValue().setIfAbsent(key, id , timeOut, TimeUnit.SECONDS);

return Boolean.TRUE.equals(absent);

} 釋放鎖: public void unLock() {

//獲取存儲的線程標識

String value = stringRedisTemplate.opsForValue().get(key);

//當(dāng)前線程的線程標識

String id =ID_PREFIX+ Thread.currentThread().getId();

//線程標識相同則刪除否,則不刪除

if (id.equals(value)){

redisTemplate.delete(key);

}

} 情況二 加入線程標識后，線程一不能隨便刪除其它線程的鎖，但是線程1又開始不當(dāng)人了。線程1表示判斷線程標識和釋放鎖的操作我可以分開執(zhí)行，這又不是一個原子性的操作,線程1干完活以后就準備去釋放鎖，當(dāng)線程1判斷鎖是自己的后表示開鎖太累了,休息一會在開。此時其它線程就想無所謂，反正過期時間一到鎖就會自動釋放。但是線程1已經(jīng)判斷了鎖是自己的以后就不會執(zhí)行判斷鎖的操作(線程1已經(jīng)執(zhí)行了if判斷，只是沒有執(zhí)行方法體),當(dāng)線程2獲得鎖后，線程1仍然能刪除線程2的鎖。

?解鎖時，`查 - 刪` 操作是 2 個操作，由兩個命令完成，非原子性。 redis底層執(zhí)行這個setnx不是一個原子操作，而是有兩步操作完成的，首先set hello world，然后第二步設(shè)置key的過期時間： expire hello 3，那么如果執(zhí)行完第一步剛好redis宕機了，此時key一直保存到redis。永遠也無法刪除了。

三、Redisson實現(xiàn)分布式鎖【日常使用】

1、Redisson 如何解決上述問題

Redisson 解決 “過期自動刪除時長” 問題的思路和方案 Redisson 中客戶端一旦加鎖成功，就會啟動一個后臺線程（慣例稱之為 watch dog 看門狗）。watch dog 線程默認會每隔 10 秒檢查一下，如果鎖 key 還存在，那么它會不斷的延長鎖 key 的生存時間，直到你的代碼中去刪除鎖 key 。 Redisson 解決setnx和 解鎖的非原子性 問題的思路和方案 Redisson 的上鎖和解鎖操作都是通過 Lua 腳本實現(xiàn)的。Redis 中 執(zhí)行 Lua 腳本能保證原子性，整段 Lua 腳本的執(zhí)行是原子性的，在其執(zhí)行期間 Redis 不會再去執(zhí)行其它命令

2、Redisson 的使用

添加依賴

org.redisson

redisson

3.15.6

配置 RedissonConfig /**

* redission配置類

*/

@Configuration

public class RedissonConfig {

@Value("${spring.redis.host}")

private String host;

@Value("${spring.redis.port}")

private String port;

@Value("${spring.redis.database}")

private int database;

@Bean

public RedissonClient getRedissonClient(){

String address="redis://"+host+":"+port; //拼接redis地址

Config config = new Config();

config.useSingleServer().setAddress(address).setDatabase(database).setKeepAlive(true);

return Redisson.create(config);

}

}

a、基本用法

@RestController

public class HelloController {

@Autowired

private RedissonClient redissonClient;

@SneakyThrows

@GetMapping("/add")

public String addGoods(String goodsname) {

String message = "";

//獲取一把鎖對象，一般將需要上鎖的類名+方法名做為鎖的鍵

RLock rLock=redissonClient.getLock("HelloController.addGoods");

try{

rLock.lock(); //加鎖，其實就是設(shè)置一個key-value 默認加的鎖都是30s

}catch (Exception e){

e.printStackTrace();

return "上鎖失敗，請稍后重試";

}

try{

Thread.sleep(5000);//睡五秒

message = "執(zhí)行了添加操作";

}catch (Exception e){

e.printStackTrace();

return "上鎖失敗，請稍后重試";

}finally {

//判斷是否有鎖對象，以及是否是同一個鎖

if (rLock.isLocked() && rLock.isHeldByCurrentThread()){

rLock.unlock(); //解鎖

}

}

return message;

}

}

b、批量加鎖

RedissonMultiLock 對象可以將多個 RLock 對象關(guān)聯(lián)為一個聯(lián)鎖，每個 RLock 對象實例可以來自于不同的 Redisson 實例

@RestController

public class HelloController {

@Autowired

private RedissonClient redissonClient;

@SneakyThrows

@GetMapping("/add")

public String addGoods(String goodsname) {

String message = "";

//批量鎖對象

RLock multiLock=null;

try{

//保存批量鎖對象

List rLockList=new ArrayList();

for(String skid :skuList){

RLock rLock=redissonClient.getLock("lock:"+skid);

rLockList.add(rLock);

}

//將鎖集合轉(zhuǎn)為數(shù)組

RLock[] arrayLock = rLockList.stream().toArray(RLock[]::new);

//將多個RLock整合為一個大鎖對象

multiLock=redissonClient.getMultiLock(arrayLock);

multiLock.lock(); //上鎖

}catch (Exception e){

e.printStackTrace();

return "上鎖失敗，請稍后重試";

}

try{

message="執(zhí)行了批量操作"

}catch (Exception e){

return "批量操作失敗"

}finally{

//判斷是否有鎖對象，以及是否是同一個鎖

if (multiLock.isLocked() && multiLock.isHeldByCurrentThread()){

multiLock.unlock(); //解鎖

}

}

return message;

}

}

3、Redisson分析

你通過 RedissonClient 拿到的鎖都是 “可重入鎖” 這里的 “可重入” 的意思是：持有鎖的線程可以反復(fù)上鎖，而不會失敗，或阻塞等待；鎖的非持有者上鎖時，則會失敗，或需要等待。當(dāng)然，如果你對一個鎖反復(fù)上鎖，那么邏輯上，你應(yīng)該對它執(zhí)行同樣多次的解鎖操作 @Autowired

private RedissonClient redissonClient;

@Test

void contextLoads() {

RLock rLock = redissonClient.getLock("hello");

rLock.lock(); System.out.println("lock success!");

rLock.lock(); System.out.println("lock success!");

rLock.lock(); System.out.println("lock success!");

rLock.unlock();

rLock.unlock();

rLock.unlock();

} 使用 lock( )上鎖時由于你沒有指定過期刪除時間，所以，邏輯上只有當(dāng)你調(diào)用 unlock( )之后，Redis 中代表這個鎖的鍵值對才會被刪除。當(dāng)然你也可以在 lock 時指定超時自動解鎖時間： rLock.lock(3,TimeUnit.SECONDS); //3秒鐘 自動解鎖 這種情況下，如果你有意或無意沒有調(diào)用 unlock 進行解鎖，那么 3秒后，Redis 也會自動刪除代表這個鎖的鍵值對 當(dāng)兩個不同的線程對同一個鎖進行 lock 時，第二個線程的上鎖操作會失敗 而上鎖失敗的默認行為是阻塞等待，直到前一個線程釋放掉鎖。這種情況下，如果你不愿意等待，那么你可以調(diào)用 tryLock() 方法上鎖。tryLock 上鎖會立刻（或最多等一段時間）返回，而不會一直等（直到所得持有線程釋放）。 // 拿不到就立刻返回

rLock.tryLock();

// 拿不到最多等 1 秒。1 秒內(nèi)始終拿不到，就返回

rLock.tryLock(1, TimeUnit.SECONDS);

// 嘗試在1s內(nèi)去拿鎖，拿不到就返回false，拿到了10s自動釋放這個鎖

rLock.tryLock(1, 10, TimeUnit.SECONDS); Redisson 在上鎖時，向 Redis 中添加的鍵值對時，通過hset設(shè)置k-v的 其中鍵就是hello，當(dāng)然你也可以是其它的值，那么這個值里面的鍵是redisson內(nèi)部幫我們生成的UUID +“：” +thread-id 拼接而成的字符串；值是這個鎖的上鎖次數(shù)，默認是1 Redisson 如何保證線程間的互斥以及鎖的重入（反復(fù)上鎖）？ 因為代表著鎖的鍵值對的鍵中含有線程 ID ，因此，當(dāng)你執(zhí)行上鎖操作時，Redisson 會判斷你是否是鎖的持有者，即，當(dāng)前線程的 ID 是否和鍵值對中的線程 ID 一樣。 如果當(dāng)前執(zhí)行 lock 的線程 ID 和之前執(zhí)行 lock 成功的線程的 ID 不一致，則意味著是 “第二個人在申請鎖” ，那么就 lock 失??；如果 ID 是一樣的，那么就是 “同一個” 在反復(fù) lock，那么就累加鎖的上鎖次數(shù)，即實現(xiàn)了重入。

4、watch dog 自動延期機制

如果在使用 lock/tryLock 方法時，你指定了超時自動刪除時間，如：hello.tryLock(10, TimeUnit.SECONDS);Redis 會自動10s后將當(dāng)前線程鎖的鍵值對給刪除掉，不會自動續(xù)期，而且如果你的業(yè)務(wù)執(zhí)行時間過長，超過了key的過期時間， 而你在執(zhí)行完業(yè)務(wù)之后也去刪除這個key，就會報錯，提示錯誤為：當(dāng)前線程不能刪除這個key，因為你刪的key不是你之前的key，而是另外一個線程給redis重新設(shè)置的key。所以設(shè)置帶過期時間的hello.tryLock(10, TimeUnit.SECONDS)鍵值對時，時長一定要超過業(yè)務(wù)執(zhí)行的時長

如果，你在使用 lock/tryLock 方法時，沒有指定超時自動刪除時間，那么，就完全依靠你的手動刪除（ unlock 方法 ），那么，這種情況下你會遇到一個問題：如果你有意或無意中忘記了 unlock 釋放鎖，那么鎖背后的鍵值對將會在 Redis 中長期存在！

一定要注意Redisson 看門狗（watch dog）在指定了加鎖時間時，是不會對鎖時間自動續(xù)租的。

在 watch dog 機制中，有一個被 “隱瞞” 的細節(jié)：表面上看，你的 lock 方法沒有指定鎖定時長，但是 Redisson 去 Redis 中添加代表鎖的鍵值對時，它還是添加了自動刪除時間。默認 30 秒（可配置）。這意味著，如果，你沒有主動 unlock 進行解鎖，那么這個代表鎖的鍵值對也會在 30 秒之后被 Redis 自動刪除，但是實際上，并沒有。這正是因為 Redisson 利用 watch dog 機制對它進行了續(xù)期（ 使用 Redis 的 expire 命令重新指定新的過期時間）。也就是內(nèi)部有一個定時任務(wù)，每隔10s會會自動啟動定時任務(wù)，該任務(wù)重新給key續(xù)期30s。

柚子快報激活碼778899分享：Redisson實現(xiàn)分布式鎖

http://yzkb.51969.com/

參考閱讀