實現(xiàn)分布式鎖通常有三種方式：數(shù)據(jù)庫、Redis 和 Zookeeper。我們比較常用的是通過 Redis 和 Zookeeper 實現(xiàn)分布式鎖。Redisson 框架中封裝了通過 Redis 實現(xiàn)的分布式鎖，下面我們分析一下它的具體實現(xiàn)。

by emanjusaka from https://www.emanjusaka.top/2024/03/redisson-distributed-lock 彼岸花開可奈何
本文歡迎分享與聚合，全文轉(zhuǎn)載請留下原文地址。

關(guān)鍵點

• 原子性

  要么都成功，要么都失敗

• 過期時間

  如果鎖還沒來得及釋放就遇到了服務(wù)宕機，就會出現(xiàn)死鎖的問題。給 Redis 的 key 設(shè)置過期時間，即使服務(wù)宕機了超過設(shè)置的過期時間鎖會自動進(jìn)行釋放。

• 鎖續(xù)期

  因為給鎖設(shè)置了過期時間而我們的業(yè)務(wù)邏輯具體要執(zhí)行多長時間可能是變化和不確定的，如果設(shè)定了一個固定的過期時間，可能會導(dǎo)致業(yè)務(wù)邏輯還沒有執(zhí)行完，鎖被釋放了的問題。鎖續(xù)期能保證鎖是在業(yè)務(wù)邏輯執(zhí)行完才被釋放。

• 正確釋放鎖

  保證釋放自己持有的鎖，不能出現(xiàn) A 釋放了 B 持有鎖的情況。

Redis 實現(xiàn)分布式鎖的幾種部署方式

• 單機

  在這種部署方式中，Redis 的所有實例都部署在同一臺服務(wù)器上。這種部署方式簡單易行，但存在單點故障的風(fēng)險。如果 Redis 實例宕機，則所有分布式鎖都將失效。

• 哨兵

  在這種部署方式中，Redis 的多個實例被配置為哨兵。哨兵負(fù)責(zé)監(jiān)控 Redis 實例的狀態(tài)，并在主實例宕機時自動選舉一個新的主實例。這種部署方式可以提供更高的可用性和容錯性。

• 集群

  在這種部署方式中，Redis 的多個實例被配置為一個集群。集群中的每個實例都是平等的，并且可以處理讀寫操作。這種部署方式可以提供最高的可用性和容錯性。

• 紅鎖

  搞幾個獨立的 Master，比如 5 個，然后挨個加鎖，只要超過一半以上（這里是 5/2+1=3 個）就代表加鎖成功，然后釋放鎖的時候也逐臺釋放。

使用方式

引入依賴

<!--        pom.xml文件-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.17.7</version>
</dependency>

版本依賴：

yml配置

spring:
  redis:
    redisson:
      config:
        singleServerConfig:
          address: redis://127.0.0.1:6379
          database: 0
          password: null
          timeout: 3000

直接注入使用

package top.emanjusaka;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;
/**
 * @Author emanjusaka www.emanjusaka.top
 * @Date 2024/2/28 16:41
 * @Version 1.0
 */
@Service
public class Lock {
    @Resource
    private RedissonClient redissonClient;
    public void lock() {
        // 寫入redis的key值
        String lockKey = "lock-test";
        // 獲取一個Rlock鎖對象
        RLock lock = redissonClient.getLock(lockKey);
        // 獲取鎖，并為其設(shè)置過期時間為10s
        lock.lock(10, TimeUnit.SECONDS);
        try {
            // 執(zhí)行業(yè)務(wù)邏輯....
            System.out.println("獲取鎖成功！");
        } finally {
            // 釋放鎖
            lock.unlock();
            System.out.println("釋放鎖成功！");
        }
    }
}

底層剖析

lock()

關(guān)鍵代碼

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        return commandExecutor.syncedEval(getRawName(), LongCodec.INSTANCE, command,
                "if ((redis.call('exists', KEYS[1]) == 0) " +
                       "or (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 nil; " +
                    "end; " +
                    "return redis.call('pttl', KEYS[1]);",
                Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
    }

• RFuture<T>：表示返回一個異步結(jié)果對象，其中泛型參數(shù) T 表示結(jié)果的類型。

