天天看點
傳回

值得收藏，一文掌握 Redisson 分布式鎖原理！

值得收藏，一文掌握 Redisson 分布式鎖原理！

在這裡插入圖檔描述

ReentrantLock 重入鎖

在說 Redisson 之前我們先來說一下 JDK 可重入鎖: ReentrantLock

ReentrantLock 保證了 JVM 共享資源同一時刻隻允許單個線程進行操作

實作思路

ReentrantLock 内部公平鎖與非公平鎖繼承了 AQS[AbstractQueuedSynchronizer]

1、AQS 内部通過 volatil 修飾的 int 類型變量 state 控制并發情況下線程安全問題及鎖重入

2、将未競争到鎖的線程放入 AQS 的隊列中通過 LockSupport#park、unPark 挂起喚醒

簡要描述哈, 詳情可以檢視具體的文章

Redisson

Redisson 是什麼

Redisson 是架設在 Redis 基礎上的一個 Java 駐記憶體資料網格架構, 充分利用 Redis 鍵值資料庫提供的一系列優勢, 基于 Java 實用工具包中常用接口, 為使用者提供了 一系列具有分布式特性的常用工具類

Redisson 的優勢

使得原本作為協調單機多線程并發程式的工具包 獲得了協調分布式多機多線程并發系統的能力, 大大降低了設計和研發大規模分布式系統的難度

同時結合各富特色的分布式服務, 更進一步 簡化了分布式環境中程式互相之間的協作

了解到這裡就差不多了, 就不向下擴充了, 想要了解詳細用途的, 翻一下上面的目錄

Redisson 重入鎖

由于 Redisson 太過于複雜, 設計的 API 調用大多用 Netty 相關, 是以這裡隻對 如何加鎖、如何實作重入鎖進行分析以及如何鎖續時進行分析

建立鎖

我這裡是将 Redisson 的源碼下載下傳到本地了

下面這個簡單的程式, 就是使用 Redisson 建立了一個非公平的可重入鎖

lock() 方法加鎖成功 預設過期時間 30 秒, 并且支援 “看門狗” 續時功能

public static void main(String[] args) {
    Config config = new Config();
    config.useSingleServer()
            .setPassword("123456")
            .setAddress("redis://127.0.0.1:6379");
    RedissonClient redisson = Redisson.create(config);

    RLock lock = redisson.getLock("myLock");

    try {
        lock.lock();
        // 業務邏輯
    } finally {
        lock.unlock();
    }
}

我們先來看一下 RLock 接口的聲明

public interface RLock extends Lock, RLockAsync {}

RLock 繼承了 JDK 源碼 JUC 包下的 Lock 接口, 同時也繼承了 RLockAsync

RLockAsync 從字面意思看是 支援異步的鎖, 證明擷取鎖時可以異步擷取

看了 Redisson 的源碼會知道, 注釋比黃金貴 🙃️

由于擷取鎖的 API 較多, 我們這裡以 lock() 做源碼講解, 看接口定義相當簡單

/**
 * lock 并沒有指定鎖過期時間, 預設 30 秒
 * 如果擷取到鎖, 會對鎖進行續時
 */
void lock();

擷取鎖執行個體

根據上面的小 Demo, 看下第一步擷取鎖是如何做的

RLock lock = redisson.getLock("myLock");

// name 就是鎖名稱
public RLock getLock(String name) {
    // 預設建立的同步執行器, (存在異步執行器, 因為鎖的擷取和釋放是有強一緻性要求, 預設同步)
    return new RedissonLock(connectionManager.getCommandExecutor(), name);
}

Redisson 中所有 Redis 指令都是通過 …Executor 執行的

擷取到預設的同步執行器後, 就要初始化 RedissonLock

public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
    super(commandExecutor, name);
    this.commandExecutor = commandExecutor;
    // 唯一ID
    this.id = commandExecutor.getConnectionManager().getId();
    // 等待擷取鎖時間
    this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
    // ID + 鎖名稱
    this.entryName = id + ":" + name;
    // 釋出訂閱, 後面關于加、解鎖流程會用到
    this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
}

嘗試擷取鎖

我們來看一下 RLock#lock() 底層是如何擷取鎖的

