日常開發中，基於 Redis 天然支持分散式鎖，大家線上上分散式項目中都使用過 Redis 鎖。本文主要針對日常開發中加鎖過程中某些異常場景進行講解與分析。本文講解示例代碼都在 https://github.com/wayn111/newbee-mall-pro 項目 test 目錄下 RedisL ...

日常開發中，基於 Redis 天然支持分散式鎖，大家線上上分散式項目中都使用過 Redis 鎖。本文主要針對日常開發中加鎖過程中某些異常場景進行講解與分析。本文講解示例代碼都在 https://github.com/wayn111/newbee-mall-pro 項目 test 目錄下 RedisLockTest 類中。

版本聲明：

Spring Boot 版本 3.0.2
演示項目地址：https://github.com/wayn111/newbee-mall-pro
github地址：http://github.com/wayn111 歡迎大家關註，點個star

一、任務超時，鎖已經過期

這個異常場景說實話發生概率很低，大部分情況下加鎖時任務執行都會很快，鎖還沒到期，任務自己就會刪除鎖。除非說任務調用第三方介面不穩定導致超時、資料庫查詢突然變得非常慢就可能會產生這個異常場景。

那怎麼處理這個異常嘞？大部分人可能都會回答添加一個定時任務，在定時任務內檢測鎖快過期時，進行續期操作。OK，這麼做好像是可以解決這個異常，那麼博主在這裡給出自己的見解。

1.1 先說一個暴論：如果料想到有這類異常產生，為什麼不在加鎖時，就把加鎖過期時間設置大一點

不管所續期還是增大加鎖時長，都會導致一個問題，其他線程會遲遲獲取不到鎖，一直被阻塞。那結果都一樣，為什麼不直接增大加鎖時間？

想法是好的，但是實際上，加鎖時間的設置是我們主觀臆斷的，我們無法保證這個加鎖代碼的執行時間一定在我們的鎖過期時間內。作為一個嚴謹的程式員，我們需要對我們的代碼有客觀認知，任務執行可能幾千上億萬次都是正常，但就是那麼一次它執行超時了，可能由於外部依賴、當前運行環境的異常導致。

1.2 直接不設置過期時間，任務不執行完，不釋放鎖

如果在加鎖時就不設置過期時間的話，理論上好像是可以解決這個問題，任務不執行完，鎖就不會釋放。但是作為程式員，總覺得哪裡怪怪的，任務不執行完，鎖就不會釋放！

仔細想想，我們一般在 try 中進行加鎖在 finally 進行鎖釋放，這個好像也沒毛病哦。但是實際針對一些極端異常場景下，如果任務執行過程中，伺服器宕機、程式突然被殺掉、網路斷連等都可能造成這個鎖釋放不了，另一個任務就一直獲取不到鎖。

這個方案程式正常的情況下，可以滿足我們的要求，但是一旦發生異常將導致鎖無法釋放的後果，也就是說只要我們解決這個鎖在異常場景下無法釋放的問題，這個方案還是OK的。博主這裡直接給出方案：

在不設置過期時間的加鎖操作成功時，給一個預設過期時間比如三十秒，同時啟動一個定時任務，給我們的鎖進行自動續期，每隔 預設過期時間 / 3 秒後執行一次續期操作，發生鎖剩餘時長小於 預設過期時間 / 2 就重新賦值過期時長為三十秒。這樣的話，可以保證鎖必須由任務執行完才能釋放，當程式異常發生時，仍然能保證鎖會在三十秒內釋放。

1.3 設置過期時間，任務不執行完，不釋放鎖

這個方案本質上與方案二的解決方案相同，還是啟動定時任務進行續期操作，流程這裡不做多餘講述。需要註意的就是加鎖指定過期時間會比較符合我們的客觀認知。實際上他的底層邏輯跟方案二相同，無非就是定時任務執行間隔，鎖剩餘時長續期判斷要根據過期時間來計算。

綜合來看：方案三會最合適，符合我們的客觀認知，跟我們之前對 Redis 的使用邏輯較為相近。

二、線程B加鎖執行中未釋放鎖，線程A釋放了線程B的鎖

說實話我仔細思考了一下這個異常場景，發現這個異常是個偽命題，如果線程 B 正在執行時，線程 A 怎麼能獲取到線程B的鎖！線程 A 獲取不到線程 B 的鎖，談何來去釋放線程 B 的鎖！如果線程 A 能獲取到線程 B 的鎖那麼這個分散式鎖的代碼一開始就已經錯了。

這裡回到這個異常場景本身，我們可以給每個線程設置請求ID，加鎖成功將請求ID設置為加鎖 key 的對應 value，線程釋放鎖時需要判斷當前線程的請求ID與
加鎖 key 的對應 value 是否相同，相同則可以釋放鎖，不相同則不允許釋放。

