四、分布式鎖

4.1 基本原理和實現(xiàn)方式對比

分布式鎖：滿足分布式系統(tǒng)或集群模式下多進(jìn)程可見并且互斥的鎖。

分布式鎖的核心思想就是讓大家都使用同一把鎖，只要大家使用的是同一把鎖，那么我們就能鎖住線程，不讓線程進(jìn)行，讓程序串行執(zhí)行，這就是分布式鎖的核心思路。

分布式鎖滿足的條件：

可見性：多個線程都能看到相同的結(jié)果。注意：這個地方說的可見性并不是并發(fā)編程中指的內(nèi)存可見性，只是說多個進(jìn)程之間都能感知到變化的意思。

互斥：互斥是分布式鎖最基本的條件，使得程序串行執(zhí)行。

高可用：程序不易崩潰，時時刻刻都保證較高的可用性。

高性能：由于加鎖本身就讓性能降低，所以對于分布式鎖本身需要它有較高的加鎖性能和釋放鎖性能。

安全性：安全是程序中必不可少的一環(huán)。

常見的三種分布式鎖：

• MySQL：mysql本身就帶有鎖機制，但是由于mysql性能本身一般，所以采用分布式鎖的情況下，其實使用mysql作為分布式鎖比較少見。
• Redis：redis作為分布式鎖是非常常見的一種使用方式，現(xiàn)在企業(yè)級開發(fā)中基本都使用redis或zookeeper作為分布式鎖，利用setnx這個方法，如果插入key成功，則表示獲得了鎖，如果有人插入成功，其他人插入失敗則表示無法獲得到鎖，利用這套邏輯來實現(xiàn)分布式鎖。
• zookeeper：zookeeper也是企業(yè)級開發(fā)中較好的一個實現(xiàn)分布式鎖的方案。

4.2 Redis分布式鎖的實現(xiàn)核心思路

實現(xiàn)分布式鎖時需要實現(xiàn)的兩個基本方法：

• 獲取鎖：

  • 互斥：確保只能有一個線程獲取鎖
  • 非阻塞：嘗試一次，成功返回true，失敗返回false

  

• 釋放鎖：

  • 手動釋放

  • 超時釋放：獲取鎖時添加一個超時時間

    

核心思路：

我們利用redis的setNx方法，當(dāng)有多個線程進(jìn)入時，我們就利用該方法，第一個線程進(jìn)入時，redis中就有這個key了，返回了1，如果結(jié)果是1表示他搶到了鎖，那么他去執(zhí)行業(yè)務(wù)，然后再刪除鎖，退出鎖邏輯，如果沒有搶到鎖，等待一定時間后重試即可

4.3 實現(xiàn)分布式鎖版本一

鎖的基本接口

public interface ILock {/*** 嘗試獲取鎖** @param timeoutSec 鎖持有的超時時間，過期后自動釋放* @return true代表獲取鎖成功；false代表獲取鎖失敗*/boolean tryLock(long timeoutSec);/*** 釋放鎖*/void unlock();
}

SimpleRedisLock

利用setnx方法進(jìn)行加鎖，同時增加過期時間，防止死鎖，此方法可以保證加鎖和增加過期時間，具有原子性

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) {//獲取線程標(biāo)識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);}
}

修改seckillVoucher業(yè)務(wù)代碼

@Autowired
private StringRedisTemplate stringRedisTemplate;public Result seckillVoucher(Long voucherId) {//1.查詢優(yōu)惠券信息SeckillVoucher voucher = seckillVoucherService.getById(voucherId);//2.判斷秒殺是否開始if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {return Result.fail("秒殺尚未開始！");}//3.判斷秒殺是否已經(jīng)結(jié)束if (voucher.getEndTime().isBefore(LocalDateTime.now())) {return Result.fail("秒殺已經(jīng)結(jié)束!");}//4.判斷庫存是否充足if (voucher.getStock() < 1) {return Result.fail("庫存不足！");}Long userId = UserHolder.getUser().getId();//創(chuàng)建鎖對象SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);//獲取鎖boolean isLock = lock.tryLock(1200);//判斷釋放獲取鎖成功if (!isLock) {//獲取鎖失敗，返回錯誤或重試return Result.fail("不允許重復(fù)下單！");}try {//獲取代理對象（事務(wù)）IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();return proxy.createVoucherOrder(voucherId);} finally {//釋放鎖lock.unlock();}
}

