Java并发之AQS原理

作者: 第四单元 | 来源:发表于2018-12-21 21:46 被阅读1次

    本文是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详解

    相关文章

      网友评论

        本文标题:Java并发之AQS原理

        本文链接:https://www.haomeiwen.com/subject/nadvkqtx.html