1. 程式人生 > >Redisson實現分散式鎖—RedissonLock

Redisson實現分散式鎖—RedissonLock

Redisson實現分散式鎖—RedissonLock

有關Redisson實現分散式鎖上一篇部落格講了分散式的鎖原理:Redisson實現分散式鎖---原理

這篇主要講RedissonLock和RLock。Redisson分散式鎖的實現是基於RLock介面,RedissonLock實現RLock介面。

一、RLock介面

1、概念

public interface RLock extends Lock, RExpirable, RLockAsync

很明顯RLock是繼承Lock鎖,所以他有Lock鎖的所有特性,比如lock、unlock、trylock等特性,同時它還有很多新特性:強制鎖釋放,帶有效期的鎖,。

2、RLock鎖API

這裡針對上面做個整理,這裡列舉幾個常用的介面說明

public interface RRLock {
    //----------------------Lock介面方法-----------------------

    /**
     * 加鎖 鎖的有效期預設30秒
     */
    void lock();
    /**
     * tryLock()方法是有返回值的,它表示用來嘗試獲取鎖,如果獲取成功,則返回true,如果獲取失敗(即鎖已被其他執行緒獲取),則返回false .
     */
    boolean tryLock();
    /**
     * tryLock(long time, TimeUnit unit)方法和tryLock()方法是類似的,只不過區別在於這個方法在拿不到鎖時會等待一定的時間,
     * 在時間期限之內如果還拿不到鎖,就返回false。如果如果一開始拿到鎖或者在等待期間內拿到了鎖,則返回true。
     *
     * @param time 等待時間
     * @param unit 時間單位 小時、分、秒、毫秒等
     */
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    /**
     * 解鎖
     */
    void unlock();
    /**
     * 中斷鎖 表示該鎖可以被中斷 假如A和B同時調這個方法,A獲取鎖,B為獲取鎖,那麼B執行緒可以通過
     * Thread.currentThread().interrupt(); 方法真正中斷該執行緒
     */
    void lockInterruptibly();

    //----------------------RLock介面方法-----------------------
    /**
     * 加鎖 上面是預設30秒這裡可以手動設定鎖的有效時間
     *
     * @param leaseTime 鎖有效時間
     * @param unit      時間單位 小時、分、秒、毫秒等
     */
    void lock(long leaseTime, TimeUnit unit);
    /**
     * 這裡比上面多一個引數,多新增一個鎖的有效時間
     *
     * @param waitTime  等待時間
     * @param leaseTime 鎖有效時間
     * @param unit      時間單位 小時、分、秒、毫秒等
     */
    boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;
    /**
     * 檢驗該鎖是否被執行緒使用,如果被使用返回True
     */
    boolean isLocked();
    /**
     * 檢查當前執行緒是否獲得此鎖(這個和上面的區別就是該方法可以判斷是否當前執行緒獲得此鎖,而不是此鎖是否被執行緒佔有)
     * 這個比上面那個實用
     */
    boolean isHeldByCurrentThread();
    /**
     * 中斷鎖 和上面中斷鎖差不多,只是這裡如果獲得鎖成功,新增鎖的有效時間
     * @param leaseTime  鎖有效時間
     * @param unit       時間單位 小時、分、秒、毫秒等
     */
    void lockInterruptibly(long leaseTime, TimeUnit unit);  
}

RLock相關介面,主要是新添加了 leaseTime 屬性欄位,主要是用來設定鎖的過期時間,避免死鎖。


二、RedissonLock實現類

public class RedissonLock extends RedissonExpirable implements RLock

RedissonLock實現了RLock介面,所以實現了介面的具體方法。這裡我列舉幾個方法說明下

