JAVA AQS源码分析

转自: 　　http://www.cnblogs.com/pfan8/p/5010526.html

JAVA AQS的全称为（AbstractQueuedSynchronizer），用于JAVA多线程的开发，从名称我们也可以看出，它实现了同步的队列，而这个队列是指线程队列。AQS类在java.util.concurrent.locks下面。

AQS和CAS作为JAVA5之后非常重要的特性，能在并发应用中提高程序性能，具体要就实际情况使用，因为JVM也在一直优化synchronized关键字，在JAVA7之后其性能也趋于稳定，不会随着线程数增加而导致性能骤降（具体可以取网上搜索对比数据）。

总之，一般情况下还是建议用synchronized

CAS（CompareAndSet）是最小粒度的操作，保证了原子性，通过硬件指令集实现。简单来说，CAS有3个操作数，内存值V，旧的预期值A，要修改的新值B。当且仅当预期值A和内存值V相同时，将内存值V修改为B，否则返回V。

基于此，我们才能完成非阻塞同步操作（当然还有一些其他原子命令，例如FAI，LL/SC等），目的就是用乐观锁来换取性能的提升。

为什么要说CAS呢？因为AQS也是基于CAS实现的，下面进入正题，我们通过源码来具体分析下AQS的实现：

-----------------------------------------------------------------------------华丽的分隔线---------------------------------------------------------------------------------------

AQS的包结构如下　　　　　　　　　　　　　　　　　　　　　　　　　　　　

继承的子类有

我们从ReentrantLock来分析AQS的实现原理：

ReentrantLock是一个可重入的排他锁。可重入指当前拥有锁的对象可重复进入同步区域，防止重复操作，例如网页登陆时重复点击按钮。排他锁就是指该锁（Lock）是互斥锁，只允许一个对象拥有锁。

先看看ReentrantLock的类结构

可以看到有3个内部类Sync，NonfairSync和FairSync。其中NonfairSync和FairSync均继承Sync，而Sync继承了AQS。NonfairSync和FairSync的区别在于当一个线程释放了锁的时候，队列里的其他线程是否按照FIFO的规则去获取锁的。

换句话说，FairSync能够保证先到的线程先拿到锁（有一个特殊情况，就是队列里的线程在unpark到获取锁的过程中有新的还未加入到队列中的线程获取到锁，不过这种情况发生的概率很小，基本不用考虑），而NonfairSync不保证

下面我们看看ReentrantLock是如何实现上锁的，这是lock函数：

    public void lock() {

        sync.lock();

    }

sync的lock函数为抽象的，由子类实现，这里只给出FairSync的实现

    final void lock() {

        acquire(1);

    }

acquire就是AQS提供的接口

    public final void acquire(int arg) {

        if (!tryAcquire(arg) &&

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

            selfInterrupt();

    }

这里涉及到4个新方法，让我们慢慢来分析：

首先，acquire有一个参数arg，是用于判断锁持有的次数，也就是重入的次数，当锁持有者需要释放锁的时候，则要将锁的state减去arg的值。在上面可以看到，调用acquire时传入的参数为1。

1. tryAcquire：AQS里tryAcquire的实现如下

    protected boolean tryAcquire(int arg) {

        throw new UnsupportedOperationException();

    }

明显是不对的，只抛出了一个异常，所以应该是子类覆盖了，那么看看FairSync的实现

    protected final boolean tryAcquire(int acquires) {

        final Thread current = Thread.currentThread();

        int c = getState();

        if (c == 0) {

            if (!hasQueuedPredecessors() &&

                compareAndSetState(0, acquires)) {

                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;

    }

getState()是从ReentrantLock获取的状态，表示锁当前是否被持有，为0时表示没有线程持有锁。此时当前线程会去争取锁的持有权。

首先判断队列中是否有排在当前线程之前的线程，有的话放弃争抢锁。hasQueuedPredecessors是AQS里的方法：

    public final boolean hasQueuedPredecessors() {

        // The correctness of this depends on head being initialized

        // before tail and on head.next being accurate if the current

        // thread is first in queue.

        Node t = tail; // Read fields in reverse initialization order

        Node h = head;

        Node s;

        return h != t &&

            ((s = h.next) == null || s.thread != Thread.currentThread());

    }

tail记录队列里的尾节点，head就是头结点。这个函数就是判断当前线程是否是下一个可持锁的线程

然后，当当前线程满足条件时，就通过CAS设置c的值，同时通过AbstractOwnableSynchronizer类提供的setExclusiveOwnerThread接口将当前线程锁住，用于防止volatile字段被其他线程修改（这里是看注释后的个人理解）

如果c != 0，同时锁持有者为当前线程，那么这个请求就是重入请求了，将c+=acquire。如果上诉两个条件都不满足，返回false。

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

    }

根据注释可以知道，首先快速插入队列，其实就是判断队列是否为空，否则就要通过enq(node)将当前节点添加至队列中，为什么这样会慢些？我们看看enq函数

    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;

                }

            }

        }

    }

