SpringBoot(二十七)整合Redis之分散式鎖

Redis 分散式系統 SpringBoot · 發表 2019-02-11 23:28:00

摘要：在之前的一篇文章( 《Java分散式鎖,搞懂分散式鎖實現看這篇文章就對了》 )，已經介紹過幾種java分散式鎖，今天來個Redis分散式鎖的demo。redis 現在已經成為系統快取的必備元件，針對快取讀取更新操作，通常我們希望當快取過期之後能夠只有一個請求去更新快取，其它請求依然使...

在之前的一篇文章( 《Java分散式鎖,搞懂分散式鎖實現看這篇文章就對了》 )，已經介紹過幾種java分散式鎖，今天來個Redis分散式鎖的demo。redis 現在已經成為系統快取的必備元件，針對快取讀取更新操作，通常我們希望當快取過期之後能夠只有一個請求去更新快取，其它請求依然使用舊的資料。這就需要用到鎖，因為應用服務多數以叢集方式部署，因此這裡的鎖就必需要是分散式鎖才能符合需求。

學習本章節之前，建議依次閱讀以下文章，更好的串聯全文內容，如已掌握以下列出知識點，請跳過：

v 簡單實現

鎖是針對某個資源的狀態，保證其訪問的互斥性，在實際使用當中，這個狀態一般是一個字串。使用 Redis 實現鎖，主要是將狀態放到 Redis 當中，利用其原子性，當其他執行緒訪問時，如果 Redis 中已經存在這個狀態，就不允許之後的一些操作。spring boot使用Redis的操作主要是通過RedisTemplate(或StringRedisTemplate )來實現。

1.1 將鎖狀態放入 Redis：

redisTemplate.opsForValue().setIfAbsent("lockkey", "value"); // setIfAbsent如果鍵不存在則新增,存在則不改變已經有的值。

1.2 設定鎖的過期時間

redisTemplate.expire("lockkey", 30000, TimeUnit.MILLISECONDS);

1.3 刪除/解鎖

redisTemplate.opsForValue().get("lockkey");

這麼就是簡單實現，但是1.1和1.2這麼做，這兩步違背了原子性，也就是一旦鎖被建立，而沒有設定過期時間，則鎖會一直存在。

1.4 獲取鎖

redisTemplate.delete("lockkey");

1.5 解決方案

spring data的 RedisTemplate 當中並沒有這樣的方法。但是在jedis當中是有這種原子操作的方法的，需要通過 RedisTemplate 的 execute 方法獲取到jedis裡操作命令的物件.

String result = template.execute(new RedisCallback<String>() {
@Override
public String doInRedis(RedisConnection connection) throws DataAccessException {
JedisCommands commands = (JedisCommands) connection.getNativeConnection();
return commands.set(key, "鎖定的資源", "NX", "PX", 3000);
}
});

注意： Redis 從2.6.12版本開始 set 命令支援 NX 、 PX 這些引數來達到 setnx 、 setex 、 psetex 命令的效果，文件參見： SET — Redis 命令參考

NX：表示只有當鎖定資源不存在的時候才能 SET 成功。利用 Redis 的原子性，保證了只有第一個請求的執行緒才能獲得鎖，而之後的所有執行緒在鎖定資源被釋放之前都不能獲得鎖。

v 鎖的進階

模擬一個比較常見的秒殺場景，這時候就需要用到鎖。

2.1 建立RedisLockHelper

package com.demo.common;

import com.google.common.base.Strings;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

/**
 * Created by toutou on 2019/1/27.
 */
