AQS是個(gè)啥？

AQS隊(duì)列

AQS隊(duì)列示意圖

AQS隊(duì)列中的主要屬性

//  等待隊(duì)列頭部
private transient volatile Node head;

// 等待隊(duì)列尾部
private transient volatile Node tail;

// 鎖的狀態(tài)（加鎖成功則為1，解鎖為0，重入再 1）
private volatile int state;

//  當(dāng)前持有鎖的線(xiàn)程，注意這個(gè)屬性是從AbstractOwnableSynchronizer繼承而來(lái)
private transient Thread exclusiveOwnerThread;

Node類(lèi)中的主要屬性

static final class Node {
    // 標(biāo)記表示節(jié)點(diǎn)正在共享模式中等待
    static final Node SHARED = new Node();
    // 標(biāo)記表示節(jié)點(diǎn)正在獨(dú)占模式下等待
    static final Node EXCLUSIVE = null;

    // 節(jié)點(diǎn)的等待狀態(tài) 還有一個(gè)初始化狀態(tài)0 不屬于以下四種狀態(tài)
    // 表示Node所代表的當(dāng)前線(xiàn)程已經(jīng)取消了排隊(duì)，即放棄獲取鎖
    static final int CANCELLED =  1;
    // 當(dāng)一個(gè)節(jié)點(diǎn)的waitStatus被置為SIGNAL，就說(shuō)明它的下一個(gè)節(jié)點(diǎn)（即它的后繼節(jié)點(diǎn)）已經(jīng)被掛起了（或者馬上就要被掛起了），
    // 只要前繼結(jié)點(diǎn)釋放鎖，就會(huì)通知標(biāo)識(shí)為SIGNAL狀態(tài)的后繼結(jié)點(diǎn)的線(xiàn)程執(zhí)行
    static final int SIGNAL    = -1;
    // 節(jié)點(diǎn)在等待隊(duì)列中
    // 當(dāng)其他線(xiàn)程對(duì)Condition調(diào)用了signal()后，該節(jié)點(diǎn)將會(huì)從等待隊(duì)列中轉(zhuǎn)移到同步隊(duì)列中，加入到同步狀態(tài)的獲取中
    static final int CONDITION = -2;
    // 表示下一次共享式同步狀態(tài)獲取，將會(huì)無(wú)條件地傳播下去
    static final int PROPAGATE = -3;

    // 節(jié)點(diǎn)等待狀態(tài)，該字段初始化為0，
    volatile int waitStatus;

        // 當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)
    volatile Node prev;

    // 當(dāng)前節(jié)點(diǎn)的后置節(jié)點(diǎn)
    volatile Node next;

        // 在此節(jié)點(diǎn)上排隊(duì)的線(xiàn)程信息
    volatile Thread thread;
}

ReentrantLock實(shí)現(xiàn)

繼承關(guān)系圖

主要屬性

private final Sync sync;

// 公平鎖內(nèi)部是FairSync,非公平鎖內(nèi)部是NonfairSync。
// 兩者都通過(guò)繼承 Sync間接繼承自AbstractQueuedSynchronizer這個(gè)抽象類(lèi)
abstract static class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = -5179523762034025860L;
    
    // 加鎖
    abstract void lock();

    // 嘗試獲取鎖
    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;
    }
}

構(gòu)造方法

//默認(rèn)創(chuàng)建一個(gè)非公平鎖
public ReentrantLock() {
    sync = new NonfairSync();
}

//傳入true創(chuàng)建公平鎖，false非公平鎖
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

ReentrantLock公平鎖

我們以公平鎖為例對(duì)其中重要方法源碼分析

// 繼承了 Sync，從而間接繼承了 AbstractQueuedSynchronizer這個(gè)抽象類(lèi)
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

       // 上鎖
    final void lock() {
        //調(diào)用 AQS 中 acquire方法
        acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                // CAS操作設(shè)置 state
                // 設(shè)置當(dāng)前線(xiàn)程為擁有鎖的線(xiàn)程
                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;
    }
}

acquire方法源碼分析

