JAVA多線程之JDK中的各種鎖詳解(看這一篇就夠了)

更新時(shí)間：2024年07月18日 11:01:38 作者：_BugMan

多線程編程可以說(shuō)是在大部分平臺(tái)和應(yīng)用上都需要實(shí)現(xiàn)的一個(gè)基本需求,下面這篇文章主要給大家介紹了關(guān)于JAVA多線程之JDK中各種鎖的相關(guān)資料,文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下

1.概論

1.1.實(shí)現(xiàn)鎖的要素

JAVA中的鎖都是可重入的鎖，因?yàn)椴豢芍厝氲脑囉玫臅r(shí)候很容易造成死鎖。這個(gè)道理很好想明白：

當(dāng)一個(gè)線程已經(jīng)持有一個(gè)鎖，并在持有該鎖的過(guò)程中再次嘗試獲取同一把鎖時(shí)，如果沒(méi)有重入機(jī)制，第二次請(qǐng)求會(huì)被阻塞，因?yàn)殒i已經(jīng)被自己持有。這會(huì)導(dǎo)致線程自我死鎖，因?yàn)樗诘却约横尫诺逆i。

可重入是指獲取鎖的線程可以繼續(xù)重復(fù)的獲得此鎖。其實(shí)我們想都能想到要實(shí)現(xiàn)一把鎖需要些什么，首先肯定是：

標(biāo)志位，也叫信號(hào)量，標(biāo)記鎖的狀態(tài)和重入次數(shù)，這樣才能完成持有鎖和釋放鎖。

接下來(lái)要考慮的是拒接策略，當(dāng)前鎖被持有期間，后續(xù)的請(qǐng)求線程該怎么處理，當(dāng)然可以直接拒絕，JAVA的選擇委婉點(diǎn)，選擇了允許這些線程躺在鎖上阻塞等待鎖被釋放。要實(shí)現(xiàn)讓線程躺在鎖上等待，我們想想無(wú)非要：

需要支持對(duì)一個(gè)線程的阻塞、喚醒
需要記錄當(dāng)前哪個(gè)線程持有鎖
需要一個(gè)隊(duì)列維護(hù)所有阻塞在當(dāng)前鎖上的線程

OK，以上四點(diǎn)就是JAVA鎖的核心，總結(jié)起來(lái)就是信號(hào)量+隊(duì)列，分別用來(lái)記錄持有者和等待者。

1.2.阻塞、喚醒操作

首先我們來(lái)看看阻塞和喚醒的操作，在JDK中提供了一個(gè)Unsafe類(lèi)，該類(lèi)中提供了阻塞或喚醒線程的一對(duì)操作原語(yǔ)——park/unpark：

public native void unpark(Object var1);
public native void park(boolean var1, long var2);

這對(duì)原語(yǔ)最終會(huì)調(diào)用操作系統(tǒng)的程序接口執(zhí)行線程操作。

1.2.阻塞隊(duì)列

拿來(lái)維護(hù)所有阻塞在當(dāng)前鎖上的線程的隊(duì)列能是個(gè)普通隊(duì)列嗎？很顯然不是，它的操作必須是線程安全的是吧，所以這個(gè)隊(duì)列用阻塞隊(duì)列實(shí)現(xiàn)才合適。什么是阻塞隊(duì)列：

阻塞隊(duì)列提供了線程安全的元素插入和移除操作，并且在特定條件下會(huì)阻塞線程，直到滿足操作條件。

說(shuō)到JDK中的阻塞隊(duì)列，其核心就是AbstractQueuedSynchronizer，簡(jiǎn)稱(chēng)AQS，由雙向鏈表實(shí)現(xiàn)的一個(gè)元素操作絕對(duì)安全的隊(duì)列，用來(lái)在鎖的實(shí)現(xiàn)中維護(hù)阻塞在鎖上的線程上的隊(duì)列的這個(gè)角色。

來(lái)看看AQS的源碼：

它有指向前后節(jié)點(diǎn)的指針、有一個(gè)標(biāo)志位state、還有一個(gè)提供線程操作原原語(yǔ)（阻塞、喚醒）的unsafe類(lèi)。

所以其實(shí)AQS就長(zhǎng)這樣：

點(diǎn)進(jìn)源碼可以看到其隨便一個(gè)方法都是線程安全的：

由于本文不是專(zhuān)門(mén)聊AQS這里就不擴(kuò)展了，反正知道AQS是一個(gè)線程安全的阻塞隊(duì)列就對(duì)了。

1.3.Lock接口和Sync類(lèi)

JAVA中所有鎖的頂級(jí)父接口，用來(lái)規(guī)范定義一把鎖應(yīng)該有那些行為職責(zé)：

public interface Lock {
    void lock();
    void lockInterruptibly() throws InterruptedException;
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    void unlock();
    Condition newCondition();
}