@Component
@Slf4j
public class RedisLockHelper {
@Autowired
private StringRedisTemplate stringRedisTemplate;

/**
* 加鎖
* @param targetIdtargetId - 商品的唯一標誌
* @param timeStamp當前時間+超時時間 也就是時間戳
* @return
*/
public boolean lock(String targetId,String timeStamp){
if(stringRedisTemplate.opsForValue().setIfAbsent(targetId,timeStamp)){
// 對應setnx命令，可以成功設定,也就是key不存在
return true;
}

// 判斷鎖超時 - 防止原來的操作異常，沒有執行解鎖操作防止死鎖
String currentLock = stringRedisTemplate.opsForValue().get(targetId);
// 如果鎖過期 currentLock不為空且小於當前時間
if(!Strings.isNullOrEmpty(currentLock) && Long.parseLong(currentLock) < System.currentTimeMillis()){
// 獲取上一個鎖的時間value 對應getset，如果lock存在
String preLock =stringRedisTemplate.opsForValue().getAndSet(targetId,timeStamp);

// 假設兩個執行緒同時進來這裡，因為key被佔用了，而且鎖過期了。獲取的值currentLock=A(get取的舊的值肯定是一樣的),兩個執行緒的timeStamp都是B,key都是K.鎖時間已經過期了。
// 而這裡面的getAndSet一次只會一個執行，也就是一個執行之後，上一個的timeStamp已經變成了B。只有一個執行緒獲取的上一個值會是A，另一個執行緒拿到的值是B。
if(!Strings.isNullOrEmpty(preLock) && preLock.equals(currentLock) ){
// preLock不為空且preLock等於currentLock，也就是校驗是不是上個對應的商品時間戳，也是防止併發
return true;
}
}
return false;
}


/**
* 解鎖
* @param target
* @param timeStamp
*/
public void unlock(String target,String timeStamp){
try {
String currentValue = stringRedisTemplate.opsForValue().get(target);
if(!Strings.isNullOrEmpty(currentValue) && currentValue.equals(timeStamp) ){
// 刪除鎖狀態
stringRedisTemplate.opsForValue().getOperations().delete(target);
}
} catch (Exception e) {
log.error("警報！警報！警報！解鎖異常{}",e);
}
}
}

這個是Redis加鎖和解鎖的工具類，裡面使用的主要是兩個命令，SETNX和GETSET。

SETNX命令將key設定值為value，如果key不存在，這種情況下等同SET命令。當key存在時，什麼也不做

GETSET命令先查詢出原來的值，值不存在就返回nil。然後再設定值對應的Java方法在程式碼中提示了。注意一點的是，Redis是單執行緒的！所以在執行GETSET和SETNX不會存在併發的情況。

2.2 建立Controller模擬秒殺場景

package com.demo.controller;

import com.demo.common.RedisLockHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * Created by toutou on 2019/1/27.
 */
@RestController
@Slf4j
public class RedisController {

@Autowired
RedisLockHelper redisLockHelper;

/**
* 超時時間 5s
*/
private static final int TIMEOUT = 5*1000;

@RequestMapping(value = "/seckilling")
public String Seckilling(String targetId){
//加鎖
long time = System.currentTimeMillis() + TIMEOUT;
if(!redisLockHelper.lock(targetId,String.valueOf(time))){
return "排隊人數太多，請稍後再試.";
}

int surplusCount = 0;
// 查詢該商品庫存，為0則活動結束 e.g. getStockByTargetId
if(surplusCount==0){
return "活動結束.";
}else {
// 下單 e.g. buyStockByTargetId

//減庫存 不做處理的話，高併發下會出現超賣的情況，下單數，大於減庫存的情況。雖然這裡減了，但由於併發，減的庫存還沒存到map中去。新的併發拿到的是原來的庫存
surplusCount =surplusCount-1;
try{
Thread.sleep(100);//模擬減庫存的處理時間
}catch (InterruptedException e){
e.printStackTrace();
}
// 減庫存操作資料庫 e.g. updateStockByTargetId

// buyStockByTargetId 和 updateStockByTargetId 可以同步完成(或者事物)，保證原子性。
}

//解鎖
redisLockHelper.unlock(targetId,String.valueOf(time));

return "恭喜您，秒殺成功。";
}
}

其他參考資料：

注：本文中很多內容來自以上鍊接的學習心得，感謝以上人員分享，也請轉載本文的各站保持以上鍊接。

v 原始碼地址

https://github.com/toutouge/javademo/tree/master/hellospringboot

作者： 請叫我頭頭哥

出處： http://www.cnblogs.com/toutou/

關於作者：專注於基礎平臺的專案開發。如有問題或建議，請多多賜教！

版權宣告：本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連結。

特此宣告：所有評論和私信都會在第一時間回覆。也歡迎園子的大大們指正錯誤，共同進步。或者直接私信我

聲援博主：如果您覺得文章對您有幫助，可以點選文章右下角一下。您的鼓勵是作者堅持原創和持續寫作的最大動力！

SpringBoot(二十七)整合Redis之分散式鎖

v 簡單實現

v 鎖的進階

v 原始碼地址

您可能也會喜歡…