從 AbstractQueuedSynchronizer 理解 ReentrantLock

LeoHsiun 發(fā)布于2019-08-19 10:51 / 3153人閱讀

摘要：當(dāng)前線程已經(jīng)獲取過這個(gè)鎖，則此時(shí)是重入，改變的計(jì)數(shù)即可，返回表示加鎖成功。的核心在于使用更新鎖的狀態(tài)，并利用一個(gè)同步隊(duì)列將獲取鎖失敗的線程進(jìn)行排隊(duì)，當(dāng)前驅(qū)節(jié)點(diǎn)解鎖后再喚醒后繼節(jié)點(diǎn)，是一個(gè)幾乎純實(shí)現(xiàn)的加鎖與解鎖。

簡介

Java 并發(fā)編程離不開鎖， Synchronized 是常用的一種實(shí)現(xiàn)加鎖的方式，使用比較簡單快捷。在 Java 中還有另一種鎖，即 Lock 鎖。 Lock 是一個(gè)接口，提供了超時(shí)阻塞、可響應(yīng)中斷以及公平非公平鎖等特性，相比于 Synchronized，Lock 功能更強(qiáng)大，可以實(shí)現(xiàn)更靈活的加鎖方式。

Lock 的主要實(shí)現(xiàn)類是 ReentrantLock，而 ReetrantLock 中具體的實(shí)現(xiàn)方式是利用另外一個(gè)類 AbstractQueuedSynchronizer，所有的操作都是委托給這個(gè)類完成。AbstractQueuedSynchronizer 是 Lock 鎖的重要組件，本文從 AbstractQueuedSynchronizer 來分析 ReetrantLock 的實(shí)現(xiàn)原理。

基本用法

先看一下 Lock 的基本用法：

Lock lock = ...;
lock.lock();
try{
    //處理任務(wù)
}catch(Exception ex){
     
}finally{
    lock.unlock();   //釋放鎖
}

lock.lock() 即是加鎖， lock.unolck() 是釋放鎖，為了保證所能夠釋放，unlock() 應(yīng)該放到 finally 中。

下面分別從 lock() 和 unlock() 方法來分析加鎖和解鎖到底做了什么。

lock

下面是 lock() 的代碼：

    public void lock() {
        sync.lock();
    }

可以看到，只是簡單調(diào)用了 sync 對應(yīng)的 lock() 方法。那么這個(gè) sync 是什么呢？其實(shí)這個(gè)就是 AbstractQueuedSynchronizer 的實(shí)現(xiàn)類。可以看一下 ReentrantLock 的構(gòu)造方法：

    /**
     * Creates an instance of {@code ReentrantLock}.
     * This is equivalent to using {@code ReentrantLock(false)}.
     */
    public ReentrantLock() {
        sync = new NonfairSync();
    }

    /**
     * Creates an instance of {@code ReentrantLock} with the
     * given fairness policy.
     *
     * @param fair {@code true} if this lock should use a fair ordering policy
     */
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

ReentrantLock 有兩個(gè)方法，主要目的是選擇是公平鎖還是非公平鎖。公平鎖指的是先來后到，先爭搶鎖的線程先獲得鎖，而非公平鎖則不一定。ReentrantLock 默認(rèn)使用的是非公平鎖，也可以通過構(gòu)造參數(shù)選擇公平鎖。選擇哪個(gè)鎖其實(shí)是生成了一個(gè)對象并賦值給變量 sync，下面是涉及到的代碼：

/**
     * Base of synchronization control for this lock. Subclassed
     * into fair and nonfair versions below. Uses AQS state to
     * represent the number of holds on the lock.
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;

        /**
         * Performs {@link Lock#lock}. The main reason for subclassing
         * is to allow fast path for nonfair version.
         */
        abstract void lock();

        /**
         * Performs non-fair tryLock.  tryAcquire is implemented in
         * subclasses, but both need nonfair try for trylock method.
         */
        final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }

        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }

        protected final boolean isHeldExclusively() {
            // While we must in general read state before owner,
            // we don"t need to do so to check if current thread is owner
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

        final ConditionObject newCondition() {
            return new ConditionObject();
        }

        // Methods relayed from outer class

        final Thread getOwner() {
            return getState() == 0 ? null : getExclusiveOwnerThread();
        }

        final int getHoldCount() {
            return isHeldExclusively() ? getState() : 0;
        }

        final boolean isLocked() {
            return getState() != 0;
        }

        /**
         * Reconstitutes the instance from a stream (that is, deserializes it).
         */
        private void readObject(java.io.ObjectInputStream s)
            throws java.io.IOException, ClassNotFoundException {
            s.defaultReadObject();
            setState(0); // reset to unlocked state
        }
    }
    // 非公平鎖
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

    /** 
     * Sync object for fair locks
     *  公平鎖
     */
    static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

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

        /**
         * Fair version of tryAcquire.  Don"t grant access unless
         * recursive call or no waiters or is first.
         */
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() &&
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

