Java中ReentrantLock的用法和原理

更新時(shí)間：2023年06月07日 10:02:24 作者：IT利刃出鞘

本文主要介紹了Java中ReentrantLock的用法和原理，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

簡介

說明

本文介紹Java的JUC中的ReentrantLock（可重入獨(dú)占式鎖）。包括：用法、原理。

概述

ReentrantLock主要利用AQS隊(duì)列來實(shí)現(xiàn)。它支持公平鎖和非公平鎖。

AQS隊(duì)列使用了CAS，所以ReentrantLock有CAS的優(yōu)缺點(diǎn)。優(yōu)點(diǎn)：性能高。缺點(diǎn)：CPU占用高。

ReentrantLock的流程

state初始化為0，表示未鎖定狀態(tài)
A線程lock()時(shí)，會(huì)調(diào)用tryAcquire()獲取鎖并將state+1
其他線程tryAcquire獲取鎖會(huì)失敗，直到A線程unlock() 到state=0，其他線程才有機(jī)會(huì)獲取該鎖。
A釋放鎖之前，自己可以重復(fù)獲取此鎖（state累加），這就是可重入的概念。

注意：獲取多少次鎖就要釋放多少次鎖，保證state能回到0。

示例

private Lock lock = new ReentrantLock();
public void test(){
    lock.lock();
    try{
        doSomeThing();
    }catch (Exception e){
        // ignored
    }finally {
        lock.unlock();
    }
}

公平與非公平

ReentrantLock的默認(rèn)實(shí)現(xiàn)是非公平鎖，但是也可以設(shè)置為公平鎖。

非公平鎖

如果同時(shí)還有另一個(gè)線程進(jìn)來嘗試獲取，那么有可能會(huì)讓這個(gè)線程搶先獲??；

公平鎖

如果同時(shí)還有另一個(gè)線程進(jìn)來嘗試獲取，當(dāng)它發(fā)現(xiàn)自己不是在隊(duì)首的話，就會(huì)排到隊(duì)尾，由隊(duì)首的線程獲取到鎖。

ReentrantLock提供了兩個(gè)構(gòu)造器，分別是