• tryLockInnerAsync 方法接受一下參數(shù)：

  • waitTime：等待時間，用于指定在獲取鎖時的最大等待時間。
  • leaseTime：租約時間，用于指定鎖的持有時間
  • unit：時間單位，用于將 leaseTime 轉(zhuǎn)換為毫秒
  • threadId：線程 ID，用于標(biāo)識當(dāng)前線程
  • command：Redis 命令對象，用于執(zhí)行 Redis 操作

• 方法體中的代碼使用 Lua 腳本來實現(xiàn)分布式鎖的邏輯。

  • if ((redis.call('exists', KEYS[1]) == 0) or (redis.call('hexists', KEYS[1], ARGV[2]) == 1)): 如果鍵不存在或者哈希表中已經(jīng)存在對應(yīng)的線程ID，則執(zhí)行以下操作：
    • redis.call('hincrby', KEYS[1], ARGV[2], 1): 將哈希表中對應(yīng)線程ID的值加1。
    • redis.call('pexpire', KEYS[1], ARGV[1]): 設(shè)置鍵的過期時間為租約時間。
    • return nil: 返回nil表示成功獲取鎖。
  • else: 如果鍵存在且哈希表中不存在對應(yīng)的線程ID，則執(zhí)行以下操作：
    • return redis.call('pttl', KEYS[1]): 返回鍵的剩余生存時間。

• commandExecutor.syncedEval：表示同步執(zhí)行 Redis 命令

• LongCodec.INSTANCE：用于編碼和解碼長整型數(shù)據(jù)

• Collections.singletonList(getRawName())：創(chuàng)建一個只包含一個元素的列表，元素為鎖的名稱

• unit.toMillis(leaseTime)：將租約時間轉(zhuǎn)換為毫秒

• getLockName(threadId)：根據(jù)線程 ID 生成鎖的名稱

// 省去了那些無關(guān)重要的代碼
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    // tryAcquire就是上面分析的lua完整腳本
    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
    // 返回null就代表上鎖成功。
    if (ttl == null) {
        return;
    }
    // 如果沒成功，也就是鎖的剩余時間不是null的話，那么就執(zhí)行下面的邏輯
    // 其實就是說 如果有鎖（鎖剩余時間不是null），那就死循環(huán)等待重新?lián)屾i。
    try {
        while (true) {
            // 重新?lián)屾i
            ttl = tryAcquire(-1, leaseTime, unit, threadId);
            // 搶鎖成功就break退出循環(huán)
            if (ttl == null) {
                break;
            }
            // 省略一些代碼
        }
    } finally {}
}

上面代碼實現(xiàn)了一個分布式鎖的功能。它使用了Lua腳本來嘗試獲取鎖，并在成功獲取鎖后返回鎖的剩余時間（ttl）。如果獲取鎖失敗，則進(jìn)入一個死循環(huán)，不斷嘗試重新獲取鎖，直到成功為止。

unlock()

關(guān)鍵代碼

    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
              "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                        "return nil;" +
                    "end; " +
                    "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                    "if (counter > 0) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                        "return 0; " +
                    "else " +
                        "redis.call('del', KEYS[1]); " +
                        "redis.call(ARGV[4], KEYS[2], ARGV[1]); " +
                        "return 1; " +
                    "end; " +
                    "return nil;",
                Arrays.asList(getRawName(), getChannelName()),
                LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), getSubscribeService().getPublishCommand());
    }

• RFuture<Boolean>: 表示返回一個異步結(jié)果對象，其中泛型參數(shù)Boolean表示結(jié)果的類型。
• unlockInnerAsync方法接受以下參數(shù)：
  • threadId: 線程ID，用于標(biāo)識當(dāng)前線程。