public final void acquire(int arg) {
    // tryAcquire(arg)嘗試加鎖，如果加鎖失敗則會(huì)調(diào)用acquireQueued方法加入隊(duì)列去排隊(duì)，如果加鎖成功則不會(huì)調(diào)用
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

tryAcquire方法

protected final boolean tryAcquire(int acquires) {
    // 獲取當(dāng)前線(xiàn)程
    final Thread current = Thread.currentThread();
    // 獲取lock對(duì)象的上鎖狀態(tài)，如果鎖是自由狀態(tài)則=0，如果被上鎖則為1，大于1表示重入
    int c = getState();

    // c=0 代表沒(méi)人占用鎖，當(dāng)前線(xiàn)程可以直接獲取鎖資源執(zhí)行
    if (c == 0) {
        // 下面介紹hasQueuedPredecessors()方法，判斷自己是否需要排隊(duì)
        if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            // CAS操作設(shè)置 state
            // 設(shè)置當(dāng)前線(xiàn)程為擁有鎖的線(xiàn)程
            setExclusiveOwnerThread(current);
            return true;
        }
    }

    // 非重入鎖直接返回false，加鎖失敗
    else if (current == getExclusiveOwnerThread()) {
        // 若為重入鎖, state 加1 (acquires)
        int nextc = c   acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

hasQueuedPredecessors方法

public final boolean hasQueuedPredecessors() {
    // 獲取隊(duì)列頭、尾節(jié)點(diǎn)信息
    Node t = tail; 
    Node h = head;
    Node s;
    // h != t 有幾種情況
    // 1、隊(duì)列尚未初始化完成，第一個(gè)線(xiàn)程獲取鎖資源，
    //    此時(shí)h和t都是null, h != t返回fasle初始化隊(duì)列
    // 2、隊(duì)列已經(jīng)被初始化了，其他的線(xiàn)程嘗試獲取資源，
    //    此時(shí)頭尾節(jié)點(diǎn)不相同，h!=t返回true，
    //    繼續(xù)判斷s.thread != Thread.currentThread() 當(dāng)前來(lái)參與競(jìng)爭(zhēng)鎖的線(xiàn)程和第一個(gè)排隊(duì)的線(xiàn)程是同一個(gè)線(xiàn)程,則需要排隊(duì)。
    // 3、隊(duì)列已經(jīng)被初始化了，但是由于鎖釋放的原因?qū)е玛?duì)列里面只有一個(gè)數(shù)據(jù)
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

addWaiter方法

private Node addWaiter(Node mode) {
    // AQS隊(duì)列中的元素類(lèi)型為Node，需要把當(dāng)前線(xiàn)程封裝成為一個(gè)Node對(duì)象
    Node node = new Node(Thread.currentThread(), mode);

    // tail為隊(duì)尾，賦值給pred
    Node pred = tai
    // 判斷pred是否為空，其實(shí)就是判斷隊(duì)尾是否有節(jié)點(diǎn)，其實(shí)只要隊(duì)列被初始化了隊(duì)尾肯定不為空，
    if (pred != null) {
        // 拼裝node隊(duì)列鏈的過(guò)程
        // 直接把當(dāng)前線(xiàn)程封裝的node的上一個(gè)節(jié)點(diǎn)設(shè)置成為pred即原來(lái)的隊(duì)尾
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            // pred的下一個(gè)節(jié)點(diǎn)設(shè)置為當(dāng)node
            pred.next = node;
            return node;
        }
    }

    // 拼接aqs隊(duì)列鏈
    enq(node);
    return node;
}


private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

acquireQueued方法

final boolean acquireQueued(final Node node, int arg) {
    // 標(biāo)記是否成功拿到資源
    boolean failed = true;
    try {
        // 標(biāo)記等待過(guò)程中是否被中斷過(guò)
        boolean interrupted = false;
        // 自旋
        for (;;) {
            final Node p = node.predecessor();
            // 判斷自己是否為隊(duì)列中的第二個(gè)節(jié)點(diǎn)
            // 成為隊(duì)列中第二個(gè)節(jié)點(diǎn)才有資格獲取資源
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                // 返回等待過(guò)程中是否被中斷過(guò)
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

公平鎖和非公平鎖的主要區(qū)別

為了方便對(duì)比，在這里列舉了兩種鎖的上鎖過(guò)程源碼，注意紅色標(biāo)識(shí)片段

// 公平鎖上鎖過(guò)程
final void lock() {
    //調(diào)用 AQS 中 acquire方法
    acquire(1);
}　　

// 非公平鎖上鎖過(guò)程
final void lock() {
    // 嘗試獲取鎖，加鎖不成功則排隊(duì)。排隊(duì)之前僅有的一次插隊(duì)機(jī)會(huì)。
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

總結(jié)