2023.12.10
集群模式下的并發安全問題及解決
????????隨著現在分布式系統越來越普及，一個應用往往會部署在多臺機器上（多節點），通過加鎖可以解決在單機情況下的一人一單安全問題，但是在集群模式下就不行了。見下圖：
? ? ? ? 多臺服務器會對應多個jvm，?synchronized鎖可以鎖住單臺服務器的多線程，多臺服務器就鎖不住了，所以我們需要有一個多服務器共享的鎖監視器，這里就需要使用到分布式鎖了，這里我們使用redis的SETNX這個方法來實現。? 流程圖如下：
? ? ? ? 首先定義一個鎖的接口，并實現它：
public interface ILock {/*** 嘗試獲取鎖* @param timeoutSec 鎖持有的超時時間，過期后自動釋放* @return true代表獲取鎖成功; false代表獲取鎖失敗*/boolean tryLock(long timeoutSec);/*** 釋放鎖*/void unlock();
}public class SimpleRedisLock implements ILock{private String name;private StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "lock:";@Overridepublic boolean tryLock(long timeoutSec) {//獲取線程標識long threadId = Thread.currentThread().getId();//獲取鎖Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name,threadId + "",timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(success);//防止拆箱的時候出現空指針異常}@Overridepublic void unlock() {//釋放鎖stringRedisTemplate.delete(KEY_PREFIX + name);}
}再修改業務代碼：
@Overridepublic Result seckillVoucher(Long voucherId) {//1.查詢優惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);//2.判斷秒殺活動是否開始if(voucher.getBeginTime().isAfter(LocalDateTime.now())){//尚未開始return Result.fail("秒殺尚未開始！");}//3.判斷秒殺活動是否已經結束if(voucher.getEndTime().isBefore(LocalDateTime.now())){//尚未開始return Result.fail("秒殺已經結束！");}//4.判斷庫存是否充足if(voucher.getStock() < 1){return Result.fail("庫存不足");}//此處需要將整個函數鎖起來Long userId = UserHolder.getUser().getId();//創建鎖對象SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);//獲取鎖boolean isLock = lock.tryLock(1200);//判斷是否獲取鎖成功if(!isLock){//獲取鎖失敗，不能讓黃牛不斷重復，所以直接返回失敗return Result.fail("不允許重復下單!");}//獲取鎖成功try {//獲取代理對象IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);} finally {//釋放鎖lock.unlock();}????????再使用jmeter+多臺服務器進行測試，集群模式下的并發安全問題得到解決。
redis分布式鎖誤刪問題及解決
? ? ? ? 考慮一種情況：假設線程1獲取了分布式鎖，然后業務阻塞了，阻塞的時間超過了redis中鎖的超時時間，redis將鎖釋放了。這時線程2就順利獲取了該鎖，并執行它的業務。此時線程1蘇醒了并執行完自己的業務，于是釋放鎖，此時釋放的鎖是線程2剛剛獲取的鎖，意味著此時其他線程也可以獲取鎖進來了，這就又出現了并發安全問題了。 核心原因就在于：線程1在釋放鎖之前沒有判斷一下這把鎖是不是自己之前獲取的鎖，導致誤刪了其他線程的鎖。
????????解決辦法就是：在獲取鎖的時候存入線程標識（用UUID標識，在一個JVM中，ThreadId一般不會重復，但是我們現在是集群模式，有多個JVM，多個JVM之間可能會出現ThreadId重復的情況），在釋放鎖的時候先獲取鎖的線程標識，判斷是否與當前線程標識一致：如果一致則允許釋放。
? ? ? ? 流程圖改為：
? ? ? ? 需要修改SimpleRedisLock.java代碼：
public class SimpleRedisLock implements ILock{private String name;private StringRedisTemplate stringRedisTemplate;public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "lock:";private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";@Overridepublic boolean tryLock(long timeoutSec) {//獲取線程標識String threadId = ID_PREFIX + Thread.currentThread().getId();//獲取鎖Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name,threadId,timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(success);//防止拆箱的時候出現空指針異常}@Overridepublic void unlock() {//獲取線程標示String threadId = ID_PREFIX + Thread.currentThread().getId();//獲取鎖中的標示String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);//判斷標示是否一致if(threadId.equals(id)) {//釋放鎖stringRedisTemplate.delete(KEY_PREFIX + name);}}
}分布式鎖的原子性問題及解決
? ? ? ? 然而上述解決方案在極端情況下還有問題：當線程1在判斷完鎖的標示之后，準備釋放鎖之前如果出現阻塞的話(由于jvm的垃圾回收機制等原因)，redis的超時時間到了，將鎖釋放掉，其他線程又可以獲取鎖了，則又會出現和上述一樣的情況：線程1會將其他線程的鎖誤釋放掉。 產生此問題的核心原因就在于：判斷鎖標示和釋放鎖這兩個操作不具有原子性。?導致在這期間又有可能出現并發安全問題。
? ? ? ? 這里我們使用Lua腳本解決多條命令原子性問題。Redis提供了Lua腳本功能，在一個腳本中編寫多條Redis命令，確保多條命令執行時的原子性。
? ? ? ? 編寫lua腳本：
--比較線程標示與鎖中的標示是否一致
if(redis.call('get',KEYS[1]) == ARGY[1]) then--釋放鎖return redis.call('del',KEYS[1])
end
return 0? ? ? ? 調用lua腳本：
@Overridepublic void unlock() {//調用lua腳本stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name),ID_PREFIX + Thread.currentThread().getId());}? ? ? ? 這樣判斷和釋放操作就具有原子性了。