• 方法體中的代碼使用Lua腳本來實現(xiàn)分布式鎖的解鎖邏輯。以下是對Lua腳本的解釋：
  • if (redis.call('hexists', KEYS[1], ARGV[3]) == 0): 如果哈希表中不存在對應(yīng)的線程ID，則返回nil表示無法解鎖。
  • local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1): 將哈希表中對應(yīng)線程ID的值減1，并將結(jié)果賦值給變量counter。
  • if (counter > 0): 如果counter大于0，表示還有其他線程持有鎖，執(zhí)行以下操作：
    • redis.call('pexpire', KEYS[1], ARGV[2]): 設(shè)置鍵的過期時間為租約時間。
    • return 0: 返回0表示鎖仍然被其他線程持有。
  • else: 如果counter等于0，表示當(dāng)前線程是最后一個持有鎖的線程，執(zhí)行以下操作：
    • redis.call('del', KEYS[1]): 刪除鍵，釋放鎖。
    • redis.call(ARGV[4], KEYS[2], ARGV[1]): 調(diào)用發(fā)布命令，通知其他線程鎖已經(jīng)釋放。
    • return 1: 返回1表示成功釋放鎖。
  • return nil: 如果前面的條件都不滿足，返回nil表示無法解鎖。
• evalWriteAsync方法用于執(zhí)行Lua腳本并返回異步結(jié)果對象。
• getRawName(): 獲取鎖的名稱。
• LongCodec.INSTANCE: 用于編碼和解碼長整型數(shù)據(jù)。
• RedisCommands.EVAL_BOOLEAN: 指定Lua腳本的返回類型為布爾值。
• Arrays.asList(getRawName(), getChannelName()): 創(chuàng)建一個包含兩個元素的列表，元素分別為鎖的名稱和頻道名稱。
• LockPubSub.UNLOCK_MESSAGE: 發(fā)布消息的內(nèi)容。
• internalLockLeaseTime: 鎖的租約時間。
• getLockName(threadId): 根據(jù)線程ID生成鎖的名稱。
• getSubscribeService().getPublishCommand(): 獲取發(fā)布命令。

鎖續(xù)期

watchDog

核心工作流程是定時監(jiān)測業(yè)務(wù)是否執(zhí)行結(jié)束，沒結(jié)束的話在看你這個鎖是不是快到期了（超過鎖的三分之一時間），那就重新續(xù)期。這樣防止如果業(yè)務(wù)代碼沒執(zhí)行完，鎖卻過期了所帶來的線程不安全問題。

Redisson 的 watchDog 機制底層不是調(diào)度線程池，而是直接用的 netty 事件輪。

Redisson的WatchDog機制是用于自動續(xù)期分布式鎖和監(jiān)控對象生命周期的一種機制，確保了分布式環(huán)境下鎖的正確性和資源的及時釋放。

• 自動續(xù)期：當(dāng)Redisson客戶端獲取了一個分布式鎖后，會啟動一個WatchDog線程。這個線程負(fù)責(zé)在鎖即將到期時自動續(xù)期，保證持有鎖的線程可以繼續(xù)執(zhí)行任務(wù)。默認(rèn)情況下，鎖的初始超時時間是30秒，每10秒鐘WatchDog會檢查一次鎖的狀態(tài)，如果鎖依然被持有，它會將鎖的過期時間重新設(shè)置為30秒。
• 參數(shù)配置：可以通過設(shè)置lockWatchdogTimeout參數(shù)來調(diào)整WatchDog檢查鎖狀態(tài)的頻率和續(xù)期的超時時間。這個參數(shù)默認(rèn)值是30000毫秒（即30秒），適用于那些沒有明確指定leaseTimeout參數(shù)的加鎖請求。
• 重連機制：除了鎖自動續(xù)期外，WatchDog機制還用作Redisson客戶端的自動重連功能。當(dāng)客戶端與Redis服務(wù)器失去連接時，WatchDog會自動嘗試重新連接，從而恢復(fù)服務(wù)的正常運作。
• 資源管理：WatchDog也負(fù)責(zé)監(jiān)控Redisson對象的生命周期，例如分布式鎖。當(dāng)對象的生命周期到期時，WatchDog會將其從Redis中刪除，避免過期數(shù)據(jù)占用過多內(nèi)存空間。
• 異步加鎖：在加鎖的過程中，WatchDog會在RedissonLock#tryAcquireAsync方法中發(fā)揮作用，該方法是進(jìn)行異步加鎖的邏輯所在。通過這種方式，加鎖操作不會阻塞當(dāng)前線程，提高了系統(tǒng)的性能。

到此這篇關(guān)于Redisson 框架中的分布式鎖的文章就介紹到這了,更多相關(guān)Redisson分布式鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論