在 ReentrantLock 中有一個(gè)抽象的內(nèi)部類 Sync，繼承于 AbstractQueuedSynchronizer 并實(shí)現(xiàn)了一些方法。另有兩個(gè)類 FairSync 和 NoFairSync 繼承了 Sync，它們自然就是公平鎖以及非公平鎖的實(shí)現(xiàn)。下面分析將從公平鎖出發(fā)，非公平鎖與公平鎖差別并不是很多。

公平鎖 FairSync 加鎖的代碼如下：

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

只有一行，調(diào)用了 acquire，這是 AbstractQueuedSynchronizer 中的一個(gè)方法，代碼如下：

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

acquire 的實(shí)現(xiàn)也很短，不多在其中卻包含了加鎖的具體實(shí)現(xiàn)，關(guān)鍵就在內(nèi)部調(diào)用的幾個(gè)方法中。

為了理解加鎖和解鎖的過程，下面具體介紹一下 AbstractQueuedSynchronizer（以下簡稱 AQS）。

AbstractQueuedSynchronizer

AQS 中使用一個(gè)同步隊(duì)列來實(shí)現(xiàn)線程同步狀態(tài)的管理，當(dāng)一個(gè)線程獲取鎖失敗的時(shí)候， AQS將此線程構(gòu)造成一個(gè)節(jié)點(diǎn)(Node)并加入同步隊(duì)列并且阻塞線程。當(dāng)鎖釋放時(shí)，會(huì)從同步隊(duì)列中將第一個(gè)節(jié)點(diǎn)喚醒并使其再次獲取鎖。

同步隊(duì)列中的節(jié)點(diǎn)用來保存獲取鎖失敗的線程的相關(guān)信息，包含如下屬性：

static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
        // 標(biāo)識(shí)共享模式
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        // 標(biāo)識(shí)獨(dú)占模式
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled. */
        // 線程取消
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor"s thread needs unparking. */
        // 需要喚醒后繼節(jié)點(diǎn)
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition. */
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate.
         */
        static final int PROPAGATE = -3;
        // 節(jié)點(diǎn)狀態(tài)，為上面的幾個(gè)狀態(tài)之一
        volatile int waitStatus;
        // 前置節(jié)點(diǎn)
        volatile Node prev;
        // 后繼節(jié)點(diǎn)
        volatile Node next;
        // 節(jié)點(diǎn)所表示的線程
        volatile Thread thread;
        
        Node nextWaiter;
        ...
}

Node 是 AQS 的內(nèi)部類，其中包含一些屬性標(biāo)識(shí)一個(gè)阻塞線程的節(jié)點(diǎn)，包括是獨(dú)占模式還是共享模式、節(jié)點(diǎn)的狀態(tài)、前驅(qū)結(jié)點(diǎn)、后繼結(jié)點(diǎn)以及節(jié)點(diǎn)所代表的線程。

同步隊(duì)列是一個(gè)雙向列表，在 AQS 中有這樣幾個(gè)屬性：

     /**
     * Head of the wait queue, lazily initialized.  Except for
     * initialization, it is modified only via method setHead.  Note:
     * If head exists, its waitStatus is guaranteed not to be
     * CANCELLED.
     */
     // 頭結(jié)點(diǎn)
    private transient volatile Node head;

    /**
     * Tail of the wait queue, lazily initialized.  Modified only via
     * method enq to add new wait node.
     */
     // 尾節(jié)點(diǎn)
    private transient volatile Node tail;

    /**
     * The synchronization state.
     */
     // 鎖的狀態(tài)
    private volatile int state;

其中，head 和 tail 分別指向同步隊(duì)列的頭結(jié)點(diǎn)和尾節(jié)點(diǎn)，state 標(biāo)識(shí)鎖當(dāng)前的狀態(tài)，為 0 時(shí)表示當(dāng)前鎖未被占用，大于 1 表示被占用，之所以是大于 1 是因?yàn)殒i可以重入，每重入一次增加 1。同步隊(duì)列的結(jié)構(gòu)大致如下圖：

了解了同步隊(duì)列后，下面具體看看加鎖和解鎖的過程。

加鎖

    final void lock() {
        acquire(1);
    }
    public final void acquire(int arg) {
        // 加鎖的主要代碼
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

主要邏輯其實(shí)就是一行代碼：if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，tryAcquire 是嘗試獲取一下鎖，為什么說是嘗試呢？看代碼：

protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();  // 獲取當(dāng)前狀態(tài)
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                // 成功獲取到鎖
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            // 重入
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
}

tryAcquire 可以分為三條分支：

當(dāng)前鎖未被占用（getState() == 0），則判斷是否有前驅(qū)結(jié)點(diǎn)，沒有的話就用 CAS 加鎖(compareAndSetState(0, acquires))，加鎖成功則調(diào)用 setExclusiveOwnerThread(current) 標(biāo)示一下并返回 true。

當(dāng)前線程已經(jīng)獲取過這個(gè)鎖，則此時(shí)是重入，改變 state 的計(jì)數(shù)即可，返回 true 表示加鎖成功。

