併發編程第八天--------大廠必問之 AQS 源碼解析

時間  2020-06-03
標籤 併發 編程 第八 大廠 aqs 源碼 解析 简体版
原文   原文鏈接

LockSupport 工具類

LockSupport 是個工具類，主要做用是掛起和喚醒線程，該工具類是建立鎖和其餘同步類的基礎。java

LockSupport 類與每一個使用他的線程都會關聯一個許可證，在默認狀況下調用 LockSupport 類方法的線程是不持有許可證的。LockSupport 使用 Unsafe 類實現的。node

park() 方法

若是調用 park 方法的線程已經拿到了與 LockSupport 關聯的許可證，則調用Locksupport. park()時會立刻返回，不然調用線程會被阻塞掛起。安全

public static void main(String[] args) {
    System.out.println("beign park");
    
    LockSupport.park();
    
    System.out.println("end park");
}
複製代碼

這個例子中，主線程打印出 begin park 後就會被阻塞掛起，由於默認狀況下調用線程不持有許可證。如何才能喚醒由於調用 LockSupport.park() 被阻塞的線程呢，有下面兩種方法:框架

  • 其餘線程調用 LockSupport.unpark(Thread t) ,以被阻塞的線程做爲參數，那麼被阻塞的線程會被喚醒。
  • 其餘線程調用了被阻塞線程的 interrupt() ，被則塞的線程會被喚醒。
public static void main(String[] args) throws InterruptedException {
        Thread t1 = new Thread(()->{
            System.out.println("t1 beign park");

            LockSupport.park();

            System.out.println("t1 end park");
        });

        t1.start();

        Thread.sleep(2000);
        
        // t1 被喚醒
        LockSupport.unpark(t1);  // 用這個也行 t1.interrupt();
    }
複製代碼

unpark()

當一個線程調用 unpakr() 時，若是線程沒有持有與 LockSupport 關聯的許可證，則會讓線程持有許可證。工具

public static void main(String[] args) throws InterruptedException {
    // 讓主線程持有許可證
    LockSupport.unpark(Thread.currentThread());

    System.out.println("beign park");

    LockSupport.park();
    
    // LockSupport.park(); (1)
    
    System.out.println("end park");
}
複製代碼

此時 begin park 和 end park 都能被輸出。ui

park() 方法不具備可重入性，將上面的代碼 (1) 處註釋打開，就會發現線程仍然會被阻塞。this

Condition 接口

任意一個 Java 對象，都有一組監視器方法(定義在 java.lang.Object 上)，主要包括 wait()、notify()、notifyAll()，這些方法與 synchronized 關鍵字配合，實現等待通知模式。spa

Condition 接口也提供了相似 Object 的監視器方法，與 Lock 配合能夠實現等待通知模式，await() 相似 wait()，signal() 相似 notify()，signalAll() 相似 notifyAll()。線程

他們之間最主要的區別是 synchronized 監視器鎖只能實現一個等待隊列，而 lock 能實現多個等待隊列 (一個 lock 對象能夠建立出多個不一樣的 Condition 實例)。3d

使用

使用方法相似 Object 中的 wait() 和 notify()

