關(guān)于ReentrantLock的實現(xiàn)原理解讀

更新時間：2023年06月19日 10:08:30 作者：盛夏溫暖流年

這篇文章主要介紹了關(guān)于ReentrantLock的實現(xiàn)原理，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

ReentrantLock 簡介

ReentrantLock 實現(xiàn)了 Lock 接口，是一種可重入的獨占鎖。

相比于 synchronized 同步鎖，ReentrantLock 更加靈活，擁有更加強大的功能，比如可以實現(xiàn)公平鎖機制。

首先，先來了解一下什么是公平鎖機制。

ReentrantLock 的公平鎖機制

我們知道，ReentrantLock 分為公平鎖和非公平鎖，可以通過構(gòu)造方法來指定具體類型：

//默認(rèn)非公平鎖
public ReentrantLock() {
    sync = new NonfairSync();
}
//公平鎖
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

公平鎖

在多個線程競爭獲取鎖時，公平鎖傾向于將訪問權(quán)授予等待時間最長的線程。

也就是說，公平鎖相當(dāng)于有一個線程等待隊列，先進(jìn)入隊列的線程會先獲得鎖，按照 "FIFO（先進(jìn)先出）" 的原則，對于每一個等待線程都是公平的。

非公平鎖

非公平鎖是搶占模式，線程不會關(guān)注隊列中是否存在其他線程，也不會遵守先來后到的原則，直接嘗試獲取鎖。

接下來進(jìn)入正題，一起分析下 ReentrantLock 的底層是如何實現(xiàn)的。

ReentrantLock 的底層實現(xiàn)

ReentrantLock 實現(xiàn)的前提是 AbstractQueuedSynchronizer（抽象隊列同步器），簡稱 AQS，是 java.util.concurrent 的核心，

常用的線程并發(fā)類 CountDownLatch、CyclicBarrier、Semaphore、ReentrantLock 等都包括了一個繼承自 AQS 抽象類的內(nèi)部類。

同步標(biāo)志位 state

AQS 內(nèi)部維護(hù)了一個同步標(biāo)志位 state，用來實現(xiàn)同步加鎖控制：

private volatile int state;

同步標(biāo)志位 state 的初始值為 0，線程每加一次鎖，state 就會加 1，也就是說，已經(jīng)獲得鎖的線程再次加鎖，state 值會再次加 1。

可以看出，state 實際上表示的是已獲得鎖的線程進(jìn)行加鎖操作的次數(shù)。

CLH 隊列

除了 state 同步標(biāo)志位外，AQS 內(nèi)部還使用一個 FIFO 的隊列（也叫 CLH 隊列）來表示排隊等待鎖的線程，當(dāng)線程爭搶鎖失敗后會封裝成 Node 節(jié)點加入 CLH 隊列中去。

Node 的代碼實現(xiàn)：

static final class Node {
      // 標(biāo)識當(dāng)前節(jié)點在共享模式
      static final Node SHARED = new Node();
      // 標(biāo)識當(dāng)前節(jié)點在獨占模式
      static final Node EXCLUSIVE = null;
      static final int CANCELLED =  1;
      static final int SIGNAL    = -1;
      static final int CONDITION = -2;
      static final int PROPAGATE = -3;
　　　 volatile int waitStatus;
      //前驅(qū)節(jié)點
      volatile Node prev;
      //后驅(qū)節(jié)點
      volatile Node next;
      //當(dāng)前線程
      volatile Thread thread;
      //存儲在condition隊列中的后繼節(jié)點
      Node nextWaiter;
      //是否為共享鎖
      final boolean isShared() {
            return nextWaiter == SHARED;
      }
      final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
      }
      Node() {}
      Node(Thread thread, Node mode) {
            this.nextWaiter = mode;
            this.thread = thread;
      }
      Node(Thread thread, int waitStatus) {
            this.waitStatus = waitStatus;
            this.thread = thread;
      }
}

分析代碼可知，每個 Node 節(jié)點都有兩個指針，分別指向直接后繼節(jié)點和直接前驅(qū)節(jié)點。

Node 節(jié)點的變化過程

當(dāng)出現(xiàn)鎖競爭以及釋放鎖的時候，AQS 同步隊列中的 Node 節(jié)點會發(fā)生變化，如下圖所示：

線程封裝成 Node 節(jié)點追加到隊列末尾，設(shè)置當(dāng)前節(jié)點的 prev 節(jié)點和 next 節(jié)點的指向；通過 CAS 將 tail 重新指向新的尾部節(jié)點，即當(dāng)前插入的 Node 節(jié)點；

head 節(jié)點表示獲取鎖成功的節(jié)點，當(dāng)頭結(jié)點釋放鎖后，會喚醒后繼節(jié)點，如果后繼節(jié)點獲得鎖成功，就會把自己設(shè)置為頭結(jié)點，節(jié)點的變化過程如下：

修改 head 節(jié)點指向下一個獲得鎖的節(jié)點；新的獲得鎖的節(jié)點，將 prev 的指針指向 null；

和設(shè)置 tail 的重新指向不同，設(shè)置 head 節(jié)點不需要用 CAS，是因為設(shè)置 head 節(jié)點是由獲得鎖的線程來完成的，而同步鎖只能由一個線程獲得，所以不需要 CAS 保證。