如果不是上面兩種情況，那么說明鎖被占用或者 CAS 沒有搶過其它線程，則需要進(jìn)入同步隊(duì)列，返回 false 表示嘗試加鎖失敗。

回到 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 這一行，如果 tryAcquire(arg) 返回 false 將會(huì)執(zhí)行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)。先看一下 addWaiter(Node.EXCLUSIVE) ，這個(gè)方法的代碼如下所示：

private Node addWaiter(Node mode) {
        // 將線程包裝成 Node 
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        // 尾節(jié)點(diǎn)為空
        if (pred != null) { 
            node.prev = pred; 
            if (compareAndSetTail(pred, node)) { 
                pred.next = node;
                return node;
            }
        }
        
        enq(node);
        return node;
}

其中主要邏輯是將當(dāng)前線程包裝為一個(gè) Node 節(jié)點(diǎn)并加入同步隊(duì)列。如果尾節(jié)點(diǎn)為空，則用 CAS 設(shè)置尾節(jié)點(diǎn)，如果入隊(duì)失敗則調(diào)用 enq(node)，這個(gè)方法內(nèi)部是一個(gè)循環(huán)，利用自旋 CAS 把節(jié)點(diǎn)加入同步隊(duì)列，具體代碼就不分析了。

在節(jié)點(diǎn)加入隊(duì)列之后，執(zhí)行的是 acquireQueued 方法，代碼如下：

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            // 這是一個(gè)無限循環(huán)
            for (;;) {
                final Node p = node.predecessor();
                // 如果前驅(qū)節(jié)點(diǎn)是 head，則嘗試獲取鎖
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                 // 獲取鎖失敗則進(jìn)入判斷是否要進(jìn)入睡眠
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

acquireQueued 實(shí)現(xiàn)了線程的睡眠與喚醒。在內(nèi)部是一個(gè)無限循環(huán)，每次獲取前驅(qū)節(jié)點(diǎn)，如果前驅(qū)結(jié)點(diǎn)是 HEAD，那么嘗試去獲取鎖，獲取成功則將此節(jié)點(diǎn)變?yōu)樾碌念^結(jié)點(diǎn)并將原先的頭結(jié)點(diǎn)出隊(duì)。如果前驅(qū)節(jié)點(diǎn)不是頭結(jié)點(diǎn)或者獲取鎖失敗，那么就會(huì)進(jìn)入 shouldParkAfterFailedAcquire 方法，判斷是否進(jìn)入睡眠，如果這個(gè)方法返回 true，則調(diào)用 parkAndCheckInterrupt 讓線程進(jìn)入睡眠狀態(tài)。下面是 parkAndCheckInterrupt 的代碼：

private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this); // 線程在這一步進(jìn)入阻塞狀態(tài)
        return Thread.interrupted();
}

對于 shouldParkAfterFailedAcquire 來說，如果前驅(qū)節(jié)點(diǎn)正常，那么會(huì)返回 true，表示當(dāng)前線程應(yīng)該掛起，如果前驅(qū)結(jié)點(diǎn)取消了排隊(duì)，那么當(dāng)前線程有機(jī)會(huì)搶鎖，此時(shí)返回 false，并繼續(xù) acquireQueued 中的循環(huán)。

解鎖

相比于加鎖，解鎖稍微簡單一點(diǎn)，看一下 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)
            // 喚醒后繼節(jié)點(diǎn)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

首先調(diào)用 tryRelease 解鎖，如果解鎖成功則喚醒后繼結(jié)點(diǎn)，返回值表示是否成功釋放鎖。那為什么會(huì)解鎖不成功，其實(shí)是因?yàn)橹厝耄匆幌?tryRelease 的代碼：

protected final boolean tryRelease(int releases) {
    //  更新 state 計(jì)數(shù)值
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    // 是否完全釋放鎖
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

將 state 減去對應(yīng)的值，如果 state == 0，那么說明鎖已經(jīng)完全釋放。

在 release 中，如果鎖已經(jīng)完全釋放，那么將調(diào)用 unparkSuccessor 喚醒后繼節(jié)點(diǎn)，喚醒的節(jié)點(diǎn)所代表的線程阻塞在 parkAndCheckInterrupt 中：

private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this); // 線程在這一步進(jìn)入阻塞狀態(tài)
        return Thread.interrupted();
}

線程被喚醒后，將繼續(xù) acquireQueued 中的循環(huán)，嘗試獲取鎖。

總結(jié)

本文簡要分析了 Lock 鎖的原理，主要是利用 AbstractQueuedSynchronizer這個(gè)關(guān)鍵的類。AQS 的核心在于使用 CAS 更新鎖的狀態(tài)，并利用一個(gè)同步隊(duì)列將獲取鎖失敗的線程進(jìn)行排隊(duì)，當(dāng)前驅(qū)節(jié)點(diǎn)解鎖后再喚醒后繼節(jié)點(diǎn)，是一個(gè)幾乎純 Java 實(shí)現(xiàn)的加鎖與解鎖。

如果我的文章對您有幫助，不妨點(diǎn)個(gè)贊支持一下(^_^)