分布式鎖（Redisson）-從零開始，深入理解與不斷最佳化

作者：大程子的技術成長路
連結：https://www.jianshu.com/p/bc4ff4694cf3

分布式鎖場景

案例1

如下程式碼模擬了下單減庫存的場景，我們分析下在高並行場景下會存在什麽問題

package com.wangcp.redisson;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public classIndexController{
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 模擬下單減庫存的場景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
// 從redis 中拿當前庫存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣減成功，剩余庫存：" + realStock);
}else{
System.out.println("扣減失敗，庫存不足");
}
return"end";
}
}




假設在redis中庫存（stock）初始值是100。

現在有5個客戶端同時請求該介面，可能就會存在同時執行

int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));


這行程式碼，獲取到的值都為100，緊跟著判斷大於0後都進行-1操作，最後設定到redis 中的值都為99。但正常執行完成後redis中的值應為 95。

案例2-使用synchronized 實作單機鎖

在遇到案例1的問題後，大部份人的第一反應都會想到加鎖來控制事務的原子性，如下程式碼所示：

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
synchronized (this){
// 從redis 中拿當前庫存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣減成功，剩余庫存：" + realStock);
}else{
System.out.println("扣減失敗，庫存不足");
}
}
return"end";
}


現在當有多個請求存取該介面時，同一時刻只有一個請求可進入方法體中進行庫存的扣減，其余請求等候。

但我們都知道，synchronized 鎖是屬於JVM級別的，也就是我們俗稱的「單機鎖」。但現在基本大部份公司使用的都是集群部署，現在我們思考下以上程式碼在集群部署的情況下還能保證庫存數據的一致性嗎？

答案是不能，如上圖所示，請求經Nginx分發後，可能存在多個服務同時從Redis中獲取庫存數據，此時只加synchronized （單機鎖）是無效的，並行越高，出現問題的機率就越大。

案例3-使用SETNX實作分布式鎖

setnx ：將 key 的值設為 value，若且唯若 key 不存在。

若給定 key 已經存在，則 setnx 不做任何動作。


使用setnx實作簡單的分布式鎖：

/**
 * 模擬下單減庫存的場景
 * @return
 */
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
// 使用 setnx 添加分布式鎖
// 返回 true 代表之前redis中沒有key為 lockKey 的值，並已進行成功設定
// 返回 false 代表之前redis中已經存在 lockKey 這個key了
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
if(!result){
// 代表已經加鎖了
return"error_code";
}
// 從redis 中拿當前庫存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣減成功，剩余庫存：" + realStock);
}else{
System.out.println("扣減失敗，庫存不足");
}
// 釋放鎖
stringRedisTemplate.delete(lockKey);
return"end";
}


我們知道 Redis 是單執行緒執行，現在再看 案例2 中的流程圖時，哪怕高並行場景下多個請求都執行到了setnx的程式碼，redis會根據請求的先後順序進行排列，只有排列在隊頭的請求才能設定成功。其它請求只能返回「error_code」。

當setnx設定成功後，可執行業務程式碼對庫存扣減，執行完成後對鎖進行釋放。

我們再來思考下以上程式碼已經完美實作分布式鎖了嗎？能夠支撐高並行場景嗎？答案並不是，上面的程式碼還是存在很多問題的，離真正的分布式鎖還差的很遠。我們分析下以上程式碼存在的問題：

死結：假如第一個請求在setnx加鎖完成後，執行業務程式碼時出現了異常，那釋放鎖的程式碼就無法執行，後面所有的請求也都無法進行操作了。

針對死結的問題，我們對程式碼再次進行最佳化，添加try-finally，在finally中添加釋放鎖程式碼，這樣無論如何都會執行釋放鎖程式碼，如下所示：