@Override
public void lock() {
    try {
        lock(-1, null, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

leaseTime: 加鎖到期時間, -1 使用預設值 30 秒

unit: 時間機關, 毫秒、秒、分鐘、小時…

interruptibly: 是否可被中斷标示

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
    // 擷取目前線程ID
    long threadId = Thread.currentThread().getId();
    // 🚩 嘗試擷取鎖, 下面重點分析
    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
    // 成功擷取鎖, 過期時間為空
    if (ttl == null) {
        return;
    }

    // 訂閱分布式鎖, 解鎖時進行通知
    RFuture<RedissonLockEntry> future = subscribe(threadId);
    if (interruptibly) {
        commandExecutor.syncSubscriptionInterrupted(future);
    } else {
        commandExecutor.syncSubscription(future);
    }

    try {
        while (true) {
            // 再次嘗試擷取鎖
            ttl = tryAcquire(-1, leaseTime, unit, threadId);
                    // 成功擷取鎖, 過期時間為空, 成功傳回
            if (ttl == null) {
                break;
            }

            // 鎖過期時間如果大于零, 則進行帶過期時間的阻塞擷取
            if (ttl >= 0) {
                try {
                    // 擷取不到鎖會在這裡進行阻塞, Semaphore, 解鎖時釋放信号量通知
                    future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                    future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                }
                // 鎖過期時間小于零, 則死等, 區分可中斷及不可中斷
            } else {
                if (interruptibly) {
                    future.getNow().getLatch().acquire();
                } else {
                    future.getNow().getLatch().acquireUninterruptibly();
                }
            }
        }
    } finally {
        // 取消訂閱
        unsubscribe(future, threadId);
    }
}

這一段代碼是用來執行加鎖, 繼續看下方法實作

Long ttl = tryAcquire(-1, leaseTime, unit, threadId);

private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}

lock() 以及 tryLock(…) 方法最終都會調用此方法, 分為兩個流程分支

1、tryLock(…) API 異步加鎖傳回

2、lock() & tryLock() API 異步加鎖并進行鎖續時

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    // 執行 tryLock(...) 才會進入
    if (leaseTime != -1) {
        // 進行異步擷取鎖
        return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // 嘗試異步擷取鎖, 擷取鎖成功傳回空, 否則傳回鎖剩餘過期時間
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
            commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
            TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    // ttlRemainingFuture 執行完成後觸發此操作
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }
        // ttlRemaining == null 代表擷取了鎖
        // 擷取到鎖後執行續時操作
        if (ttlRemaining == null) {
            scheduleExpirationRenewal(threadId);
        }
    });
    return ttlRemainingFuture;
}

繼續看一下 tryLockInnerAsync(…) 詳細的加鎖流程, 内部采用的 Lua 腳本形式, 保證了原子性操作