这个方法可以好好看看：用for无限循环，考虑下多线程同时加入队列，通过CAS保证设置当前节点到尾节点。这里可能队列里的线程已经执行完任务释放锁了，所以还需要重新判断队列是否为空，因此其执行效率当然会有影响。

否则，设置前置节点，加入队列。用图例来帮助理解：

由于本身没有锁，可以有多个线程进来，如果有多个线程并发进入这个if判定区域，可能就会同时存在多个这样的数据结构，在各自形成数据结构后，多个线程都会去做compareAndSetHead(h)的动作，也就是尝试将这个临时h节点设置为head，

显然并发时只有一个线程会成功，因此成功的那个线程会执行tail = node的操作，整个AQS的链表就成为：

3. 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);

        }

    }

同样是for无限循环，判断当前节点之前是否没有线程节点了，如果是，就去争抢锁。用上面给出的子类tryAcquire函数，成功的话设置相关参数，这里也解释了释放指针，帮助垃圾回收（GC）。

如果争抢失败，判断是否需要阻塞当前线程

    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;

    }

waitState参数有以下几个：

SIGNAL：当前线程持有锁
CANCELLED：超时获取被中断
CONDITION
PROPAGATE
0：不属于以上任何一种情况

后面两个没怎么看，大致就是需要满足多个条件，PROPAGATE用于shared锁队列。

回到上面的方法：因为AQS支持中断等待，所以如果线程中断了争抢锁（CANCELLED），那么就不需要阻塞，直接返回。acquireQueued方法没有返回，而是设置一个interrupt参数为true而已，线程争抢锁失败的话继续休眠等待，而AQS里的doAcquireInterruptibly()发现争抢失败的话就直接throw new InterruptedException()。在ReentrantLock里需要调用ReentrantLock.lockInterruptibly()就会实现中断返回。否则AQS会尝试将当前线程状态设置成SIGNAL，失败就循环继续尝试

下面是parkAndCheckInterrupt

    private final boolean parkAndCheckInterrupt() {

        LockSupport.park(this);

        return Thread.interrupted();

    }

LockSupport.park通过Unsafe.park实现阻塞，它会设置一个AQS的blocker，让队列里的线程阻塞在一个地方，然后返回线程中断的判断

4. selfInterrupt

    private static void selfInterrupt() {

        Thread.currentThread().interrupt();

    }

acquireQueued会返回boolean值表示线程是否中断，如果未中断，就调用Thread.interrupt()中断线程

---------------------------------------------------------------------------------------------------------------------

以上是锁的实现原理，当tryAcquire()成功之后，线程获取锁，执行任务，执行完毕之后，会调用AQS的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;

    }

tryRelease是子类Sync的实现：

    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;

    }

其他就不多说了，比较简单明晰。就看看unparkSuccessor怎么做的

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

        /*

         * Thread to unpark is held in successor, which is normally

         * just the next node.  But if cancelled or apparently null,

         * traverse backwards from tail to find the actual

         * non-cancelled successor.

         */

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

    }

通过CAS修改节点的waitStatus，然后将后续节点去除，这里会去遍历后续节点，判断其是否状态为CANCELLED，将所有非CANCELLED的节点唤醒。

巴特西

JAVA AQS源码分析

最新文章

热门文章