public static void main(String[] args) throws InterruptedException {
        Lock lock = new ReentrantLock();

        Condition condition = lock.newCondition();
        
        Thread t1 = new Thread(()->{
            lock.lock();
            System.out.println("t1 begin await");
            try {
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("t1 end await");
            lock.unlock();
        });

        t1.start();

        Thread.sleep(2000);

        System.out.println("begin signal");

        lock.lock();
        condition.signal();
        lock.unlock();
    }
複製代碼

Condition 對象是由 Lock 對象建立的。await() 和 notify() 在使用前要先獲取到鎖。

AbstractQueuedSynchronizer

抽象同步隊列 AbstractQueuedSynchronizer 簡稱 AQS 是用來構建鎖或者其餘同步組件的基礎框架，經過內置的 FIFO 隊列來完成同步狀態的管理。如下是 AQS 的結構圖。

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { 
    
    // 隊列的頭節點
    private transient volatile Node head;
    
    // 隊列的尾節點
    private transient volatile Node tail;
    
    //同步狀態
    private volatile int state;
    
    protected final int getState() { return state;}
    
    protected final void setState(int newState) { state = newState;}
    
    ...
}
複製代碼

AQS 內部維護了一個狀態變量 state，對於 ReentrantLock 的實現，state 表示當前獲取鎖的線程的重入次數，對於讀寫鎖來 ReentrantReadWriteLock 或者 CountDonwLatch 來講, state 又表示不一樣的含義，後面再詳解。

對於 AQS 來講，線程同步的關鍵就是對狀態值 state 進行操做，根據 state 的值判斷鎖是否被線程佔用。

同步器提供了 getState()、setState()、compareAndSetState() 三個方法來操做 state 變量。

exclusiveOwnerThread 記錄了當前持有資源的線程。

AQS 是一個雙向隊列，內部經過 head 和 tail 記錄隊首和隊尾元素。

隊列元素的類型是 Node。咱們來看看 Node 節點的具體信息。

Node 節點

當前線程獲取同步狀態失敗時，同步器會將當前線程以及等待信息構形成一個 Node 節點並將其加入同步隊列。

static final class Node {
        
        // 表示當前節點在 共享模式下 等待的標誌
        static final Node SHARED = new Node();
        // 表示當前節點在 獨佔模式下 等待的標誌
        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;

        // 前驅節點
        volatile Node prev;

        // 後繼節點
        volatile Node next;

        // 節點內的線程
        volatile Thread thread;
    }
複製代碼

每一個節點除了存儲當前線程和先後節點的引用外，還有一個 waitStatus，表示節點中線程的等待狀態。

Node 節點有線程四種等待狀態

  • CANCELLED，表示同步隊列中等待的線程由於某種緣由（好比中斷、超時）而取消等待
  • SIGNAL，指示後續節點的線程須要 unpark()，只有前驅節點的線程狀態爲 SIGNAL 時，當前節點的線程才能被阻塞
  • CONDITION : 指示線程正在條件隊列中等待，用於條件隊列
  • PROPAGATE : 不是很經常使用，能夠不用知道

下面咱們經過 ReentrantLock 的實現來深刻分析入隊和出隊過程。

ReentrantLock

ReentrantLock 內部有一個 Sync 抽象類繼承自 AbstractQueuedSynchronizer

abstract static class Sync extends AbstractQueuedSynchronizer {
        
        // 由 Sync 的子類實現
        abstract void lock();

        final boolean nonfairTryAcquire(int acquires) {
            // 具體實現....
        }

        protected final boolean tryRelease(int releases) {
            // 具體實現....
        }
        ...
    }
複製代碼

ReentranLock 鎖有兩種模式，公平模式和非公平模式，分別在 Sync 的兩個子類 FairSync 和 NonfairSync 中實現。

private final Sync sync;
    
    public ReentrantLock() {
        sync = new NonfairSync();
    }

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

sync 變量的默認實現是非公平方式。咱們下面分析的源碼是基於非公平模式。

lock()

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

一個線程獲取鎖時，先嚐試 CAS 設置 state，若是設置成功，則設置當前線程爲鎖的持有者，設置失敗則嘗試調用 acquire()。

nonfairTryAcquire()

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&  // 獲取失敗則調用 addWaiter()
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

// 非公平模式嘗試獲取，獲取失敗返回 false
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) { // 再次嘗試 CAS 設置 state 
        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; // 獲取成功
}
複製代碼

若是 state 爲 0，則直接 嘗試 CAS 設置 state 並佔有鎖，或者當前線程是鎖的持有線程，state 則 +1 (ReentrantLock 可重入的體現)，不然獲取失敗，返回 false。