到這一步大家就很明了了, 将 Lua 腳本被 Redisoon 包裝最後通過 Netty 進行傳輸

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);

    return evalWriteAsync(getName(), LongCodec.INSTANCE, command,
            "if (redis.call('exists', KEYS[1]) == 0) then " +
                    "redis.call('hincrby', 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.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
evalWriteAsync(…) 中是對 Eval 指令的封裝以及 Netty 的應用就不繼續跟進了

加鎖 Lua

執行 Redis 加鎖的 Lua 腳本, 截個圖讓大家看一下參數以及具體含義

值得收藏，一文掌握 Redisson 分布式鎖原理！

image

KEYS[1]: myLock

ARGV[1]: 36000… 這個是過期時間, 自己測試的, 機關毫秒

ARGV[2]: UUID + 線程 ID

# KEYS[1] 代表上面的 myLock
# 判斷 KEYS[1] 是否存在, 存在傳回 1, 不存在傳回 0
if (redis.call('exists', KEYS[1]) == 0) then
  # 當 KEYS[1] == 0 時代表目前沒有鎖
  # 使用 hincrby 指令發現 KEYS[1] 不存在并建立一個 hash
  # ARGV[2] 就作為 hash 的第一個key, val 為 1
  # 相當于執行了 hincrby myLock 91089b45... 1
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
  # 設定 KEYS[1] 過期時間, 機關毫秒
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
end;
# 查找 KEYS[1] 中 key ARGV[2] 是否存在, 存在回傳回 1
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
  # 同上, ARGV[2] 為 key 的 val +1
    redis.call('hincrby', KEYS[1], ARGV[2], 1);
  # 同上
    redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
# 傳回 KEYS[1] 過期時間, 機關毫秒
return redis.call('pttl', KEYS[1]);

整個 Lua 腳本加鎖的流程畫圖如下:

值得收藏，一文掌握 Redisson 分布式鎖原理！

image

現在回過頭看一下擷取到鎖之後, 是如何為鎖進行延期操作的

鎖續時

之前有和軍哥聊過這個話題, 他說的思路和 Redisson 中展現的基本一緻

值得收藏，一文掌握 Redisson 分布式鎖原理！

image

說一下 Redisson 的具體實作思路吧, 中文翻譯叫做 “看門狗”

1、擷取到鎖之後執行 “看門狗” 流程

2、使用 Netty 的 Timeout 實作定時延時

3、比如鎖過期 30 秒, 每過 1/3 時間也就是 10 秒會檢查鎖是否存在, 存在則更新鎖的逾時時間

可能會有小夥伴會提出這麼一個疑問, 如果檢查傳回存在, 設定鎖過期時剛好鎖被釋放了怎麼辦?

有這樣的疑問, 代表确實用心去考慮所有可能發生的情況了, 但是不必擔心哈

Redisson 中使用的 Lua 腳本做的檢查及設定過期時間操作, 本身是原子性的不會出現上面情況

如果不想要引用 Netty 的包, 使用延時隊列等包工具也是可以完成 “看門狗”

這裡也貼一哈相關代碼, 能夠讓小夥伴更直覺的了解如何鎖續時的

值得收藏，一文掌握 Redisson 分布式鎖原理！

image

我可真是個暖男, 上代碼 RedissonLock#tryAcquireAsync(…)

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    // ...
    // 嘗試異步擷取鎖, 擷取鎖成功傳回空, 否則傳回鎖剩餘過期時間
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime,
            commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
            TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    // ttlRemainingFuture 執行完成後觸發此操作
    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
        if (e != null) {
            return;
        }
        // 擷取到鎖後執行續時操作
        if (ttlRemaining == null) {
            scheduleExpirationRenewal(threadId);
        }
    });
    return ttlRemainingFuture;
}

可以看到續時方法将 threadId 當作辨別符進行續時

private void scheduleExpirationRenewal(long threadId) {
    ExpirationEntry entry = new ExpirationEntry();
    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
    if (oldEntry != null) {
        oldEntry.addThreadId(threadId);
    } else {
        entry.addThreadId(threadId);
        renewExpiration();
    }
}

知道核心理念就好了, 沒必要研究每一行代碼哈