JAVA中所有鎖的實(shí)現(xiàn)都是依托AQS去作為阻塞隊(duì)列，每個(gè)鎖內(nèi)部都會(huì)實(shí)現(xiàn)一個(gè)Sync內(nèi)部類(lèi)，在自身Sync內(nèi)部以不同的策略去操作AQS實(shí)現(xiàn)不同種類(lèi)的鎖。

abstract static class Sync extends AbstractQueuedSynchronizer {......}

2.各種鎖

2.1.互斥鎖

2.1.1.概論

ReentrantLock，互斥鎖，ReentrantLock本身沒(méi)有任何代碼邏輯，依靠?jī)?nèi)部類(lèi)Sync干活兒：

public class ReentrantLock implements Lock, Serializable {
    private final ReentrantLock.Sync sync;
    public void lock() {
        this.sync.lock();
    }
    public void unlock() {
        this.sync.release(1);
    }
    ......
}

ReentrantLock的Sync繼承了AQS

abstract static class Sync extends AbstractQueuedSynchronizer {......}

Sync是抽象類(lèi)，有兩個(gè)實(shí)現(xiàn)：

NonfairSync，公平鎖
FairSync，非公平鎖

實(shí)例化ReentrantLock的實(shí)例時(shí)，根據(jù)傳入的標(biāo)志位可以創(chuàng)建公平和公平的實(shí)現(xiàn)

public class ReentrantLock implements Lock, java.io.Serializable{
public ReentrantLock() {
        sync = new NonfairSync();
    }
?
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }
    ......
}
}

2.1.2.源碼

1.lock()

公平鎖的lock()：

static final class FairSync extends Sync {
        final void lock() {
            acquire(1);//進(jìn)來(lái)直接排隊(duì)
        }

非公平鎖的lock():

static final class NonfairSync extends Sync {
        final void lock() {
            if (compareAndSetState(0, 1))//進(jìn)來(lái)直接搶鎖
                setExclusiveOwnerThread(Thread.currentThread());//將鎖的持有者設(shè)置為當(dāng)前線程
            else
                acquire(1);//沒(méi)搶過(guò)再去排隊(duì)
        }
    }

acquire（）是AQS的模板方法：

tryAcquire，嘗試再去獲取一次鎖，公平鎖依然是排隊(duì)搶?zhuān)タ纯醋枞?duì)列是否為空；非公平鎖依然是直接搶。

acquireQueued，將線程放入阻塞隊(duì)列。

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

acquireQueued（..）是lock()最關(guān)鍵的一部分，addWaiter（..）把Thread對(duì)象加入阻塞隊(duì)列，acquireQueued（..）完成對(duì)線程的阻塞。

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {//如果發(fā)現(xiàn)自己在隊(duì)頭就去拿鎖
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())//調(diào)用原語(yǔ)，阻塞自己
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

acquireQueued（..）函數(shù)有一個(gè)返回值，表示什么意思呢？雖然該函數(shù)不會(huì)中斷響應(yīng)，但它會(huì)記錄被阻塞期間有沒(méi)有其他線程向它發(fā)送過(guò)中斷信號(hào)。如果有，則該函數(shù)會(huì)返回true；否則，返回false。所以才有了以下邏輯：

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

當(dāng) acquireQueued（..）返回 true 時(shí)，會(huì)調(diào)用 selfInterrupt （），自己給自己發(fā)送中斷信號(hào)，也就是自己把自己的中斷標(biāo)志位設(shè) 為true。之所以要這么做，是因?yàn)樽约涸谧枞陂g，收到其他線程中斷信號(hào)沒(méi)有及時(shí)響應(yīng)，現(xiàn)在要進(jìn)行補(bǔ)償。這樣一來(lái)，如果該線程在loc k代碼塊內(nèi)部有調(diào)用sleep（）之類(lèi)的阻塞方法，就可以拋出異常，響應(yīng)該中斷信號(hào)。

2.unlock()

unlock的邏輯很簡(jiǎn)單，每次unlock，state-1，直到state=0時(shí)，將鎖的擁有者置null，釋放鎖。由于只有鎖的持有線程才能操作lock，所以u(píng)nlock()不需要用CAS，操作時(shí)直接判斷一下是不是鎖的持有線程在操作即可。

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);//喚醒阻塞隊(duì)列中的后繼者
            return true;
        }
        return false;
    }

釋放鎖：

protected final boolean tryRelease(int releases) {
            int c = getState() - releases;//每次unlock，state減1
            if (Thread.currentThread() != getExclusiveOwnerThread())//判斷是不是鎖的持有線程
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {//state為0表示該鎖沒(méi)有被持有
                free = true;
                setExclusiveOwnerThread(null);//將鎖的持有者置null
            }
            setState(c);
            return free;
        }

喚醒后繼者：

private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

2.2.讀寫(xiě)鎖

讀寫(xiě)鎖是一個(gè)實(shí)現(xiàn)讀寫(xiě)互斥的鎖，讀寫(xiě)鎖包含一個(gè)讀鎖、一個(gè)寫(xiě)鎖：