三、線程加鎖成功後繼續申請加鎖

這個場景主要發生在加鎖代碼內部調用棧過深，比如說加鎖成功執行方法 a，在方法 a 內又重覆申請了同一把鎖，導致線程把自己鎖住了，這個業界的主流叫法是叫鎖的可重入性。

解決方式有兩種，一是修改方法內的加鎖邏輯，不要加同一把鎖，修改方法 a 內的加鎖 key 名稱。二是針對加鎖邏輯做修改，實現可重入性。

這裡簡單介紹如何實現可重入性，給每個線程設置請求ID，加鎖成功將請求ID設置為加鎖 key 的對應 value，針對同一個線程的重覆加鎖，判斷當前線程已存在請求ID的情況下，請求ID直接與加鎖 key 的對應 value 相比較，相同則直接返回加鎖成功。

四、代碼實踐

4.1 加鎖自動續期實踐

設置鎖過期時間為10秒，然後該任務執行15秒，代碼如下：

ps: 以下代碼都可以在 https://github.com/wayn111/newbee-mall-pro 項目 test 目錄下 RedisLockTest 類中找到

@Slf4j
@SpringBootTest
@RunWith(SpringRunner.class)
public class RedisLockTest {
    @Autowired
    private RedisLock redisLock;
    @Test
    @Test
    public void redisLockNeNewTest() {
        String key = "test";
        try {
            log.info("---申請加鎖");
            if (redisLock.lock(key, 10)) {
                // 模擬任務執行15秒
                log.info("---加鎖成功");
                Thread.sleep(15000);
                log.info("---執行完畢");
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            redisLock.unLock(key);
        }
    }
}

執行如下：

可以看出就算任務執行超過過期時間也能通過自動續期讓代碼正常執行。

4.2 多線程下其他線程無法共同申請到同一把鎖實踐

啟動兩個線程，線程 A 先加鎖，線程 B 後枷鎖

@Test
public void redisLockReleaseSelfTest() throws IOException {
    new Thread(() -> {
        String key = "test";
        try {
            log.info("---申請加鎖");
            if (redisLock.lock(key, 10)) {
                // 模擬任務執行15秒
                log.info("---加鎖成功");
                Thread.sleep(15000);
                log.info("---執行完畢");
            } else {
                log.info("---加鎖失敗");
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            redisLock.unLock(key);
        }
    }, "thread-A").start();
    new Thread(() -> {
        String key = "test";
        try {
            Thread.sleep(100L);
            log.info("---申請加鎖");
            if (redisLock.lock(key, 10)) {
                // 模擬任務執行15秒
                log.info("---加鎖成功");
                Thread.sleep(15000);
                log.info("---執行完畢");
            } else {
                log.info("---加鎖失敗");
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            redisLock.unLock(key);
        }
    }, "thread-B").start();
    System.in.read();
}

結果如下：

可以看到，線程 A 先申請到鎖，線程 B 後申請鎖，結果線程 B 申請加鎖失敗。

4.3 鎖得可重入性實踐

當前線程加鎖成功後，線上程執行中繼續申請同一把鎖，代碼如下：

@Test
public void redisLockReEntryTest() {
    String key = "test";
    try {
        log.info("---申請加鎖");
        if (redisLock.lock(key, 10)) {
            // 模擬任務執行15秒
            log.info("---加鎖第一次成功");
            if (redisLock.lock(key, 10)) {
                // 模擬任務執行15秒
                log.info("---加鎖第二次成功");
                Thread.sleep(15000);
                log.info("---加鎖第二次執行完畢");
            } else {
                log.info("---加鎖第二次失敗");
            }
            Thread.sleep(15000);
            log.info("---加鎖第一次執行完畢");
        } else {
            log.info("---加鎖第一次失敗");
        }
    } catch (Exception e) {
        log.error(e.getMessage(), e);
    } finally {
        redisLock.unLock(key);
    }
}

結果如下：

4.4 加鎖邏輯講解

直接貼出本文最核心 RedisLock 類全部代碼：

@Slf4j
@Component
public class RedisLock {
    @Autowired
    public RedisTemplate redisTemplate;
    /**
     * 預設鎖過期時間20秒
     */
    public static final Integer DEFAULT_TIME_OUT = 30;
    /**
     * 保存線程id-ThreadLocal
     */
    private ThreadLocal<String> stringThreadLocal = new ThreadLocal<>();
    /**
     * 保存定時任務（watch-dog）-ThreadLocal
     */
    private ThreadLocal<ExecutorService> executorServiceThreadLocal = new ThreadLocal<>();
    /**
     * 加鎖，不指定過期時間
     *
     * @param key key名稱
     * @return boolean
     */
    public boolean lock(String key) {
        return lock(key, null);
    }

    /**
     * 加鎖
     *
     * @param key     key名稱
     * @param timeout 過期時間
     * @return boolean
     */
    public boolean lock(String key, Integer timeout) {
        Integer timeoutTmp;
        if (timeout == null) {
            timeoutTmp = DEFAULT_TIME_OUT;
        } else {
            timeoutTmp = timeout;
        }
        String nanoId;
        if (stringThreadLocal.get() != null) {
            nanoId = stringThreadLocal.get();
        } else {
            nanoId = IdUtil.nanoId();
            stringThreadLocal.set(nanoId);
        }
        RedisScript<Long> redisScript = new DefaultRedisScript<>(buildLuaLockScript(), Long.class);
        Long execute = (Long) redisTemplate.execute(redisScript, Collections.singletonList(key), nanoId, timeoutTmp);
        boolean flag = execute != null && execute == 1;
        if (flag) {
            ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
            executorServiceThreadLocal.set(scheduledExecutorService);
            scheduledExecutorService.scheduleWithFixedDelay(() -> {
                RedisScript<Long> renewRedisScript = new DefaultRedisScript<>(buildLuaRenewScript(), Long.class);
                Long result = (Long) redisTemplate.execute(renewRedisScript, Collections.singletonList(key), nanoId, timeoutTmp);
                if (result != null && result == 2) {
                    ThreadUtil.shutdownAndAwaitTermination(scheduledExecutorService);
                }
            }, 0, timeoutTmp / 3, TimeUnit.SECONDS);
        }
        return flag;
    }

    /**
     * 釋放鎖
     *
     * @param key key名稱
     * @return boolean
     */
    public boolean unLock(final String key) {
        String nanoId = stringThreadLocal.get();
        RedisScript<Long> redisScript = new DefaultRedisScript<>(buildLuaUnLockScript(), Long.class);
        Long execute = (Long) redisTemplate.execute(redisScript, Collections.singletonList(key), nanoId);
        boolean flag = execute != null && execute == 1;
        if (flag) {
            if (executorServiceThreadLocal.get() != null) {
                ThreadUtil.shutdownAndAwaitTermination(executorServiceThreadLocal.get());
            }
        }
        return flag;
    }

    private String buildLuaLockScript() {
        return """
                local key = KEYS[1]
                local value = ARGV[1]
                local time_out = ARGV[2]
                local result = redis.call('get', key)
                if result == value then
                    return 1;
                end
                local lock_result = redis.call('setnx', key, value)
                if tonumber(lock_result) == 1 then
                    redis.call('expire', key, time_out)
                    return 1;
                else
                    return 0;
                end
                """;
    }

    private String buildLuaUnLockScript() {
        return """
                local key = KEYS[1]
                local value = ARGV[1]
                local result = redis.call('get', key)
                if result ~= value then
                    return 0;
                else
                    redis.call('del', key)
                end
                return 1;
                """;
    }

    private String buildLuaRenewScript() {
        return """
                local key = KEYS[1]
                local value = ARGV[1]
                local timeout = ARGV[2]
                local result = redis.call('get', key)
                if result ~= value then
                    return 2;
                end
                local ttl = redis.call('ttl', key)
                if tonumber(ttl) < tonumber(timeout) / 2 then
                    redis.call('expire', key, timeout)
                    return 1;
                else
                    return 0;
                end
                """;
    }
}

加鎖邏輯：這裡我把加鎖邏輯分解成三步展示給大家

加鎖前：先判斷當前線程是否存在請求ID，不存在則生成，存在就直接使用
加鎖中：通過 lua 腳本執行原子加鎖操作，
加鎖時先判斷當前線程ID與加鎖 key 得 value 是否相等，相等則是同一個線程的鎖重入，直接返加鎖成功。不相等則設置加鎖 value 為請求ID以及過期時間。
加鎖後：啟動一個定時任務，每隔 過期時間 / 3 秒後執行一次續期操作，發現鎖剩餘時間不足 過期時間 / 2 秒後，通過 lua 腳本進行續期操作。

解鎖邏輯：這裡我把解鎖邏輯分解成兩步展示給大家

解鎖中：通過 lua 腳本執行解鎖操作，先判斷加鎖 key 的 value 是否與自身請求ID相同，相同則讓解鎖，不相同則不讓解鎖。
解鎖後：刪除定時任務。

五、總結

其實本文得核心邏輯有許多都是參考 Redission 客戶端而寫，對於這些常見得坑點，博主結合自身思考，業界知識總結並自己實現一個分散式鎖得工具類。希望大家看了有所收穫，對日常業務中 Redis 分散式鎖的使用能有更深的理解。

Redis分散式鎖常見坑點分析