private void renewExpiration() {
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) {
        return;
    }

    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {
            ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
            if (ent == null) {
                return;
            }
            Long threadId = ent.getFirstThreadId();
            if (threadId == null) {
                return;
            }

            RFuture<Boolean> future = renewExpirationAsync(threadId);
            future.onComplete((res, e) -> {
                if (e != null) {
                    log.error("Can't update lock " + getName() + " expiration", e);
                    return;
                }

                if (res) {
                    // 調用本身
                    renewExpiration();
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

    ee.setTimeout(task);
}

解鎖操作

解鎖時的操作相對加鎖還是比較簡單的

@Override
public void unlock() {
    try {
        get(unlockAsync(Thread.currentThread().getId()));
    } catch (RedisException e) {
        if (e.getCause() instanceof IllegalMonitorStateException) {
            throw (IllegalMonitorStateException) e.getCause();
        } else {
            throw e;
        }
    }
}

解鎖成功後會将之前的"看門狗" Timeout 續時取消, 并傳回成功

@Override
public RFuture<Void> unlockAsync(long threadId) {
    RPromise<Void> result = new RedissonPromise<Void>();
    RFuture<Boolean> future = unlockInnerAsync(threadId);

    future.onComplete((opStatus, e) -> {
        // 取消自動續時功能
        cancelExpirationRenewal(threadId);

        if (e != null) {
            // 失敗
            result.tryFailure(e);
            return;
        }

        if (opStatus == null) {
            IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + threadId);
            result.tryFailure(cause);
            return;
        }
                // 解鎖成功
        result.trySuccess(null);
    });

    return result;
}

又是一個精髓點, 解鎖的 Lua 腳本定義

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return evalWriteAsync(getName(), 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('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                    "end; " +
                    "return nil;",
            Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}

還是來張圖了解哈, Lua 腳本會詳細分析

值得收藏，一文掌握 Redisson 分布式鎖原理！

image

解鎖 Lua

老規矩, 圖檔加參數說明

值得收藏，一文掌握 Redisson 分布式鎖原理！

image

KEYS[1]: myLock

KEYS[2]: redisson_lock_channel:{myLock}

ARGV[1]: 0

ARGV[2]: 360000… (過期時間)

ARGV[3]: 7f0c54e2…(Hash 中的鎖 Key)

# 判斷 KEYS[1] 中是否存在 ARGV[3]
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
return nil;
end;
# 将 KEYS[1] 中 ARGV[3] Val - 1
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);
# 如果傳回大于0 證明是一把重入鎖
if (counter > 0) then
  # 重制過期時間
    redis.call('pexpire', KEYS[1], ARGV[2]);
return 0;
else
  # 删除 KEYS[1]
    redis.call('del', KEYS[1]);
  # 通知阻塞等待線程或程序資源可用
    redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
return nil;

Redlock 算法

不可否認, Redisson 設計的分布式鎖真的很 NB, 但是還是沒有解決 主從節點下異步同步資料導緻鎖丢失問題

是以 Redis 作者 Antirez 推出 紅鎖算法, 這個算法的精髓就是: 沒有從節點, 如果部署多台 Redis, 各執行個體之間互相獨立, 不存在主從複制或者其他叢集協調機制

值得收藏，一文掌握 Redisson 分布式鎖原理！

image

如何使用

建立多個 Redisson Node, 由這些無關聯的 Node 組成一個完整的分布式鎖

public static void main(String[] args) {
    String lockKey = "myLock";
    Config config = new Config();
    config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6379");
    Config config2 = new Config();
    config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6380");
    Config config3 = new Config();
    config.useSingleServer().setPassword("123456").setAddress("redis://127.0.0.1:6381");

    RLock lock = Redisson.create(config).getLock(lockKey);
    RLock lock2 = Redisson.create(config2).getLock(lockKey);
    RLock lock3 = Redisson.create(config3).getLock(lockKey);

    RedissonRedLock redLock = new RedissonRedLock(lock, lock2, lock3);

    try {
        redLock.lock();
    } finally {
        redLock.unlock();
    }
}
值得收藏，一文掌握 Redisson 分布式鎖原理！

image

當然, 對于 Redlock 算法不是沒有質疑聲, 大家可以去 Redis 官網檢視Martin Kleppmann 與 Redis 作者Antirez 的辯論

CAP 原則之間的取舍

CAP 原則又稱 CAP 定理, 指的是在一個分布式系統中, Consistency(一緻性)、 Availability(可用性)、Partition tolerance(分區容錯性), 三者不可得兼

一緻性© : 在分布式系統中的所有資料備份, 在同一時刻是否同樣的值(等同于所有節點通路同一份最新的資料副本)

可用性(A): 在叢集中一部分節點故障後, 叢集整體是否還能響應用戶端的讀寫請求(對資料更新具備高可用性)

分區容忍性§: 以實際效果而言, 分區相當于對通信的時限要求. 系統如果不能在時限内達成資料一緻性, 就意味着發生了分區的情況, 必須就目前操作在 C 和 A 之間做出選擇

分布式鎖選型

如果要滿足上述分布式鎖之間的強一緻性, 可以采用 Zookeeper 的分布式鎖, 因為它底層的 ZAB協定(原子廣播協定), 天然滿足 CP

但是這也意味着性能的下降, 是以不站在具體資料下看 Redis 和 Zookeeper, 代表着性能和一緻性的取舍

如果項目沒有強依賴 ZK, 使用 Redis 就好了, 因為現在 Redis 用途很廣, 大部分項目中都引用了 Redis

沒必要對此再引入一個新的元件, 如果業務場景對于 Redis 異步方式的同步資料造成鎖丢失無法忍受, 在業務層處理就好了

作者：馬稱

原文連結：https://machen.blog.csdn.net/article/details/108819152

分布式 java Redis 多線程 緩存
上一篇: [oeasy]python0012_字元_character_chr函數_根據序号得到字元
下一篇: 圖像處理：使用Numpy和OpenCV實作傅裡葉和逆傅裡葉變換

繼續閱讀