并发编程之AQS源码

特别鸣谢，参考文章地址：

https://www.cnblogs.com/waterystone/p/4920797.html

https://javadoop.com/post/AbstractQueuedSynchronizer-2

1.什么是AQS?

最近一直在学并发编程。上一边博客写了CountDownLatch。有兴趣的可以去瞄一瞄：https://www.cnblogs.com/pluto-charon/p/12887902.html

谈到并发，就不得不谈到AbstractQueuedSynchronizer(AQS)。AQS,根据字面意思，抽象的队列式同步器，也叫同步器。AQS定义了一套多线程访问共享资源的同步器框架，许多并发的技术的实现都依赖于它，例如：ReentrantLock/Semaphore/CountDownLatch…

​

2.AQS的说明

​ 如上图所示，AQS内部维护了一个volatile int state（代表共享资源，表示加锁的状态）和一个先入先出的队列Node（多线程挣用资源被阻塞时会进入此队列）。AQS还提供了一个内部类ConditionObject，实现了条件队列Condition接口。当前线程可以通过signal和signalAll将条件队列中的节点转移到同步队列中。当前线程存在于同步队列的头节点，可以通过await从同步队列转移到条件队列中。

AQS定义两种资源共享方式：

1.Exclusive（独占，只有一个线程能执行，如ReentrantLock、ReentrantReadWriteLock.WriteLock），独占锁的实现方式为：acquire(int)、release(int)。

2.Share（共享，多个线程可同时执行，如ReentrantReadWriteLock.ReadLock、CountDownLatch、CyclicBarrier、Semaphore），共享锁实现方式为：acquireShared(int)、releaseShared(int)。

不同的自定义同步器争用共享资源的方式也不同，同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。

AQS是一个抽象类，但这几个获取资源和释放资源的方法并没有声明为抽象方法，目的就是为了让不同的同步器根据不同的功能可以自由实现这些方法。一般来说，自定义同步器要么是独占方法，要么是共享方式，他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock。

3.源码详解

接下来我们按照acquire-release，acquireShared-releaseShared的方式来讲讲AQS的重要方法的源码！

在查看源码之前，我们需要了解等待队列中的状态值：

/** 表示线程已经被取消，当timeout或被中断（响应中断的情况下），会触发变更为此状态，进入该状态后的结点将不会再变化。*/
static final int CANCELLED =  1;
/**表示下一个节点的线程在等待当前节点唤醒，后继结点入队时，会将前继结点的状态更新为SIGNAL。 */
static final int SIGNAL    = -1;
/**表示结点等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁 */
static final int CONDITION = -2;
/**共享模式下，前继结点不仅会唤醒其后继结点，同时也可能会唤醒后继的后继结点。*/
static final int PROPAGATE = -3;
0；//新节点进入时的状态，默认状态

1.acquire(int arg)：获取资源

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

static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

tryAcquire(int)此方法尝试去获取独占资源。如果获取成功，则直接返回true，否则直接返回false。从上面源码可以看到，AQS这里只定义了一个接口，这个方法是需要具体的同步器去实现的。

acquire方法非常简单，如果tryAcquire失败（返回false），则调用acquireQueued方法，将当前线程加入到等待队列中，并中断当前线程，等待唤醒。

protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
}

addWaiter(mode)此方法用于将当前线程加入到等待队列的对位，并返回当前线程所在的节点。

private Node addWaiter(Node mode) {
    // 以给定模式构造结点。mode有两种：EXCLUSIVE（独占）和SHARED（共享）
    Node node = new Node(Thread.currentThread(), mode);
    // 尝试快速方式直接放到队尾。
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        //通过cas机制设置到队列尾部
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 上一步失败则通过enq入队。
    enq(node);
    return node;
}

addWaiter()方法中还调用了一个enq的方法，此方法用于将node添加到队列尾部。