public interface ReadWriteLock{
    Lock readLock();
    Lock writeLock();
}

讀寫(xiě)鎖的使用就是直接調(diào)用對(duì)應(yīng)鎖進(jìn)行鎖定和解鎖：

ReadWriteLock rwLock=new ReetrantReadWriteLock();
Lock rLock=rwLock.readLock();
rLock.lock();
rLock.unLock();
Lock wLock=rwLock.writeLock();
wLock.lock();
wLock.unLock();

讀寫(xiě)鎖的Sync內(nèi)部類(lèi)對(duì)讀鎖和寫(xiě)鎖采用同一個(gè)int型的信號(hào)量的高16位和低16位分別表示讀寫(xiě)鎖的狀態(tài)和重入次數(shù)，這樣一次CAS就能統(tǒng)一處理進(jìn)行讀寫(xiě)互斥操作：

abstract static class Sync extends AbstractQueuedSynchronizer {
        static final int SHARED_SHIFT   = 16;
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
}

2.3.Condition

2.3.1.概論

condition用于更加細(xì)粒度的控制鎖上面的線程阻塞、喚醒。

以下以一個(gè)經(jīng)典的生產(chǎn)、消費(fèi)者問(wèn)題為例：

隊(duì)列空的時(shí)候進(jìn)來(lái)的消費(fèi)者線程阻塞，有數(shù)據(jù)放進(jìn)來(lái)后喚醒阻塞的消費(fèi)者線程。

隊(duì)列滿的時(shí)候進(jìn)來(lái)的生產(chǎn)者線程阻塞，有空位后喚醒阻塞的生產(chǎn)者線程。

鎖粒度的實(shí)現(xiàn)：

public void enqueue(){
    synchronized(queue){
        while(queue.full()){
            queue.wait();
        }
        //入隊(duì)列
        ......
        //通知消費(fèi)者，隊(duì)列中有數(shù)據(jù)了
        queue.notify();
    }
}
?
public void dequeue(){
    synchronized(queue){
        while(queue.empty()){
            queue.wait();
        }
        //出隊(duì)列
        ......
        //通知生產(chǎn)者，隊(duì)列中有空位了，可以繼續(xù)放數(shù)據(jù)
        queue.notify();
    }
}

可以發(fā)現(xiàn)，喚醒的時(shí)候把阻塞的生產(chǎn)消費(fèi)線程一起喚醒了。

條件粒度的實(shí)現(xiàn)：

private final Lock lock = new ReentrantLock();
private final Condition notFull  = lock.newCondition(); // 用于等待隊(duì)列不滿
private final Condition notEmpty = lock.newCondition(); // 用于等待隊(duì)列非空

public void enqueue(Object item) {
    try {
        while (queue.isFull()) {
            notFull.await(); // 等待隊(duì)列不滿
        }
        // 入隊(duì)列操作
        // ...
        
        // 入隊(duì)后，通知等待的消費(fèi)者
        notEmpty.signal();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt(); // 保持中斷狀態(tài)
        // 處理中斷邏輯
    } finally {
        queue.unlock();
    }
}

public void dequeue() {
    try {
        while (queue.isEmpty()) {
            notEmpty.await(); // 等待隊(duì)列非空
        }
        // 出隊(duì)列操作
        // ...
        
        // 出隊(duì)后，通知等待的生產(chǎn)者
        notFull.signal();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt(); // 保持中斷狀態(tài)
        // 處理中斷邏輯
    } finally {
        queue.unlock();
    }
}

2.3.2.底層實(shí)現(xiàn)

Condition由Lock產(chǎn)生，因此Lock中持有Condition：

public interface Lock {
    ......
    Condition newCondition();
}

承擔(dān)功能的其實(shí)就是Syn中的ConditionObject，也就是AQS中的ConditionObject：

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

一個(gè)Condition上面阻塞著多個(gè)線程，所以每個(gè)Condition內(nèi)部都有一個(gè)隊(duì)列，用來(lái)記錄阻塞在這個(gè)condition上面的線程，這個(gè)隊(duì)列其實(shí)也是AQS實(shí)現(xiàn)的，AQS中除了實(shí)現(xiàn)一個(gè)以Node為節(jié)點(diǎn)的隊(duì)列，還實(shí)現(xiàn)了一個(gè)以ConditionObject為節(jié)點(diǎn)的隊(duì)列：

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
        public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        private transient Node firstWaiter;
        private transient Node lastWaiter;
        ......
        }
    }

Condition是個(gè)接口，定義了一系列條件操作：

public interface Condition {
    void await() throws InterruptedException;
    void awaitUninterruptibly();
    long awaitNanos(long var1) throws InterruptedException;
    boolean await(long var1, TimeUnit var3) throws InterruptedException;
    boolean awaitUntil(Date var1) throws InterruptedException;
    void signal();
    void signalAll();
}