4.4 Redis分布式鎖誤刪情況說明

邏輯說明：

持有鎖的線程在鎖的內(nèi)部出現(xiàn)了阻塞，導(dǎo)致它的鎖超時自動釋放，線程2來嘗試獲得鎖，拿到了這把鎖，然后線程2在持有鎖執(zhí)行過程中，線程1繼續(xù)執(zhí)行，而線程1執(zhí)行過程中，走到了刪除鎖邏輯，此時就會把本應(yīng)該屬于線程2的鎖進(jìn)行刪除。

解決方案：在每個線程釋放鎖的時候，去判斷一下當(dāng)前這把鎖是否屬于自己。假設(shè)還是上面的情況，線程1卡頓，鎖超時自動釋放，線程2進(jìn)入到鎖的內(nèi)部執(zhí)行邏輯，此時線程1反映過來，然后刪除鎖，但是線程1一看當(dāng)前這把鎖不是屬于自己，于是不進(jìn)行刪除鎖邏輯，當(dāng)線程2走到刪除鎖邏輯時，如果沒有卡過自動釋放鎖的時間點，則判斷當(dāng)前這把鎖是屬于自己的，于是刪除這把鎖。

4.5 解決Redis分布式鎖誤刪問題

需求：修改之前的分布式鎖實現(xiàn)，滿足：在獲取鎖時存入線程標(biāo)識（可以用UUID表示），在釋放鎖時先獲得鎖的線程標(biāo)示，判斷是否與當(dāng)前線程標(biāo)識一致。

• 如果一致則釋放鎖
• 如果不一致則不釋放鎖

核心邏輯：在存入鎖時，放入自己線程的標(biāo)識，在刪除鎖時，判斷當(dāng)前這把鎖的標(biāo)識是不是自己存入的，如果是，則進(jìn)行刪除，如果不是，則不進(jìn)行刪除。

private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";@Override
public boolean tryLock(long timeoutSec) {//獲取線程標(biāo)識String threadId = ID_PREFIX + Thread.currentThread().getId();//獲取鎖Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);return Boolean.TRUE.equals(success);
}@Override
public void unlock() {//獲取線程標(biāo)識String threadId = ID_PREFIX + Thread.currentThread().getId();//獲取鎖中的標(biāo)識String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);//判斷標(biāo)識是否一致if (threadId.equals(id)) {//釋放鎖stringRedisTemplate.delete(KEY_PREFIX + name);}
}

有關(guān)代碼實操說明：

在我們修改完此處代碼后，我們重啟工程，然后啟動兩個線程，第一個線程持有鎖后，手動釋放鎖，第二個線程 此時進(jìn)入到鎖內(nèi)部，再放行第一個線程，此時第一個線程由于鎖的value值并非是自己，所以不能釋放鎖，也就無法刪除別人的鎖，此時第二個線程能夠正確釋放鎖，通過這個案例初步說明我們解決了鎖誤刪的問題。

4.6 分布式鎖的原子性問題

更為極端的誤刪邏輯說明：

線程1現(xiàn)在持有鎖之后，在執(zhí)行業(yè)務(wù)邏輯過程中，它正準(zhǔn)備刪除鎖，而且已經(jīng)走到了條件判斷的過程中，比如它已經(jīng)拿到了當(dāng)前這把鎖確實是屬于他自己的，正準(zhǔn)備刪除鎖，但是此時它的鎖到期了，那么此時線程2進(jìn)來，但是線程1他會接著往后執(zhí)行，當(dāng)線程1執(zhí)行到刪除鎖那行代碼時，相當(dāng)于條件判斷并沒有起到作用，這就是刪鎖時的原子性問題，之所以有這個問題，是因為線程1的拿到鎖，比較鎖，刪除鎖實際上不是一個原子性的，我們要防止剛才的情況發(fā)生。

4.7 Lua腳本解決多條命令原子性問題

