Java并发之AQS原理

本文是Java并发专题（欢迎大家关注）的一篇。
以下是完整的目录：
Java并发之基础知识
Java并发之volatile关键字
Java并发之synchronized关键字
Java并发之原子类
Java并发之线程池
Java并发之并发工具类
Java并发之AQS原理

一.总体框架

AQS是指AbstractQueuedSynchronizer。它是一个笼统类，java并发包里的ReentrantLock、CountDownLatch和Semaphroe等重要的工具类都是基于AQS来实现的。

总体来说，AQS维护了一个volatile的state变量代表共享资源，还有一个FIFO的等待队列，在多线程争夺资源被阻塞时会进入此退了。等待队列是个双向链表记录则没有获取的执行许可的线程。等待队列中的结点元素是AQS自己设置的static的内部类Node。AQS支持共享和独占两种模式。ReentrantLock就是独占型的，只有一个线程可以取得到锁并执行。CountDownLatch和Semaphore就是共享型，允许多个线程同时执行。

AQS是一个笼统类，并不能被直接实例化使用。它的作用是提供等待队列的管理，包括如何入队何时唤醒等。而具体的资源如何获取和释放等由具体的自己设置同步器来实现。也就是说ReentrantLock等类自己设置了资源(state)的获取和释放，而使用AQS的来管理阻塞队列。不同的自己设置资源获取方式实现了CountDownLatch和Semaphore等类。

自己设置同步方法需要实现的方法有：

1. isHeldExclusively() //返回该线程能否正在独占资源，只有用的condition才需要去实现它
2. tryAcquire(int); //独占方式，尝试获取资源，成功返回true，失败返回false
3. tryRelease(int); //独占方式，尝试释放资源，成功返回true，失败返回false
4. tryAcquireShared(int); //共享方法，尝试获取资源。返回负数表示失败，0表示成功，但没有可用资源了，正数表示成功且有剩余资源
5. tryReleaseShared(int);//共享方式。尝试释放资源。假如释放后运行唤醒后续结点返回true，否则返回false

这其中tryAcquire和tryRelease是一组，用于实现独占资源的情况，如ReentrantLock；tryAcquireShared和tryReleaseShared是一组用于实现共享资源的情况，如CountDownLatch。

二.源码分析

2.1 acquire方法源码详解

在AQS中一个重要的方法是acquire(int)，这个方法实现请求资源和阻塞线程的功能。下面先贴一下它的源码：

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

这个方法里只有一个if语句。首先执行tryAcquire(int)方法，前面说了这个方法需要子类来自己设置，这里先看一下它的代码：

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

可以看到在AQS中tryAcquire方法直接抛出了异常。由于具体的获取资源细节需要子类根据自己要实现的功能来写，AQS只负责阻塞队列的管理等工作。同时注意到这个方法并不是一个笼统的方法。其实前面说的需要子类实现的5个方法都不是笼统的，由于子类并不肯定需要实现所有这些方法，这提供了肯定的灵活性。

2.1.1 addWaiter方法详解

先忙接着看acquire方法。在if语句里，假如tryAcquire返回true，那么acquire就返回了，说明成功获取到了资源。假如tryAcquire返回false，if语句的前半句判断就成立了，需要继续执行&&右边的acquireQueued方法，执行它之前先执行了addWaiter。先看一下addWaiter它的代码：

    private Node addWaiter(Node mode) {        Node node = new Node(Thread.currentThread(), mode);        // Try the fast path of enq; backup to full enq on failure        Node pred = tail;        if (pred != null) {            node.prev = pred;            if (compareAndSetTail(pred, node)) {                pred.next = node;                return node;            }        }        enq(node);        return node;    }

可以看到这个方法有一个Node参数。Node类便是aqs维护的FIFO队列中的元素的类型。回顾一下acquire()方法的代码，是将Node.EXCLUSIVE作为参数传入了addWaiter。查看Node源码发现有这么一句：

        /** Marker to indicate a node is waiting in exclusive mode */        static final Node EXCLUSIVE = null;

原来这是一个null值，用来表示独占性线程。不论如何，先继续看addWaiter的源码吧。
第一句代码：Node node = new Node(Thread.currentThread,mode);新建了一个表示当前线程的结点。刚才传入的null作为模式传给构造方法。进入对象构造方法查看：

        Node(Thread thread, Node mode) {     // Used by addWaiter            this.nextWaiter = mode;            this.thread = thread;        }

继续看addWaiter的后续代码，发现是获取了当前队列的尾节点，并将新建结点的prev指针执行尾节点，再使用cas尝试替换尾节点，假如成功，那么当前结点就成为新的尾节点，返回。
假如cas失败或者者当前tail为null，调用eng方法解决。下面看一下eng的代码：

    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;                }            }        }    }

