老大吩咐的可重入分布式鎖，終于完美的實作了！！！-有解無憂

重做永遠比改造簡單

最近在做一個專案，將一個其他公司的實作系統（下文稱作舊系統），完整的整合到自己公司的系統（下文稱作新系統）中，這其中需要將對方實作的功能完整在自己系統也實作一遍，

舊系統還有一批存量商戶，為了不影響存量商戶的體驗，新系統提供的對外介面，還必須得跟以前一致，最后系統完整切換之后，功能只運行在新系統中，這就要求舊系統的資料還需要完整的遷移到新系統中，

當然這些在做這個專案之前就有預期，想過這個程序很難，但是沒想到有那么難，原本感覺排期大半年，時間還是挺寬裕，現在感覺就是大坑，還不得不在坑里一點點去填，

哎，說多都是淚，不吐槽了，等到下次做完再給大家復盤下真正心得體會，

回到正文，上篇文章Redis 分布式鎖，咱們基于 Redis 實作一個分布式鎖，這個分布式鎖基本功能沒什么問題，但是缺少可重入的特性，所以這篇文章小黑哥就帶大家來實作一下可重入的分布式鎖，

本篇文章將會涉及以下內容：

可重入
基于 ThreadLocal 實作方案
基于 Redis Hash 實作方案

先贊后看，養成習慣，微信搜索「程式通事」，關注就完事了~

可重入

說到可重入鎖，首先我們來看看一段來自 wiki 上可重入的解釋：

若一個程式或子程式可以“在任意時刻被中斷然后作業系統調度執行另外一段代碼，這段代碼又呼叫了該子程式不會出錯”，則稱其為可重入（reentrant或re-entrant）的，即當該子程式正在運行時，執行執行緒可以再次進入并執行它，仍然獲得符合設計時預期的結果，與多執行緒并發執行的執行緒安全不同，可重入強調對單個執行緒執行時重新進入同一個子程式仍然是安全的，

當一個執行緒執行一段代碼成功獲取鎖之后，繼續執行時，又遇到加鎖的代碼，可重入性就就保證執行緒能繼續執行，而不可重入就是需要等待鎖釋放之后，再次獲取鎖成功，才能繼續往下執行，

用一段 Java 代碼解釋可重入：

public synchronized void a() {
    b();
}

public synchronized void b() {
    // pass
}

假設 X 執行緒在 a 方法獲取鎖之后，繼續執行 b 方法，如果此時不可重入，執行緒就必須等待鎖釋放，再次爭搶鎖，

鎖明明是被 X 執行緒擁有，卻還需要等待自己釋放鎖，然后再去搶鎖，這看起來就很奇怪，我釋放我自己~

我打我自己

可重入性就可以解決這個尷尬的問題，當執行緒擁有鎖之后，往后再遇到加鎖方法，直接將加鎖次數加 1，然后再執行方法邏輯，退出加鎖方法之后，加鎖次數再減 1，當加鎖次數為 0 時，鎖才被真正的釋放，

可以看到可重入鎖最大特性就是計數，計算加鎖的次數，所以當可重入鎖需要在分布式環境實作時，我們也就需要統計加鎖次數，

分布式可重入鎖實作方式有兩種：

基于 ThreadLocal 實作方案
基于 Redis Hash 實作方案

首先我們看下基于 ThreadLocal 實作方案，

基于 ThreadLocal 實作方案

實作方式

Java 中 ThreadLocal可以使每個執行緒擁有自己的實體副本，我們可以利用這個特性對執行緒重入次數進行技術，

下面我們定義一個ThreadLocal的全域變數 LOCKS，記憶體存盤 Map 實體變數，

private static ThreadLocal<Map<String, Integer>> LOCKS = ThreadLocal.withInitial(HashMap::new);

每個執行緒都可以通過 ThreadLocal獲取自己的 Map實體，Map 中 key 存盤鎖的名稱，而 value存盤鎖的重入次數，