public ReentrantLock() {
    sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

NonfairSync的lock()方法

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

首先用一個(gè)CAS操作，判斷state是否是0（表示當(dāng)前鎖未被占用），如果是0則把它置為1，并且設(shè)置當(dāng)前線程為該鎖的獨(dú)占線程，表示獲取鎖成功。當(dāng)多個(gè)線程同時(shí)嘗試占用同一個(gè)鎖時(shí)，CAS操作只能保證一個(gè)線程操作成功，剩下的只能去排隊(duì)啦。

“非公平”即體現(xiàn)在這里，如果占用鎖的線程剛釋放鎖，state置為0，而排隊(duì)等待鎖的線程還未喚醒時(shí)，新來的線程就直接搶占了該鎖，那么就“插隊(duì)”了。

FairSync的lock()方法

final void lock() {
    acquire(1);
}

直接調(diào)用acquire(1)方法。

非公平鎖lock()原理

場(chǎng)景

簡述：A線程獲得鎖，B和C線程失敗，B和C執(zhí)行acquire(1);

本處以非公平鎖（NonfairSync）示例進(jìn)行講解，假設(shè)有如下場(chǎng)景：有三個(gè)線程去競(jìng)爭鎖，假設(shè)線程A的CAS操作成功了，拿到了鎖開開心心的返回了，那么線程B和C則設(shè)置state失敗，走到了else里面。

NonfairSync的lock()方法

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

acquire()方法

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

（1）嘗試獲得鎖

簡述：嘗試去獲取鎖。如果嘗試獲取鎖成功，方法直接返回。

非公平鎖tryAcquire的流程是：

檢查state字段；

若為0：表示鎖未被占用，那么嘗試占用；
若不為0：檢查當(dāng)前鎖是否被自己占用，若是，則state+1（重入鎖的次數(shù)）。

若以上兩點(diǎn)都失敗，則獲取鎖失敗，返回false。

tryAcquire(arg)
final boolean nonfairTryAcquire(int acquires) {
    //獲取當(dāng)前線程
    final Thread current = Thread.currentThread();
    //獲取state變量值
    int c = getState();
    if (c == 0) { //沒有線程占用鎖
        if (compareAndSetState(0, acquires)) {
            //占用鎖成功,設(shè)置獨(dú)占線程為當(dāng)前線程
            setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) { //當(dāng)前線程已經(jīng)占用該鎖
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        // 更新state值為新的重入次數(shù)
        setState(nextc);
        return true;
    }
    //獲取鎖失敗
    return false;
}

（2）入隊(duì)

由于上文中提到線程A已經(jīng)占用了鎖，所以B和C執(zhí)行tryAcquire失敗，并且入等待隊(duì)列（acquireQueued(addWaiter(Node.EXCLUSIVE), arg))）。如果線程A拿著鎖死死不放，那么B和C就會(huì)被掛起。

先看下入隊(duì)的過程。先看addWaiter(Node.EXCLUSIVE)

/**
 * 將新節(jié)點(diǎn)和當(dāng)前線程關(guān)聯(lián)并且入隊(duì)列
 * @param mode 獨(dú)占/共享
 * @return 新節(jié)點(diǎn)
 */
private Node addWaiter(Node mode) {
    //初始化節(jié)點(diǎn),設(shè)置關(guān)聯(lián)線程和模式(獨(dú)占 or 共享)
    Node node = new Node(Thread.currentThread(), mode);
    // 獲取尾節(jié)點(diǎn)引用
    Node pred = tail;
    // 尾節(jié)點(diǎn)不為空,說明隊(duì)列已經(jīng)初始化過
    if (pred != null) {
        node.prev = pred;
        // 設(shè)置新節(jié)點(diǎn)為尾節(jié)點(diǎn)
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 尾節(jié)點(diǎn)為空,說明隊(duì)列還未初始化,需要初始化head節(jié)點(diǎn)并入隊(duì)新節(jié)點(diǎn)
    enq(node);
    return node;
}

B、C線程同時(shí)嘗試入隊(duì)列，由于隊(duì)列尚未初始化，tail==null，故至少會(huì)有一個(gè)線程會(huì)走到enq(node)。我們假設(shè)同時(shí)走到了enq(node)里。

/**
 * 初始化隊(duì)列并且入隊(duì)新節(jié)點(diǎn)
 */
private Node enq(final Node node) {
    //開始自旋
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            // 如果tail為空,則新建一個(gè)head節(jié)點(diǎn),并且tail指向head
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            // tail不為空,將新節(jié)點(diǎn)入隊(duì)
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

這里體現(xiàn)了經(jīng)典的自旋+CAS組合來實(shí)現(xiàn)非阻塞的原子操作。由于compareAndSetHead的實(shí)現(xiàn)使用了unsafe類提供的CAS操作，所以只有一個(gè)線程會(huì)創(chuàng)建head節(jié)點(diǎn)成功。假設(shè)線程B成功，之后B、C開始第二輪循環(huán)，此時(shí)tail已經(jīng)不為空，兩個(gè)線程都走到else里面。假設(shè)B線程compareAndSetTail成功，那么B就可以返回了，C由于入隊(duì)失敗還需要第三輪循環(huán)。最終所有線程都可以成功入隊(duì)。

當(dāng)B、C入等待隊(duì)列后，此時(shí)AQS隊(duì)列如下：

（3）掛起

B和C相繼執(zhí)行acquireQueued(final Node node, int arg)。這個(gè)方法讓已經(jīng)入隊(duì)的線程嘗試獲取鎖，若失敗則會(huì)被掛起。

/**
 * 已經(jīng)入隊(duì)的線程嘗試獲取鎖
 */
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é)點(diǎn)
            //如果前驅(qū)是head,即該結(jié)點(diǎn)已成老二，那么便有資格去嘗試獲取鎖
            if (p == head && tryAcquire(arg)) {
                setHead(node); // 獲取成功,將當(dāng)前節(jié)點(diǎn)設(shè)置為head節(jié)點(diǎn)
                p.next = null; // 原h(huán)ead節(jié)點(diǎn)出隊(duì),在某個(gè)時(shí)間點(diǎn)被GC回收
                failed = false; //獲取成功
                return interrupted; //返回是否被中斷過
            }
            // 判斷獲取失敗后是否可以掛起,若可以則掛起
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                // 線程若被中斷,設(shè)置interrupted為true
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

code里的注釋已經(jīng)很清晰的說明了acquireQueued的執(zhí)行流程。假設(shè)B和C在競(jìng)爭鎖的過程中A一直持有鎖，那么它們的tryAcquire操作都會(huì)失敗，因此會(huì)走到第2個(gè)if語句中。

再看下shouldParkAfterFailedAcquire和parkAndCheckInterrupt流程吧

/**
 * 判斷當(dāng)前線程獲取鎖失敗之后是否需要掛起.
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //前驅(qū)節(jié)點(diǎn)的狀態(tài)
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        // 前驅(qū)節(jié)點(diǎn)狀態(tài)為signal,返回true
        return true;
    // 前驅(qū)節(jié)點(diǎn)狀態(tài)為CANCELLED
    if (ws > 0) {
        // 從隊(duì)尾向前尋找第一個(gè)狀態(tài)不為CANCELLED的節(jié)點(diǎn)
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 將前驅(qū)節(jié)點(diǎn)的狀態(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();
}

線程入隊(duì)后能夠掛起的前提是，它的前驅(qū)節(jié)點(diǎn)的狀態(tài)為SIGNAL，它的含義是：“Hi，前面的兄弟，如果你獲取鎖并且出隊(duì)后，記得把我喚醒！”。所以shouldParkAfterFailedAcquire會(huì)先判斷當(dāng)前節(jié)點(diǎn)的前驅(qū)是否狀態(tài)符合要求，若符合則返回true，然后調(diào)用parkAndCheckInterrupt，將自己掛起；如果不符合，再看前驅(qū)節(jié)點(diǎn)是否>0(CANCELLED)，若是那么向前遍歷直到找到第一個(gè)符合要求（狀態(tài)不大于0）的前驅(qū)，若不是則將前驅(qū)節(jié)點(diǎn)的狀態(tài)設(shè)置為SIGNAL。

整個(gè)流程中，如果前驅(qū)結(jié)點(diǎn)的狀態(tài)不是SIGNAL，那么自己就不能安心掛起，需要去找個(gè)安心的掛起點(diǎn)，同時(shí)可以再嘗試下看有沒有機(jī)會(huì)去嘗試競(jìng)爭鎖。

最終隊(duì)列可能會(huì)如下圖所示

總結(jié)

用一張流程圖總結(jié)一下非公平鎖的獲取鎖的過程。

非公平鎖unlock()原理

public void unlock() {
    sync.release(1);
}
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é)點(diǎn)的狀態(tài)是否為SIGNAL，如果是則喚醒頭結(jié)點(diǎn)的下個(gè)節(jié)點(diǎn)關(guān)聯(lián)的線程，如果釋放失敗那么返回false表示解鎖失敗。這里我們也發(fā)現(xiàn)了，每次都只喚起頭結(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)關(guān)聯(lián)的線程。

最后我們?cè)倏聪聇ryRelease的執(zhí)行過程

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

這里入?yún)?。tryRelease的過程為：當(dāng)前釋放鎖的線程若不持有鎖，則拋出異常。若持有鎖，計(jì)算釋放后的state值是否為0，若為0表示鎖已經(jīng)被成功釋放，并且則清空獨(dú)占線程，最后更新state值，返回free。

公平鎖原理

公平鎖和非公平鎖不同之處在于，公平鎖在獲取鎖的時(shí)候，不會(huì)先去檢查state狀態(tài)，而是直接執(zhí)行aqcuire(1);

超時(shí)機(jī)制

在ReetrantLock的tryLock(long timeout, TimeUnit unit) 提供了超時(shí)獲取鎖的功能。它的語義是在指定的時(shí)間內(nèi)如果獲取到鎖就返回true，獲取不到則返回false。這種機(jī)制避免了線程無限期的等待鎖釋放。那么超時(shí)的功能是怎么實(shí)現(xiàn)的呢？我們還是用非公平鎖為例來一探究竟。

public boolean tryLock(long timeout, TimeUnit unit)
        throws InterruptedException {
    return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}

還是調(diào)用了內(nèi)部類里面的方法。我們繼續(xù)向前探究

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    return tryAcquire(arg) ||
        doAcquireNanos(arg, nanosTimeout);
}

這里的語義是：如果線程被中斷了，那么直接拋出InterruptedException。如果未中斷，先嘗試獲取鎖，獲取成功就直接返回，獲取失敗則進(jìn)入doAcquireNanos。tryAcquire我們已經(jīng)看過，這里重點(diǎn)看一下doAcquireNanos做了什么。

/**
 * 在有限的時(shí)間內(nèi)去競(jìng)爭鎖
 * @return 是否獲取成功
 */
private boolean doAcquireNanos(int arg, long nanosTimeout)
        throws InterruptedException {
    // 起始時(shí)間
    long lastTime = System.nanoTime();
    // 線程入隊(duì)
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        // 又是自旋!
        for (;;) {
            // 獲取前驅(qū)節(jié)點(diǎn)
            final Node p = node.predecessor();
            // 如果前驅(qū)是頭節(jié)點(diǎn)并且占用鎖成功,則將當(dāng)前節(jié)點(diǎn)變成頭結(jié)點(diǎn)
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            // 如果已經(jīng)超時(shí),返回false
            if (nanosTimeout <= 0)
                return false;
            // 超時(shí)時(shí)間未到,且需要掛起
            if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                // 阻塞當(dāng)前線程直到超時(shí)時(shí)間到期
                LockSupport.parkNanos(this, nanosTimeout);
            long now = System.nanoTime();
            // 更新nanosTimeout
            nanosTimeout -= now - lastTime;
            lastTime = now;
            if (Thread.interrupted())
                //相應(yīng)中斷
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

doAcquireNanos的流程簡述為：線程先入等待隊(duì)列，然后開始自旋，嘗試獲取鎖，獲取成功就返回，失敗則在隊(duì)列里找一個(gè)安全點(diǎn)把自己掛起直到超時(shí)時(shí)間過期。這里為什么還需要循環(huán)呢？因?yàn)楫?dāng)前線程節(jié)點(diǎn)的前驅(qū)狀態(tài)可能不是SIGNAL，那么在當(dāng)前這一輪循環(huán)中線程不會(huì)被掛起，然后更新超時(shí)時(shí)間，開始新一輪的嘗試

輪詢與中斷

ReentrantLock被保留了下來的原因是：ReentrantLock比synchronied多了兩個(gè)功能：可輪詢、可中斷。

1、可輪詢

原書上面的例子看著比較復(fù)雜，但意思很簡單。一個(gè)轉(zhuǎn)賬的操作，要么在規(guī)定的時(shí)間內(nèi)完成，要么在規(guī)定的時(shí)間內(nèi)告訴調(diào)用者，操作沒有完成。這個(gè)例子就是要了ReentrantLock的可輪詢特性，就是在規(guī)定的時(shí)間內(nèi)，反復(fù)去試圖獲得一個(gè)鎖，如果獲得成功，就能完成轉(zhuǎn)賬操作，如果在規(guī)定的時(shí)間內(nèi)，沒有獲得這個(gè)鎖，那么就是轉(zhuǎn)賬失敗。如果使用synchronized的話，肯定是無法做到的。

public boolean transferMoney(Account fromAcct,  
                             Account toAcct,  
                             DollarAmount amount,  
                             long timeout,  
                             TimeUnit unit)  
        throws InsufficientFundsException, InterruptedException {  
    long fixedDelay = getFixedDelayComponentNanos(timeout, unit);  
    long randMod = getRandomDelayModulusNanos(timeout, unit);  
    long stopTime = System.nanoTime() + unit.toNanos(timeout);  
    while (true) {  
        if (fromAcct.lock.tryLock()) {  
            try {  
                if (toAcct.lock.tryLock()) {  
                    try {  
                        if (fromAcct.getBalance().compareTo(amount) < 0)  
                            throw new InsufficientFundsException();  
                        else {  
                            fromAcct.debit(amount);  
                            toAcct.credit(amount);  
                            return true;  
                        }  
                    } finally {  
                        toAcct.lock.unlock();  
                    }  
                 }  
             } finally {  
                 fromAcct.lock.unlock();  
             }  
         }  
         if (System.nanoTime() < stopTime)  
             return false;  
         NANOSECONDS.sleep(fixedDelay + rnd.nextLong() % randMod);  
     }  
}

2、可中斷

在synchronied的代碼中，進(jìn)入臨界區(qū)的代碼是無法中斷的，這個(gè)很不靈活，如果我們使用一個(gè)線程池來分發(fā)任務(wù)，如果一個(gè)代碼長期占有鎖肯定會(huì)影響到線程池的其他任務(wù)，因此，加入中斷機(jī)制提高了對(duì)任務(wù)更強(qiáng)的控制性。

public boolean sendOnSharedLine(String message)  
        throws InterruptedException {  
    lock.lockInterruptibly();  
    try {  
        return cancellableSendOnSharedLine(message);  
    } finally {  
        lock.unlock();  
    }  
}  
private boolean cancellableSendOnSharedLine(String message)  
    throws InterruptedException { ... }

公平性：ReentrantLock默認(rèn)采用非公平鎖，synchronized鎖也是采用的非公平鎖。
如果你沒有要求鎖有可輪詢和可中斷的需求，還是使用synchronized內(nèi)置鎖吧。