基於redis實現分散式鎖的幾種方案與分析
1、分散式鎖的目的
分散式鎖能夠實現以下兩種功能
a、提高效率，避免重複計算。比如多節點同時執行一個批量任務。如果一個節點已經在執行某個任務，其他節點就沒必要重複執行這個任務。這時允許存在少量的重複計算，也就是說允許存在偶爾的失敗。
b、保證正確性。比如兩個客戶購買同一件商品，如果一個客戶購買了，其他客戶就不能購買。這種情況對分散式鎖的要求很高，如果重複計算，會對業務的正確性產生影響。也就是不允許失敗。
使用redis實現分散式鎖需要注意以下兩點：
a、加鎖和解鎖的實現，必須保證是同一把鎖。常見的解決方案是：給鎖設定唯一ID，加鎖時生成，解鎖時先判斷，再解鎖。
b、不能讓一個資源永久被鎖住。解決方案是給鎖設定過期時間，如果加鎖的節點宕機，在經過了過期時間之後，鎖消失，資源自動釋放。
以下提出了幾種redis分散式鎖的解決方案：
2、單一redis鎖實現
引入springboot redis的jar包：
1 <dependency> 2 <groupId>org.springframework.boot</groupId> 3 <artifactId>spring-boot-starter-data-redis</artifactId> 4 </dependency>引入spring-boot-starter-data-redis
配置檔案中加入redis相關配置：
spring:
redis:
host: 192.168.1.110
database: 2
port: 6380
timeout: 2000
password:在需要加鎖的操作前，使用
Boolean lockResult = stringRedisTemplate.opsForValue().setIfAbsent(key, value, time);
方法試圖向redis中存入一對key-value，返回true，表示拿到了鎖，就正常處理業務邏輯；返回false，表示沒拿到鎖，就處理沒有拿到資源的業務。
在finally程式碼塊中，要釋放這個鎖：
if ((driverId + "").equals(stringRedisTemplate.opsForValue().get(lock.intern()))) {
stringRedisTemplate.delete(lock.intern());
}釋放這個鎖時，要先判斷這個鎖是否是自己的鎖，以防止錯誤的釋放了別的服務設定的鎖，判斷是否是自己的鎖的依據是value，每個服務有一個特殊的value，比如：如果是滴滴司機搶一個訂單，那這個value可以是司機的id。
注意：
a、key必須能夠唯一表示某個資源；
b、value必須能夠唯一確定資源競爭者，以防止釋放鎖的時候，釋放了別人的鎖；（為什麼可能會存在釋放別人鎖的情況呢？當前服務設定了鎖之後，在鎖的過期時間之內，業務並沒有完成，導致鎖過期自動釋放，下一個服務獲取鎖之後，業務才完成，此時，可能會釋放下一個服務設定的鎖）
c、加鎖操作和給鎖設定過期時間的操作必須保證原子性，以防止加鎖成功，設定過期時間失敗，導致鎖無法釋放；
缺點：
a、單點問題。單一redis，對redis的可用性要求很高，一旦redis發生宕機，則整個服務不可用；
b、當因為某種異常情況（比如JVM的DC過程，或者網路抖動），導致業務處理時間超過鎖的過期時間，會產生業務還未執行完成，鎖就釋放的情況，如果此時有其他服務來獲取這個資源，會導致兩個服務同時擁有這個資源的情況，導致業務可能會出現問題。這個問題的解決方案可以是：自己實現一個deamon執行緒任務，當業務執行時間超過設定的鎖過期時間的三分之一的時候，判斷業務是否完成，如果還沒完成，就延長這個鎖的過期時間，延長長度設定為原始的過期時間的長度。（redisson框架就是這樣操作的。）
c、即使採用deamon執行緒的方案，也不能完全保證不出問題。如果上鎖之後，服務端與redis伺服器失聯，導致續期失敗，也會出現b的問題。
3、單一redisson鎖實現
除了引入redis的jar包，還需要引入redisson的jar包：
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.14.0</version>
</dependency>
redisson包
定義redisClient的bean：
@Bean
public RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);return Redisson.create(config);
}
在業務中先獲取redisson鎖：
String lock = "redisson_lock_orderid_3"; RLock lock1 = redissonClient.getLock(lock.intern());
在具體業務執行前，先給資源上鎖：
lock1.lock();
在finally語句塊中，釋放鎖：
lock1.unlock();
注意：
a、redisson預設的鎖過期時間是30s，也可以指定鎖的過期時間，但是方法並不是這樣：
lock1.lock(10, TimeUnit.SECONDS);
（因為用這個方法，雖然鎖的過期時間自定義成了10s，但是redisson將不會自動維護這個鎖的TTL。）
而是需要在定義RedissonClient這個Bean的時候，配置lockWatchdogTimeout這個量：
@Bean
public RedissonClient redissonClient() {
Config config = new Config();
config.useSingleServer().setAddress("redis://" + redisHost + ":" + redisPort).setDatabase(database);
config.setLockWatchdogTimeout(8000);
return Redisson.create(config);
}
這裡config.setLockWatchdogTimeout(8000);表示，將lock的過期時間expireTime設定為8s，而且watchDog會每過2s將這個key的TTL重新設定為8s(前提是這個鎖還存在的情況下)。
redisson框架會自己維護當前鎖的TTL，以防止業務執行的時間因為GC或者網路的原因異常增長，超過鎖的過期時間。原始碼在RedissonLock.class中：
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
if (leaseTime != -1L) {
return this.tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
RFuture<Long> ttlRemainingFuture = this.tryLockInnerAsync(waitTime, this.internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
if (e == null) {
if (ttlRemaining == null) {
this.scheduleExpirationRenewal(threadId);
}
}
});
return ttlRemainingFuture;
}
}
private void scheduleExpirationRenewal(long threadId) {
RedissonLock.ExpirationEntry entry = new RedissonLock.ExpirationEntry();
RedissonLock.ExpirationEntry oldEntry = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
if (oldEntry != null) {
oldEntry.addThreadId(threadId);
} else {
entry.addThreadId(threadId);
this.renewExpiration();
}
}
private void renewExpiration() {
RedissonLock.ExpirationEntry ee = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
if (ee != null) {
Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
public void run(Timeout timeout) throws Exception {
RedissonLock.ExpirationEntry ent = (RedissonLock.ExpirationEntry)RedissonLock.EXPIRATION_RENEWAL_MAP.get(RedissonLock.this.getEntryName());
if (ent != null) {
Long threadId = ent.getFirstThreadId();
if (threadId != null) {
RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
future.onComplete((res, e) -> {
if (e != null) {
RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", e);
RedissonLock.EXPIRATION_RENEWAL_MAP.remove(RedissonLock.this.getEntryName());
} else {
if (res) {
RedissonLock.this.renewExpiration();
}
}
});
}
}
}
}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);
ee.setTimeout(task);
}
}
當經過了鎖的過期時間的三分之一時，watchDog會檢查當前鎖是否還存在，如果還存在，就給當前鎖的過期時間重新設定為初始值。（預設情況下，redisson鎖的過期時間是30s，redis中，這個鎖的TTL從30開始倒數。在這個鎖的TTL為20s時，watchDog將這個鎖的TTL設定為30，繼續從30開始倒數。）
缺點：
a、單點問題。單一redis，對redis的可用性要求高，一旦redis宕機，則這個服務不可用。
b、無法完全保證不會出現“業務執行時間超過鎖TTL的時間”這個問題。假設獲取到鎖之後，如果與redis失聯，鎖的TTL無法被延長。
4、redisson紅鎖實現及其分析
引入redis和redisson的jar包，同上。
定義多個redissonClient：
@Bean("redissonClient1")
@Primary
public RedissonClient redissonClient1() {
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6388").setDatabase(0);
return Redisson.create(config);
}
@Bean("redissonClient2")
public RedissonClient redissonClient2() {
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6399").setDatabase(0);
return Redisson.create(config);
}
@Bean("redissonClient3")
public RedissonClient redissonClient3() {
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6380").setDatabase(2);
return Redisson.create(config);
}
這裡定義了三個redissonClient，注意第一個需要加上@Primary註解，如果不加，專案執行報錯。不明白為啥，不重要。
在業務程式碼中，先獲取紅鎖：
RLock lock1 = redissonClient1.getLock(lock.intern()); RLock lock2 = redissonClient2.getLock(lock.intern()); RLock lock3 = redissonClient3.getLock(lock.intern()); RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
然後執行業務前先上鎖：
redLock.lock();
業務完成，在finally程式碼塊中釋放鎖：
redLock.unlock();
紅鎖RedLock的工作流程：
a、獲取當前時間；
b、依次獲取N個節點的鎖。每個節點獲取鎖的方法和單一redisson的方法一樣，但這裡有個細節，在每個節點獲取鎖的時候，設定的過期時間都不同，需要減去之前獲取鎖操作所花費的時間：
-
例如設定鎖的過期時間是500ms；
-
第一個節點，設定鎖的過期時間是500ms，操作時間1ms；
-
第二個節點，設定鎖的過期時間是499ms，操作時間2ms；
-
第三個節點，設定鎖的過期時間是497ms······依次類推；
-
如果在某個節點，鎖的過期時間小於等於0了，說明獲取鎖的操作已經超時了，整個加鎖操作失敗。
c、判斷上鎖是否成功：如果超過N/2 + 1個節點上鎖成功，並且每個節點的鎖過期時間都大於0，就說明成功獲取到了鎖，否則，獲取鎖失敗。獲取鎖失敗時，釋放鎖。
d、釋放鎖。對所有節點發出釋放鎖的指令，每個節點釋放鎖的邏輯和上邊單一redisson的邏輯一致。為什麼不僅僅對於加鎖成功的節點發釋放鎖的指令，而是對所有節點都發？因為在某個節點上鎖失敗，不一點表示該節點上鎖失敗，有可能是因為網路延時導致操作超時，實際上鎖成功了。
上邊是紅鎖RedLock的執行流程，但是依然可能出現一些問題，尤其是在高併發情況下：
a、效能問題。分兩方面：
-
一方面，如果節點比較多，挨個加鎖，耗時可能會比較長，影響效能。解決辦法是：每個節點加鎖的操作可以是非同步操作，可以同時向多個節點獲取鎖。
-
另一方面，被加鎖的資源太大。加鎖操作本身就是為了保證正確性而犧牲了併發，犧牲和資源大小成正比，這時可以考慮對資源進行拆分。
b、重試問題。當多個client共同競爭一個資源時，每個client都獲取了部分鎖，但是沒有一個超過半數。這時候需要重試。且重試的時間要保證隨機，以便讓client重新獲取鎖的操作錯開。雖然無法根治，但是可以有效緩解這個問題。
c、節點宕機問題。對於紅鎖RedLock，如果redis節點不做持久化，某個節點宕機重啟了，可能導致多個client重複上鎖問題：比如，有A、B、C、D、E五個節點，client1從A、B、C三個節點獲取到了鎖，這時C宕機重啟，client2從C、D、E獲取到了鎖，這時就出現了兩個client同時獲取到鎖的情況。解決方案三種：
-
持久化。讓所有節點都支援持久化，但是持久化對效能影響很大，一般不採用這種方式。
-
延時啟動。讓運維配合，當redis節點宕機需要重啟時，設定延時啟動，延時的時長要大於所有鎖的TTL。
-
增加redis節點的數量。某一兩個節點宕機不至於影響鎖的歸屬。但是這樣會增加成本。這就需要在成本和服務正確性穩定性之間取一個平衡。
總結，無。。。
參考資訊：
https://blog.csdn.net/lpd_tech/article/details/104773257/
https://redis.io/topics/distlock