Redisson 分布式鎖實現之源碼篇 → 為什么推薦用 Redisson 客戶端

本文轉載自查看原文 2021-07-05 09:03 618 Redisson源碼/ Redis/ Redisson

開心一刻

　　一男人站在樓頂准備跳樓，樓下有個勸解員拿個喇叭准備勸解

　　勸解員：兄弟，別跳

　　跳樓人：我不想活了

　　勸解員：你想想你媳婦

　　跳樓人：媳婦跟人跑了

　　勸解員：你還有兄弟

　　跳樓人：就是跟我兄弟跑的

　　勸解員：你想想你家孩子

　　跳樓人：孩子是他倆的

　　勸解員：死吧，媽的你活着也沒啥價值了

前言

　　關於鎖，相信大家都不陌生，一般我們用其在多線程環境中控制對共享資源的並發訪問

　　單服務下，用 JDK 中的 synchronized 或 Lock 的實現類可實現對共享資源的並發訪問

　　分布式服務下，JDK 中的鎖就顯得力不從心了，分布式鎖也就應運而生了

　　分布式鎖的實現方式有很多，常見的有如下幾種

　　　　基於 MySQL，利用行級悲觀鎖（select ... for update）

　　　　基於 Redis，利用其（setnx + expire）或 set

　　　　基於 Zookeeper，利用其臨時目錄和事件回調機制

　　具體的實現細節就不展開了，網上資料很多

　　看下文之前最好先看下：Redisson 分布式鎖實現之前置篇 → Redis 的發布/訂閱與 Lua，方便更好的理解下文

分布式鎖的特點

　　可以類比 JDK 中的鎖

　　互斥

　　　　不僅要保證同個服務中不同線程的互斥，還需要保證不同服務間、不同線程的互斥

　　　　如何處理互斥，是自旋、還是阻塞，還是其他？

　　超時

　　　　鎖超時設置，防止程序異常奔潰而導致鎖一直存在，后續同把鎖一直加不上

　　續期

　　　　程序具體執行的時長無法確定，所以過期時間只能是個估值，那么就不能保證程序在過期時間內百分百能運行完

　　　　所以需要進行鎖續期，保證業務能夠正常執行完

　　可重入

　　　　可重入鎖又名遞歸鎖，是指同一個線程在外層方法已經獲得鎖，再進入該線程的中層或內層方法會自動獲取鎖

　　　　簡單點來說，就是同個線程可以反復獲取同一把鎖

　　專一釋放

　　　　通俗點來講：誰加的鎖就只有它能釋放這把鎖

　　　　為什么會出現這種錯亂釋放的問題了，舉個例子就理解了

　　　　　　線程 T1 對資源 lock_zhangsan 加了鎖，由於某些原因，加鎖業務還未執行完，鎖過期自動釋放了，此時線程 T2 對資源 lock_zhangsan 加鎖成功

　　　　　　T2 執行業務的時候，T1 業務執行完后釋放資源 lock_zhangsan 的鎖，結果把 T2 加的鎖給釋放了

　　公平與非公平

　　　　公平鎖：多個線程按照申請鎖的順序去獲得鎖，所有線程都在隊列里排隊，這樣就保證了隊列中的第一個先得到鎖

　　　　非公平鎖：多個線程不按照申請鎖的順序去獲得鎖，而是同時直接去嘗試獲取鎖

　　　　JDK 中的 ReentrantLock 就有公平和非公平兩種實現，有興趣的可以去看看它的源碼

　　　　多數情況下用的是非公平鎖，但有些特殊情況下需要用公平鎖

　　很多小伙伴覺得：引入一個簡單的分布式鎖，有必要考慮這么多嗎？

　　雖然絕大部分情況下，我們的程序都是在跑正常流程，但不能保證異常情況 100% 跑不到，出於健壯性考慮，異常情況都需要考慮到

　　下面我們就來看看 Redisson 是如何實現這些特點的

Redisson 實現分布式鎖

　　關於 Redisson，更多詳細信息可查看官方文檔

　　Redisson 是 Redis 官方推薦的 Java 版的 Redis 客戶端，它提供了非常豐富的功能，其中就包括本文關注的分布式鎖

　　環境准備

　　　　簡單示例開始之前，我們先看下環境；版本不同，會有一些差別

　　　　JDK：1.8

　　　　Redis：3.2.8

　　　　Redisson：3.13.6

　　簡單示例

　　　　先將 Redis 信息配置給 Redisson，創建出 RedissonClient

　　　　Redis 的部署方式不同，Redisson 配置模式也會不同，詳細信息可查看：Configuration

　　　　我們就配置最簡單的 Single instance mode

　　　　RedissonClient 創建出來后，就可以通過它來獲取鎖

　　　　完整示例代碼：redisson-demo

　　接下來我們從源碼層面一起看看 Redisson 具體是如何實現分布式鎖的特點的

客戶端創建

　　客服端的創建過程中，會生成一個 id 作為唯一標識，用以區分分布式下不同節點中的客戶端

　　id 值就是一個 UUID，客戶端啟動時生成

　　那么這個 id 有什么用，大家暫且在腦中留下這個疑問，我們接着往下看

鎖的獲取

　　我們從 lock 開始跟源碼

　　最終會來到有三個參數的 lock 方法

    private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        long threadId = Thread.currentThread().getId();
        
        // 嘗試獲取鎖；ttl為null表示鎖獲取成功; ttl不為null表示獲取鎖失敗，其值為其他線程占用該鎖的剩余時間
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return;
        }

        // 鎖被其他線程占用而獲取失敗,使用redis的發布訂閱功能來等待鎖的釋放通知，而非自旋監測鎖的釋放
        RFuture<RedissonLockEntry> future = subscribe(threadId);
        
        // 當前線程會阻塞，直到鎖被釋放時當前線程被喚醒（有超時等待，默認 7.5s，而不會一直等待）
        // 持有鎖的線程釋放鎖之后，redis會發布消息，所有等待該鎖的線程都會被喚醒，包括當前線程
        if (interruptibly) {
            commandExecutor.syncSubscriptionInterrupted(future);
        } else {
            commandExecutor.syncSubscription(future);
        }

        try {
            while (true) {
                // 嘗試獲取鎖；ttl為null表示鎖獲取成功; ttl不為null表示獲取鎖失敗，其值為其他線程占用該鎖的剩余時間
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message
                if (ttl >= 0) {
                    try {
                        // future.getNow().getLatch() 返回的是 Semaphore 對象，其初始許可證為 0，以此來控制線程獲取鎖的順序
                        // 通過 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);
        }
//        get(lockAsync(leaseTime, unit));
    }