獲取失敗後會調用 addWaiter()

addWaiter()

構造 Node 節點併入隊。

// 返回新添加的節點
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode); // 構造 Node 節點
   
    Node pred = tail; 
    if (pred != null) {
        node.prev = pred;
        // 嘗試將尾節點指向新節點，若是失敗則調用 enq() 方法
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}


private Node enq(final Node node) {
    for (;;) { // 無限循環
        Node t = tail;
        if (t == null) {
            // 初始化隊列
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            // 將節點添加到隊列中
            node.prev = t;
            // CAS 設置尾節點爲添加的節點
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
複製代碼

初始化隊列在 enq() 進行，將 head 和 tail 同時指向一個空引用，隊列初始化完成。

在節點入隊的過程當中，經過 CAS 將尾節點指向新添加的節點，保證了線程安全。

節點入隊後，並不會當即阻塞掛起 ，而是會調用 acquireQueued(node,args) 方法，再次嘗試獲取鎖資源，由於在節點入隊的過程當中，以前的線程可能執行完成。

acquireQueued()

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)) { // 若是 p 是頭節點 而且 當前線程嘗試獲取鎖成功
                    setHead(node);   // 頭結點指針指向當前節點，並清空節點 thread 屬性
                    p.next = null; // GC 回收
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    
    
    private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }
複製代碼

若是當前節點的前驅節點是頭結點 而且 當前線程再次嘗試獲取鎖成功後，則將 head 引用執行當前節點，並清空當前節點的 thread 和 prev 信息，而後返回 false。

不然執行 shouldParkAfterFailedAcquire(p,node)，判斷當前節點是否應該被則塞，若是返回 true 表明當前節點應該阻塞。

shouldParkAfterFailedAcquire()

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            
            // 只有前驅節點是 siganl 時，節點才能被則塞
            return true;
        if (ws > 0) {
            // 刪除前驅節點
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 設置前驅節點的 waitStatus 爲 -1
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false; // 返回 false
    }
複製代碼

只有當前面節點的線程狀態爲 SIGNAL 時，當前節點的線程才能被則塞掛起。

  • 若是 pred.waitStatus==0，則經過 CAS 修改其爲 Node.SIGNAL
  • 若是 pred.waitStatus>0，則從隊列中刪除 pred 節點
  • 若是 pred.waitStatus==-1，則返回 true，表示當前節點應該被阻塞，而後調用parkAndCheckInterrupt() 方法阻塞線程
private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this); // 阻塞
        return Thread.interrupted(); // 中斷檢測機制
    }
複製代碼

從 acquireQueued() 中能夠看出，線程被喚醒後不必定能獲取鎖，還得再次競爭，這就是非公平的體現。

流程圖總結

unlock()

咱們來看看解鎖的過程

// 釋放鎖
    public void unlock() {
        sync.release(1);
    }
複製代碼

release()

public final boolean release(int arg) {
        if (tryRelease(arg)) { // 釋放成功
            Node h = head;
            if (h != null && h.waitStatus != 0)  // 若是頭節點不爲空而且頭節點的 waitstatus 不爲 0
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    
    protected final boolean tryRelease(int releases) {
        int c = getState() - releases; // 若是 c=0，則表明 state 爲 1
        if (Thread.currentThread() != getExclusiveOwnerThread()) // 判斷當前線程是不是持有鎖的線程
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;   
            setExclusiveOwnerThread(null); // 釋放鎖資源
        }
        setState(c); // 設置 state
        return free;
    }
複製代碼

釋放鎖成功後會調用 unparkSuccessor(head)，

unparkSuccessor(head)

private void unparkSuccessor(Node node) {
        
        
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0); // CAS 設置 waitStatus

        
        
        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);  // 喚醒下一個節點
    }
複製代碼

若是頭節點的 waitStatus<0 ，經過 CAS 設置其爲 0 。而後喚醒下一個節點

流程圖總結

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程