老大吩咐的可重入分佈式鎖，終於完美的實現了！！！

時間 2020-06-16

標籤老大吩咐重入分佈式終於完美實現欄目系統架構简体版

原文原文鏈接

重作永遠比改造簡單

最近在作一個項目，將一個其餘公司的實現系統（下文稱做舊系統），完整的整合到本身公司的系統（下文稱做新系統）中，這其中須要將對方實現的功能完整在本身系統也實現一遍。javascript

舊系統還有一批存量商戶，爲了避免影響存量商戶的體驗，新系統提供的對外接口，還必須得跟之前一致。最後系統完整切換以後，功能只運行在新系統中，這就要求舊系統的數據還須要完整的遷移到新系統中。html

固然這些在作這個項目以前就有預期，想過這個過程很難，可是沒想到有那麼難。本來感受排期大半年，時間仍是挺寬裕，如今感受就是大坑，還不得不在坑裏一點點去填。java

哎，說多都是淚，不吐槽了，等到下次作完再給你們覆盤下真正心得體會。redis

回到正文，上篇文章Redis 分佈式鎖，我們基於 Redis 實現一個分佈式鎖。這個分佈式鎖基本功能沒什麼問題，可是缺乏可重入的特性，因此這篇文章小黑哥就帶你們來實現一下可重入的分佈式鎖。spring

本篇文章將會涉及如下內容：安全

可重入
基於 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 腳本中，三個返回值含義以下：