加鎖的代碼如下：

/**
 * 可重入鎖
 *
 * @param lockName  鎖名字,代表需要爭臨界資源
 * @param request   唯一標識，可以使用 uuid，根據該值判斷是否可以重入
 * @param leaseTime 鎖釋放時間
 * @param unit      鎖釋放時間單位
 * @return
 */
public Boolean tryLock(String lockName, String request, long leaseTime, TimeUnit unit) {
    Map<String, Integer> counts = LOCKS.get();
    if (counts.containsKey(lockName)) {
        counts.put(lockName, counts.get(lockName) + 1);
        return true;
    } else {
        if (redisLock.tryLock(lockName, request, leaseTime, unit)) {
            counts.put(lockName, 1);
            return true;
        }
    }
    return false;
}

ps: redisLock#tryLock 為上一篇文章實作的分布鎖，

由于公號外鏈無法直接跳轉，關注『程式通事』，回復分布式鎖獲取源代碼，

加鎖方法首先判斷當前執行緒是否已經已經擁有該鎖，若已經擁有，直接對鎖的重入次數加 1，

若還沒擁有該鎖，則嘗試去 Redis 加鎖，加鎖成功之后，再對重入次數加 1 ，

釋放鎖的代碼如下：

/**
 * 解鎖需要判斷不同執行緒池
 *
 * @param lockName
 * @param request
 */
public void unlock(String lockName, String request) {
    Map<String, Integer> counts = LOCKS.get();
    if (counts.getOrDefault(lockName, 0) <= 1) {
        counts.remove(lockName);
        Boolean result = redisLock.unlock(lockName, request);
        if (!result) {
            throw new IllegalMonitorStateException("attempt to unlock lock, not locked by lockName:+" + lockName + " with request: "
                    + request);
        }

    } else {
        counts.put(lockName, counts.get(lockName) - 1);
    }
}

釋放鎖的時首先判斷重入次數，若大于 1，則代表該鎖是被該執行緒擁有，所以直接將鎖重入次數減 1 即可，

若當前可重入次數小于等于 1，首先移除 Map中鎖對應的 key，然后再到 Redis 釋放鎖，

這里需要注意的是，當鎖未被該執行緒擁有，直接解鎖，可重入次數也是小于等于 1 ，這次可能無法直接解鎖成功，

ThreadLocal 使用程序要記得及時清理內部存盤實體變數，防止發生記憶體泄漏，背景關系資料串用等問題，

下次咱來聊聊最近使用 ThreadLocal 寫的 Bug，

基于 Redis Hash 可重入鎖

實作方式

ThreadLocal 的方案中我們使用了 Map 記載鎖的可重入次數，而 Redis 也同樣提供了 Hash （哈希表）這種可以存盤鍵值對資料結構，所以我們可以使用 Redis Hash 存盤的鎖的重入次數，然后利用 lua 腳本判斷邏輯，

加鎖的 lua 腳本如下：

---- 1 代表 true
---- 0 代表 false

if (redis.call('exists', KEYS[1]) == 0) then
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return 1;
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 1;
end ;
return 0;

如果 KEYS:[lock],ARGV[1000,uuid]

不熟悉 lua 語言同學也不要怕，上述邏輯還是比較簡單的，

加鎖代碼首先使用 Redis exists 命令判斷當前 lock 這個鎖是否存在，

如果鎖不存在的話，直接使用 hincrby創建一個鍵為 lock hash 表，并且為 Hash 表中鍵為 uuid 初始化為 0，然后再次加 1，最后再設定過期時間，

如果當前鎖存在，則使用 hexists判斷當前 lock 對應的 hash 表中是否存在 uuid 這個鍵，如果存在,再次使用 hincrby 加 1，最后再次設定過期時間，

最后如果上述兩個邏輯都不符合，直接回傳，

加鎖代碼如下:

// 初始化代碼