/**
* 模擬下單減庫存的場景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
try{
// 使用 setnx 添加分布式鎖
// 返回 true 代表之前redis中沒有key為 lockKey 的值，並已進行成功設定
// 返回 false 代表之前redis中已經存在 lockKey 這個key了
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
if(!result){
// 代表已經加鎖了
return"error_code";
}
// 從redis 中拿當前庫存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣減成功，剩余庫存：" + realStock);
}else{
System.out.println("扣減失敗，庫存不足");
}
}finally {
// 釋放鎖
stringRedisTemplate.delete(lockKey);
}
return"end";
}


經過改進後的程式碼是否還存在問題呢？我們思考正常執行的情況下應該是沒有問題，但我們假設請求在執行到業務程式碼時服務突然宕機了，或者正巧你的運維同事重新發版，粗暴的 kill -9 掉了呢，那程式碼還能執行 finally 嗎？

案例4-加入過期時間

針對想到的問題，對程式碼再次進行最佳化，加入過期時間，這樣即便出現了上述的問題，在時間到期後鎖也會自動釋放掉，不會出現「死結」的情況。

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
try{
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
if(!result){
// 代表已經加鎖了
return"error_code";
}
// 從redis 中拿當前庫存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣減成功，剩余庫存：" + realStock);
}else{
System.out.println("扣減失敗，庫存不足");
}
}finally {
// 釋放鎖
stringRedisTemplate.delete(lockKey);
}
return"end";
}


現在我們再思考一下，給鎖加入過期時間後就可以了嗎？就可以完美執行不出問題了嗎？

超時時間設定的10s真的合適嗎？如果不合適設定多少秒合適呢？如下圖所示

假設同一時間有三個請求。

請求1首先加鎖後需執行15秒，但在執行到10秒時鎖失效釋放。

請求2進入後加鎖執行，在請求2執行到5秒時，請求1執行完成進行鎖釋放，但此時釋放掉的是請求2的鎖。

請求3在請求2執行5秒時開始執行，但在執行到3秒時請求2執行完成將請求3的鎖進行釋放。

我們現在只是模擬3個請求便可看出問題，如果在真正高並行的場景下，可能鎖就會面臨「一直失效」或「永久失效」。

那麽具體問題出在哪裏呢？總結為以下幾點：

針對問題我們思考對應的解決方法：

案例5-Redisson分布式鎖

SpringBoot整合Redisson步驟

引入依賴

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


初始化客戶端

@Bean
public RedissonClient redisson(){
// 單機模式
Config config = new Config();
config.useSingleServer().setAddress("redis://192.168.3.170:6379").setDatabase(0);
return Redisson.create(config);
}


Redisson實作分布式鎖

package com.wangcp.redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public classIndexController{
@Autowired
private RedissonClient redisson;
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 模擬下單減庫存的場景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
// 1.獲取鎖物件
RLock redissonLock = redisson.getLock(lockKey);
try{
// 2.加鎖
redissonLock.lock(); // 等價於 setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
// 從redis 中拿當前庫存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣減成功，剩余庫存：" + realStock);
}else{
System.out.println("扣減失敗，庫存不足");
}
}finally {
// 3.釋放鎖
redissonLock.unlock();
}
return"end";
}
}




Redisson 分布式鎖實作原理圖

Redisson 底層源分碼析

我們點選 lock() 方法，檢視源碼，最終看到以下程式碼

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

沒錯，加鎖最終執行的就是這段 lua 手稿語言。

if (redis.call('exists', KEYS[1]) == 0) then 
redis.call('hset', KEYS[1], ARGV[2], 1); 
redis.call('pexpire', KEYS[1], ARGV[1]); 
return nil; 
end;


指令碼的主要邏輯為：

這樣來看其實和我們前面案例中的實作方法好像沒什麽區別，但實際上並不是。

這段lua指令碼命令在Redis中執行時，會被當成一條命令來執行，能夠保證原子性，故要不都成功，要不都失敗。

我們在源碼中看到Redssion的許多方法實作中很多都用到了lua指令碼，這樣能夠極大的保證命令執行的原子性。

Redisson鎖自動「續命」源碼

privatevoidscheduleExpirationRenewal(finallong threadId){
if (expirationRenewalMap.containsKey(getEntryName())) {
return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
publicvoidrun(Timeout timeout)throws Exception {
RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return 1; " +
"end; " +
"return 0;",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
future.addListener(new FutureListener<Boolean>() {
@Override
publicvoidoperationComplete(Future<Boolean> future)throws Exception {
expirationRenewalMap.remove(getEntryName());
if (!future.isSuccess()) {
log.error("Can't update lock " + getName() + " expiration", future.cause());
return;
}
if (future.getNow()) {
// reschedule itself
scheduleExpirationRenewal(threadId);
}
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
task.cancel();
}
}





這段程式碼是在加鎖後開啟一個守護執行緒進行監聽。Redisson超時時間預設設定30s，執行緒每10s呼叫一次判斷鎖還是否存在，如果存在則延長鎖的超時時間。

現在，我們再回過頭來看看案例5中的加鎖程式碼與原理圖，其實完善到這種程度已經可以滿足很多公司的使用了，並且很多公司也確實是這樣用的。但我們再思考下是否還存在問題呢？例如以下場景：

針對這些問題，我們再次思考解決方案