熟习AtomicInteger的朋友看到这段代码肯定会感动非常熟习。这里就是使用了循环尝试的方式来进行cas操作，指导成功为止。另外当tail==null时，先新建head结点再进行操作，当前这里给head变量反之也是使用了cas操作。
综上，addWaiter()进行的操作就是安全地升级队列的tail指针。

2.1.2 acquireQueue方法详解

下面继续看acquire()方法。再把代码贴一次。

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

给acqureQueued方法传入的第一个参数是addWaiter方法的返回值，回想一下刚才的addWaiter方法，发现它的返回值是新创立的表示当前线程的Node结点。acquireQueued方法的另一个参数是acquire的形参arg，这个一般是获取资源的个数，像ReentrantLock的lock方法就是调用了acquire(1)。下面看一下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)) {                    setHead(node);                    p.next = null; // help GC                    failed = false;                    return interrupted;                }                if (shouldParkAfterFailedAcquire(p, node) &&                    parkAndCheckInterrupt())                    interrupted = true;            }        } finally {            if (failed)                cancelAcquire(node);        }    }

这个方法的主体是一个死循环，不断测试两件事：1.能否是头结点的下一个节点，说明该轮到自己获取资源了。2：能否可以休息了。判断1成功后就用tryAcquire获取资源，成功后设置当前结点为头结点，返回。假如1判断不成功则执行shouldParkAfterFailedAcquire方法，先贴一下代码：

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {        int ws = pred.waitStatus;        if (ws == Node.SIGNAL)            /*             * This node has already set status asking a release             * to signal it, so it can safely park.             */            return true;        if (ws > 0) {            /*             * Predecessor was cancelled. Skip over predecessors and             * indicate retry.             */            do {                node.prev = pred = pred.prev;            } while (pred.waitStatus > 0);            pred.next = node;        } else {            /*             * waitStatus must be 0 or PROPAGATE.  Indicate that we             * need a signal, but don't park yet.  Caller will need to             * retry to make sure it cannot acquire before parking.             */            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);        }        return false;    }

这个方法的工作是找到当前结点之前的一个未取消的结点，将其waitStatue改为SIGNAL（-1）。这样在该结点释放资源时就会唤醒当前结点。
当shouldParkAfterFailedAcquire返回true之后，当前线程即可以去休息了——调用parkAndCheckInterrupt方法：

    private final boolean parkAndCheckInterrupt() {        LockSupport.park(this);        return Thread.interrupted();    }

这个方法使用了LockSupport的park方法，使线程进入waiting状态。当其它线程调用unPark方法，或者此线程被中断后才会返回。

2.1.3小结

下面来总结一下acquire方法。

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

先尝试获取资源，获取到的情况直接返回。获取不到将线程加入队列：首先将tail指向表示当前线程的结点，使用CAS操作升级tail。之后执行acquireQueued方法，假如是当前队列的第二个则再次尝试获取tryAcquire，成功后将自定义为head（head表示已经获取到的资源的结点）。不能获取资源时判断能否可以park()，判断依据是其prev的结点的waitState能否是signal，即能否会在释放资源时通知它。之后当前线程调用park进入waiting状态。waitting结束时返回能否中断标志，并重置标志。回到acquire，假如waitting期间中断过，则调用selfInterrupt响应中断。

2.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;}

这里可以看出释放资源成功时，获取到head结点（由于head结点表示的线程就是当前获取到资源的线程），执行unparkSuccessor()操作。这里便和shouldParkAfterFailedAcquire中‘休息’的代码相呼应。假如那里设置了waitStatus为signal就会使用LockSupport.unpark方法来唤醒等待的线程。

    private void unparkSuccessor(Node node) {        /*         * If status is negative (i.e., possibly needing signal) try         * to clear in anticipation of signalling.  It is OK if this         * fails or if status is changed by waiting thread.         */        int ws = node.waitStatus;        if (ws < 0)            compareAndSetWaitStatus(node, ws, 0);        /*         * 假如后继结点为null或者等待状态>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);    }

对了，tryRelease方法也是具体的同步器来实现的。

2.3 其它方法

acquireShared(int)和releaseShared()方法是共享模式下获取资源和释放资源的方法。这里不再详细开展了，请看参考资料里的文章。

参考资料：1.Java并发之AQS详解