String lockLuaScript = IOUtils.toString(ResourceUtils.getURL("classpath:lock.lua").openStream(), Charsets.UTF_8);
lockScript = new DefaultRedisScript<>(lockLuaScript, Boolean.class);

/**
 * 可重入鎖
 *
 * @param lockName  鎖名字,代表需要爭臨界資源
 * @param request   唯一標識，可以使用 uuid，根據該值判斷是否可以重入
 * @param leaseTime 鎖釋放時間
 * @param unit      鎖釋放時間單位
 * @return
 */
public Boolean tryLock(String lockName, String request, long leaseTime, TimeUnit unit) {
    long internalLockLeaseTime = unit.toMillis(leaseTime);
    return stringRedisTemplate.execute(lockScript, Lists.newArrayList(lockName), String.valueOf(internalLockLeaseTime), request);
}

Spring-Boot 2.2.7.RELEASE

只要搞懂 Lua 腳本加鎖邏輯，Java 代碼實作還是挺簡單的，直接使用 SpringBoot 提供的 StringRedisTemplate 即可，

解鎖的 Lua 腳本如下：

-- 判斷 hash set 可重入 key 的值是否等于 0
-- 如果為 0 代表 該可重入 key 不存在
if (redis.call('hexists', KEYS[1], ARGV[1]) == 0) then
    return nil;
end ;
-- 計算當前可重入次數
local counter = redis.call('hincrby', KEYS[1], ARGV[1], -1);
-- 小于等于 0 代表可以解鎖
if (counter > 0) then
    return 0;
else
    redis.call('del', KEYS[1]);
    return 1;
end ;
return nil;

首先使用 hexists 判斷 Redis Hash 表是否存給定的域，

如果 lock 對應 Hash 表不存在，或者 Hash 表不存在 uuid 這個 key，直接回傳 nil，

若存在的情況下，代表當前鎖被其持有，首先使用 hincrby使可重入次數減 1 ，然后判斷計算之后可重入次數，若小于等于 0，則使用 del 洗掉這把鎖，

解鎖的 Java 代碼如下：

// 初始化代碼：


String unlockLuaScript = IOUtils.toString(ResourceUtils.getURL("classpath:unlock.lua").openStream(), Charsets.UTF_8);
unlockScript = new DefaultRedisScript<>(unlockLuaScript, Long.class);

/**
 * 解鎖
 * 若可重入 key 次數大于 1，將可重入 key 次數減 1 <br>
 * 解鎖 lua 腳本回傳含義：<br>
 * 1:代表解鎖成功 <br>
 * 0:代表鎖未釋放，可重入次數減 1 <br>
 * nil：代表其他執行緒嘗試解鎖 <br>
 * <p>
 * 如果使用 DefaultRedisScript<Boolean>，由于 Spring-data-redis eval 型別轉化，<br>
 * 當 Redis 回傳  Nil bulk, 默認將會轉化為 false，將會影響解鎖語意，所以下述使用：<br>
 * DefaultRedisScript<Long>
 * <p>
 * 具體轉化代碼請查看：<br>
 * JedisScriptReturnConverter<br>
 *
 * @param lockName 鎖名稱
 * @param request  唯一標識，可以使用 uuid
 * @throws IllegalMonitorStateException 解鎖之前，請先加鎖，若為加鎖，解鎖將會拋出該錯誤
 */
public void unlock(String lockName, String request) {
    Long result = stringRedisTemplate.execute(unlockScript, Lists.newArrayList(lockName), request);
    // 如果未回傳值，代表其他執行緒嘗試解鎖
    if (result == null) {
        throw new IllegalMonitorStateException("attempt to unlock lock, not locked by lockName:+" + lockName + " with request: "
                + request);
    }
}

解鎖代碼執行方式與加鎖類似，只不過解鎖的執行結果回傳型別使用 Long，這里之所以沒有跟加鎖一樣使用 Boolean ,這是因為解鎖 lua 腳本中，三個回傳值含義如下：