只需要把 head 節(jié)點設(shè)置為原首節(jié)點的后繼節(jié)點，并且斷開原 head 節(jié)點的 next 引用即可。

除了前驅(qū)和后繼節(jié)點，Node 類中還包括了 SHARED 和 EXCLUSIVE 節(jié)點，它們起到了什么作用呢？這就不得不介紹一下 AQS 的兩種資源共享模式了。

AQS 的資源共享模式

AQS 通過 EXCLUSIVE 和 SHARED 兩個變量來定義獨占模式或共享模式。

獨占模式

獨占模式是最常用的模式，使用范圍很廣，比如 ReentrantLock 的加鎖和釋放鎖就是使用獨占模式實現(xiàn)的。

獨占模式中的核心加鎖方法是 acquire()：

    public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

這里首先調(diào)用 tryAcquire() 方法嘗試獲取鎖，也就是嘗試通過 CAS 修改 state 為 1，如果發(fā)現(xiàn)鎖已經(jīng)被當(dāng)前線程占用，就執(zhí)行重入，也就是給 state+1；

如果鎖被其他線程占有，那么當(dāng)前線程執(zhí)行 tryAcquire 返回失敗，則會執(zhí)行 addWaiter() 方法在等待隊列中添加一個獨占式節(jié)點，addWaiter() 方法實現(xiàn)如下：

    private Node addWaiter(Node mode) {
        //創(chuàng)建一個節(jié)點，此處mode是獨占式的
        Node node = new Node(mode);
        for (;;) {
            Node oldTail = tail;
            if (oldTail != null) {
                // 如果tail節(jié)點非空，就將新節(jié)點的前節(jié)點設(shè)置為tail節(jié)點，并將tail指向新節(jié)點
                node.setPrevRelaxed(oldTail);
                //CAS將tail更新為新節(jié)點
                if (compareAndSetTail(oldTail, node)) {
                    //把原tail的next設(shè)為當(dāng)前節(jié)點
                    oldTail.next = node;
                    return node;
                }
            } else {
                //還沒有初始化，就調(diào)用initializeSyncQueue()方法初始化
                initializeSyncQueue();
            }
        }
    }

寫入隊列后，需要掛起當(dāng)前線程，代碼如下：

/**
 * 已經(jīng)入隊的線程嘗試獲取鎖
 */
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true; //標(biāo)記是否成功獲取鎖
    try {
        boolean interrupted = false; //標(biāo)記線程是否被中斷過
        for (;;) {
            final Node p = node.predecessor(); //獲取前驅(qū)節(jié)點
            //如果前驅(qū)是head,即該結(jié)點是第二位，有資格去嘗試獲取鎖
            if (p == head && tryAcquire(arg)) {
                setHead(node); // 獲取成功,將當(dāng)前節(jié)點設(shè)置為head節(jié)點
                p.next = null; // 原h(huán)ead節(jié)點出隊
                failed = false; //獲取成功
                return interrupted; //返回是否被中斷過
            }
            // 判斷獲取失敗后是否可以掛起,若可以則掛起
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                // 線程若被中斷,設(shè)置interrupted為true
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

再看下 shouldParkAfterFailedAcquire 和 parkAndCheckInterrupt 都做了哪些事：

/**
 * 判斷當(dāng)前線程獲取鎖失敗之后是否需要掛起.
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //前驅(qū)節(jié)點的狀態(tài)
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        // 前驅(qū)節(jié)點狀態(tài)為signal,返回true
        return true;
    // 前驅(qū)節(jié)點狀態(tài)為CANCELLED
    if (ws > 0) {
        // 從隊尾向前尋找第一個狀態(tài)不為CANCELLED的節(jié)點
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 將前驅(qū)節(jié)點的狀態(tài)設(shè)置為SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
/**
 * 掛起當(dāng)前線程,返回線程中斷狀態(tài)并重置
 */
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

通過以上代碼可以看出，線程入隊后能夠掛起的前提是，它的前驅(qū)節(jié)點的狀態(tài)為 SIGNAL，這意味著當(dāng)前一個節(jié)點獲取鎖并且出隊后，需要把后面的節(jié)點進(jìn)行喚醒。

加鎖說完了再說解鎖，解鎖的方法相比來說更加簡單，核心方法是 release()：

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

代碼流程：先嘗試釋放鎖，若釋放成功，那么查看頭結(jié)點的狀態(tài)是否為 SIGNAL，如果是，則喚醒頭結(jié)點的下個節(jié)點關(guān)聯(lián)的線程，如果釋放失敗就返回 false 表示解鎖失敗。

其中的 tryRelease() 方法實現(xiàn)如下，詳細(xì)流程見注釋說明：

/**
 * 釋放當(dāng)前線程占用的鎖
 * @param releases
 * @return 是否釋放成功
 */
protected final boolean tryRelease(int releases) {
    // 計算釋放后state值
    int c = getState() - releases;
    // 如果不是當(dāng)前線程占用鎖,那么拋出異常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        // 鎖被重入次數(shù)為0,表示釋放成功
        free = true;
        // 清空獨占線程
        setExclusiveOwnerThread(null);
    }
    // 更新state值
    setState(c);
    return free;
}