private Node enq(final Node node) {
    // CAS"自旋"，直到成功加入队尾
    for (;;) {
        Node t = tail;
        if (t == null) { // 队列为空，创建一个空的标志结点作为head结点，并将tail也指向它。
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {//正常流程，放入队尾
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

acquireQueued(node,int)此方法用于让线程进入等待状态休息，知道其他线程彻底释放资源后唤醒自己，自己再拿到资源并执行。

final boolean acquireQueued(final Node node, int arg) {
    // 标记是否成功拿到资源
    boolean failed = true;
    try {
        // 标记等待过程中是否被中断过
        boolean interrupted = false;
        for (;;) {//自旋操作
            //获取到当前node节点的前驱
            final Node p = node.predecessor();
            //如果前驱是head，即当前节点为第二个，那么便有资格去尝试获取资源（可能是head节点释放完资源唤醒自己的，当然也可能被interrupt了）。
            if (p == head && tryAcquire(arg)) {
                //拿到资源后，将head指向该节点。所以head所指的标杆结点，就是当前获取到资源的那个结点或null。
                setHead(node);
                p.next = null; // help GC
                failed = false;
                // 返回等待过程中是否被中断过
                return interrupted;
            }
            //用于检查自己是否真的可以去休息了。如果自己可以休息了，就进入waiting状态，直到被unpark()
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                //如果等待过程中被中断过，哪怕只有那么一次，就将interrupted标记为true
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

shouldParkAfterFailedAcquire(Node, Node)此方法主要用于检查状态，看看自己是否真的可以去休息了（进入waiting状态）。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 拿到前一个节点的状态
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            //如果后续线程需要断开连接，表示节点已设置状态，要求前节点拿到资源后通知自己一下
            return true;
        if (ws > 0) {//大于0即被取消的状态
            //前驱已经被取消了，就一直往前找，直到找到最近的一个正常等待状态的节点，并排在他后边
            //因为有一些线程在等待的过程中，放弃了等待被取消了，这种相当于形成了无用链，会被GC回收
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
           //如果前驱正常，那就把前驱的状态设置成SIGNAL，告诉它拿完号后通知自己一下。
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

整个流程中，如果前驱结点的状态不是SIGNAL，那么自己就不能安心去休息，需要去找个安心的休息点，同时可以再尝试下看有没有机会轮到自己拿。

parkAndCheckInterrupt()此方法表示判断线程是否进入等待状态并且检查线程是否被中断

private final boolean parkAndCheckInterrupt() {
    //调用park()使线程进入waiting状态
    LockSupport.park(this);
    //我以前一度认为这个方法是中断线程的，原来是检查当前线程是否被中断过
    return Thread.interrupted();
}

public static boolean interrupted() {
    return currentThread().isInterrupted(true);
}

再来用一张流程图总结一下这个方法吧：

1. 调用自定义同步器的tryAcquire()尝试直接去获取资源，如果成功则直接返回；
2. 没成功，则addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；
3. acquireQueued()使线程在等待队列中休息，有机会时（轮到自己，会被unpark()）会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。
4. 如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。

2.release(int)：释放资源

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

此方法是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源，如果彻底释放了（即state=0）,它会唤醒等待队列里的其他线程来获取资源。这也正是unlock()的语义，当然不仅仅只限于unlock()。

逻辑并不复杂。它调用tryRelease()来释放资源。有一点需要注意的是，它是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了！所以自定义同步器在设计tryRelease()的时候要明确这一点！！

tryRelease(int)此方法尝试去释放指定量的资源

protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}

跟tryAcquire()一样，这个方法是需要独占模式的自定义同步器去实现的。正常来说，tryRelease()都会成功的。因为这是独占模式，该线程来释放资源，那么他肯定已经拿到了独占资源了，直接减掉相应量的资源即可（state -= arg）;也不需要考虑线程安全的问题。因为release()是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了！所以自义定同步器在实现时，如果已经彻底释放资源(state=0)，要返回true，否则返回false。

unparkSuccessor(Node)此方法用于唤醒等待队列中下一个有效的线程

private void unparkSuccessor(Node node) {
    //获取当前节点的等待状态
    int ws = node.waitStatus;
    if (ws < 0)
        //使用cas机制将当前节点设置为0，0是节点刚进入时的状态，表示
        compareAndSetWaitStatus(node, ws, 0);
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        //如果当前节点为空或者已经被取消，则从后往前找，找到所有的<= 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);
}

这个函数并不复杂。一句话概括：用unpark()唤醒等待队列中最前边的那个未放弃线程，这里我们也用s来表示吧。此时，再和acquireQueued()联系起来，s被唤醒后，进入if (p == head && tryAcquire(arg))的判断（即使p!=head也没关系，它会再进入shouldParkAfterFailedAcquire()寻找一个安全点。这里既然s已经是等待队列中最前边的那个未放弃线程了，那么通过shouldParkAfterFailedAcquire()的调整，s也必然会跑到head的next结点，下一次自旋p==head就成立啦），然后s把自己设置成head标杆结点，表示自己已经获取到资源了，acquire()也返回了！！

3.acquireShared(int)：共享模式下获取共享资源

此方法是共享模式下线程获取资源的顶层入口，他会获取指定量的资源，获取成功则直接返回，获取失败则进入等待队列，直到获取到资源位置，整个过程忽略中断。

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

tryAcquireShared(int)这个方法一样是需要具体的同步器实现的，尝试获取资源，如果获取到了，就直接返回，否则就进入等待队列，直到获取到资源为止才返回。

doAcquireShared(int) 此方法用于将当前咸亨加入等待队列尾部休息，直到其他线程释放资源唤醒自己，自己成功拿到相应的资源后才返回。

private void doAcquireShared(int arg) {
    //加入到队尾
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    //设置队列头，并检查后续进程是否可能在共享模式下等待，如果是这样，则在设置了propagate>0或propagate status时进行传播。
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    //如果后面的这个if判断成功了，返回true，那么这里就自我中断
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            //判断状态，寻找安全点，进入waiting状态，等着被unpark()或interrupt()这里如果返回true，在一次循环里上面的判断就会进行自我中断。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

setHeadAndPropagate(Node, int)此方法在setHead()的基础上多了一步，就是自己苏醒的同时，如果条件符合(比如还有剩余资源)，还会去唤醒后继节点，毕竟是共享模式。

4.releaseShared(int)：共享模式下释放资源

此方法是共享模式下，线程释放共享资源的顶层入口。它是释放指定量的资源，如果成功释放且允许唤醒等待线程，它会唤醒等待队列里的其他线程来获取资源。

public final boolean releaseShared(int arg) {
    // 尝试释放资源
    if (tryReleaseShared(arg)) {
        //唤醒后续节点
        doReleaseShared();
        return true;
    }
    return false;
}

doReleaseShared()此方法主要用于唤醒后继节点

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);//唤醒后继
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}