Redis提供了Lua腳本功能，在一個腳本中編寫多條Redis命令，確保多條命令執(zhí)行時的原子性。

Lua是一種編程語言，它的基本語法大家可以參考網(wǎng)站：https://www.runoob.com/lua/lua-tutorial.html，這里重點介紹Redis提供的調(diào)用函數(shù)，我們可以使用Lua去操作redis，又能保證它的原子性，這樣就可以實現(xiàn)拿鎖、比較鎖和刪除鎖是一個原子性動作了。

這里重點介紹Redis提供的調(diào)用函數(shù)，語法如下：

redis.call('命令名稱','key','其他參數(shù)',...)

例如，我們要執(zhí)行set name jack，則腳本是這樣的：

# 執(zhí)行 set name jack
redis.call('set','name','jack')

例如，我們要先執(zhí)行set name Rose，再執(zhí)行g(shù)et name，則腳本如下：

# 先執(zhí)行 set name jack
redis.call('set','name','Rose')
# 再執(zhí)行 get name
local name=redis.call('get','name')
# 返回
return name

寫好腳本以后，需要用Redis命令來調(diào)用腳本，調(diào)用腳本的常見命令如下：

例如，我們要執(zhí)行 redis.call(‘set’, ‘name’, ‘jack’) 這個腳本，語法如下：

#調(diào)用腳本
EVAL "return redis.call('set','name','jack')" 0

如果腳本中的key、value不想寫死，可以作為參數(shù)傳遞。key類型參數(shù)會放入KEYS數(shù)組，其它參數(shù)會放入ARGV數(shù)組，在腳本中可以從KEYS和ARGV數(shù)組獲取這些參數(shù)：

#調(diào)用腳本
EVAL "return redis.call('set',KEYS[1],ARGV[1])" 1 name Rose

使用Lua腳本實現(xiàn)釋放鎖的流程

--這里的KEYS[1]就是鎖的key，這里的ARGV[1]就是當(dāng)前線程標(biāo)識
--獲取鎖中的標(biāo)識，判斷是否與當(dāng)前線程標(biāo)識一致
if(redis.call('GET',KEYS[1])==ARGV[1]) then-- 一致，則刪除鎖return redis.call('DEL',KEYS[1])
end
--不一致,則直接返回
return 0

4.8 利用Java代碼調(diào)用Lua腳本改造分布式鎖

在RedisTemplate中，可以利用execute方法去執(zhí)行l(wèi)ua腳本，參數(shù)對應(yīng)關(guān)系如圖所示

private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;static {UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);}public void unlock() {// 調(diào)用lua腳本stringRedisTemplate.execute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name),ID_PREFIX + Thread.currentThread().getId());
}

經(jīng)過以上改造，我們就可以實現(xiàn)拿鎖、比較鎖、刪除鎖的原子性操作了。

測試邏輯：

第一個線程進(jìn)來，得到了鎖，手動刪除鎖，模擬鎖超時了，其他線程會來搶鎖，當(dāng)?shù)谝粋€線程利用lua刪除鎖時，lua能保證他不能刪除別人的鎖，第二個線程刪除鎖時，利用lua同樣可以保證不會刪除別人的鎖，同時還能保證原子性。

4.9 總結(jié)

基于Redis的分布式鎖實現(xiàn)思路：

• 利用set nx ex 獲取鎖，并設(shè)置過期時間，保存線程標(biāo)識
• 釋放鎖時先判斷標(biāo)識是否與自己一致，一致則刪除鎖
  • 特性：
    • 利用set nx滿足互斥性
    • 利用set ex保證故障時鎖依然能釋放，避免死鎖，提高安全性
    • 利用Redis集群保證高可用和高并發(fā)特性

一路走來，利用添加過期時間，防止死鎖問題的發(fā)生，但是有了過期時間之后，可能出現(xiàn)誤刪別人鎖的問題，這個問題開始是利用刪之前拿鎖、比較鎖、刪除鎖這個邏輯來解決的，也就是刪之前判斷這把鎖是否是屬于自己的，但是現(xiàn)在還有一個原子性問題，我們無法保證拿鎖、比較鎖和刪除鎖是一個原子性動作，最后通過lua表達(dá)式解決了這個問題。