JUC代码浅析[1]——同步器AQS

 

JUC代码浅析[1]——同步器AQS

       AQS（AbstractQueuedSynchronizer）是一个提供实现各种锁和同步器的基本框架，它实现了调度逻辑，留出具体的进入和释放规则给子类实现。JUC中基于AQS实现的有ReentrantLock,Semaphore,CountDownLatch, ReetrantRead WriteLock,FutureTask等。代码非常复杂很难描述清楚每个细节，可能还有一些理解偏差的地方。总的来说底层实现的要点有Unsafe.compareAndSwapXXX(Object o,long offset,int expected,int x);Unsafe.park() 挂起线程 ;Unsafe.unpark()唤醒被park的线程;双向列表存储线程的队列。

       被park的线程有3种方式可以唤醒它，park返回时不会报告是被哪种方式唤醒的，可以用Thread.interrupted()检查，

l  unpark(Thread thread),这种方式唤醒时Thread.interrupted()返回的是false

l  thread.interrupt()，这种方式唤醒时Thread.interrupted()返回的是true

l  不知原因的返回

 

       队列中存放着Node代表等待线程,它包含4中状态CANCELLED(1):当前节点已经退出了，SIGNAL(-1)：后继者需要被唤醒,CONDITION(-2)：节点在condition队列中等待，0；非负数代表节点不需要被唤醒了。节点有共享和互斥两种模式。

AQS提供给子类一个int state属性通过getState()和setState()暴露给子类，子类通过state和实现以下5个方法来实现不同的需求

l  tryAcquire  互斥模式下尝试获得许可（一般需要结合state的值）

l  tryRelease 互斥模式下尝试释放许可并设置state

l  tryAcquireShared  共享模式下尝试获得许可

l  tryReleaseShared 共享模式下尝试释放许可并设置state

l  isHeldExclusively 当前线程是否占有许可

       外部获取许可总的来说就是尝试获取许可，不成功时进入列表队列等待再次尝试或挂起或被打断。具体实现看互斥不可中断模式的acquire方法，

public final void acquire(int arg) {

        if (!tryAcquire(arg) &&

            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

            selfInterrupt();

    }

       首先尝试获取许可(也就是子类要实现的),不成功时当前线程添加到列表的最后并等待，必要时挂起或打断线程(注意何时挂起何时打断)。

其中addWaiter方法主要使用自旋锁和Unsafe.compareAndSwapXXX方法保证列表状态的一致.

       acquireQueued方法不断的检查和移动队列，当当前线程到达位于队列头下一个的时候尝试获取

    final boolean acquireQueued(final Node node, int arg) {

        try {

            boolean interrupted = false;

            for (;;) {

                final Node p = node.predecessor();

                if (p == head && tryAcquire(arg)) {

                    setHead(node);

                    p.next = null; // help GC

                    return interrupted;

                }

                if (shouldParkAfterFailedAcquire(p, node) &&

                    parkAndCheckInterrupt())

                    interrupted = true;

            }

        } catch (RuntimeException ex) {

            cancelAcquire(node);

            throw ex;

        }

    }

 

下面看如何检查和移动队列的

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

        int s = pred.waitStatus;

        if (s < 0)

            /*

             * 前一个节点的等待状态小于0，则当前节点被park起来

             *

             */

            return true;

        if (s > 0) {

            /*

             * 前面节点是退出状态，则调整指针往前查询跳过其他退出状态的节点，

             * 直到找到不是退出状态的节点作为它的前一个节点

             */

        do {

       node.prev = pred = pred.prev;

        } while (pred.waitStatus > 0);

        pred.next = node;

    }

        else

            /*

             * 标识当前节点需要被signal(以前一个节点状态为SIGNAL表示)，以免期间发生        * 了请求操作需要继续自旋检查

             */

            compareAndSetWaitStatus(pred, 0, Node.SIGNAL);

        return false;

    }

 

释放许可。互斥不可中断模式的release方法，

    public final boolean release(int arg) {

        if (tryRelease(arg)) {

            Node h = head;

            if (h != null && h.waitStatus != 0)

                unparkSuccessor(h);

            return true;

        }

        return false;

    }

尝试释放成功时，唤醒列表队列的下一个节点(如果需要的话)

    private void unparkSuccessor(Node node) {

        /*

         * 清除节点的SIGNAL状态 （接下来的唤醒操作之后，表示它的下一个节点不需要被唤        * 醒了）

         */

        compareAndSetWaitStatus(node, Node.SIGNAL, 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);

    }

 

       另外还有互斥可中断模式、共享不可中断模式、共享可中断模式。

       可中断和不可中断模式的区别在于可中断模式接收到中断信号后立即中断，不可中断模式要等待节点到达列表第一个位置才中断(中断过程还要进行一些操作)。

       共享模式尝试获取是执行tryAcquireShared方法并把节点标识为共享模式，其他的操作跟互斥模式有所不同但大同小异。具体的就不详细展开了。