1、void lock()方法

    @Override
    public void lock() {
        try {
            lockInterruptibly();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

發現lock鎖裡面進去其實用的是lockInterruptibly(中斷鎖,表示可以被中斷),而且捕獲異常後用 Thread.currentThread().interrupt()來真正中斷當前執行緒,其實它們是搭配一起使用的。

具體有關lockInterruptibly()方法講解推薦一個部落格。部落格:Lock的lockInterruptibly()

接下來執行流程,這裡理下關鍵幾步

   /**
     * 1、帶上預設值調另一箇中斷鎖方法
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {
        lockInterruptibly(-1, null);
    }
    /**
     * 2、另一箇中斷鎖的方法
     */
    void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException 
    /**
     * 3、這裡已經設定了鎖的有效時間預設為30秒  (commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()=30)
     */
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    /**
     * 4、最後通過lua指令碼訪問Redis,保證操作的原子性
     */
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "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 nil; " +
                        "end; " +
                        "return redis.call('pttl', KEYS[1]);",
                Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

那麼void lock(long leaseTime, TimeUnit unit)方法其實和上面很相似了,就是從上面第二步開始的。

2、tryLock(long waitTime, long leaseTime, TimeUnit unit)

介面的引數和含義上面已經說過了,現在我們開看下原始碼,這裡只顯示一些重要邏輯。

 @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        //1、 獲取鎖同時獲取成功的情況下,和lock(...)方法是一樣的 直接返回True,獲取鎖False再往下走
        if (ttl == null) {
            return true;
        }
        //2、如果超過了嘗試獲取鎖的等待時間,當然返回false 了。
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }

        // 3、訂閱監聽redis訊息,並且建立RedissonLockEntry,其中RedissonLockEntry中比較關鍵的是一個 Semaphore屬性物件,用來控制本地的鎖請求的訊號量同步,返回的是netty框架的Future實現。
        final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        //  阻塞等待subscribe的future的結果物件,如果subscribe方法呼叫超過了time,說明已經超過了客戶端設定的最大wait time,則直接返回false,取消訂閱,不再繼續申請鎖了。
        //  只有await返回true,才進入迴圈嘗試獲取鎖
        if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
                    @Override
                    public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                        if (subscribeFuture.isSuccess()) {
                            unsubscribe(subscribeFuture, threadId);
                        }
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }

       //4、如果沒有超過嘗試獲取鎖的等待時間,那麼通過While一直獲取鎖。最終只會有兩種結果
        //1)、在等待時間內獲取鎖成功 返回true。2)等待時間結束了還沒有獲取到鎖那麼返回false。
        while (true) {
            long currentTime = System.currentTimeMillis();
            ttl = tryAcquire(leaseTime, unit, threadId);
            // 獲取鎖成功
            if (ttl == null) {
                return true;
            }
           //   獲取鎖失敗
            time -= System.currentTimeMillis() - currentTime;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }
        }
    }

重點 tryLock一般用於特定滿足需求的場合,但不建議作為一般需求的分散式鎖,一般分散式鎖建議用void lock(long leaseTime, TimeUnit unit)。因為從效能上考慮,在高併發情況下後者效率是前者的好幾倍

3、unlock()

解鎖的邏輯很簡單。

@Override
    public void unlock() {
        // 1.通過 Lua 指令碼執行 Redis 命令釋放鎖
        Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE,
                RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end;" +
                        "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('publish', KEYS[2], ARGV[1]); " +
                        "return 1; "+
                        "end; " +
                        "return nil;",
                Arrays.<Object>asList(getName(), getChannelName()),
                LockPubSub.unlockMessage, internalLockLeaseTime,
                getLockName(Thread.currentThread().getId()));
        // 2.非鎖的持有者釋放鎖時丟擲異常
        if (opStatus == null) {
            throw new IllegalMonitorStateException(
                    "attempt to unlock lock, not locked by current thread by node id: "
                            + id + " thread-id: " + Thread.currentThread().getId());
        }
        // 3.釋放鎖後取消重新整理鎖失效時間的排程任務
        if (opStatus) {
            cancelExpirationRenewal();
        }
    }

使用 EVAL 命令執行 Lua 指令碼來釋放鎖:

  1. key 不存在,說明鎖已釋放,直接執行 publish 命令釋出釋放鎖訊息並返回 1
  2. key 存在,但是 field 在 Hash 中不存在,說明自己不是鎖持有者,無權釋放鎖,返回 nil
  3. 因為鎖可重入,所以釋放鎖時不能把所有已獲取的鎖全都釋放掉,一次只能釋放一把鎖,因此執行 hincrby 對鎖的值減一。
  4. 釋放一把鎖後,如果還有剩餘的鎖,則重新整理鎖的失效時間並返回 0;如果剛才釋放的已經是最後一把鎖,則執行 del 命令刪除鎖的 key,併發布鎖釋放訊息,返回 1

注意這裡有個實際開發過程中,容易出現很容易出現上面第二步異常,非鎖的持有者釋放鎖時丟擲異常。比如下面這種情況

      //設定鎖1秒過去
        redissonLock.lock("redisson", 1);
        /**
         * 業務邏輯需要諮詢2秒
         */
        redissonLock.release("redisson");
      /**
       * 執行緒1 進來獲得鎖後,執行緒一切正常並沒有宕機,但它的業務邏輯需要執行2秒,這就會有個問題,在 執行緒1 執行1秒後,這個鎖就自動過期了,
       * 那麼這個時候 執行緒2 進來了。線上程1去解鎖就會拋上面這個異常(因為解鎖和當前鎖已經不是同一執行緒了)
       */




只要自己變優秀了,其他的事情才會跟著好起來(中將6)