AbstractQueuedSynchronizer源码分析

2017-06-30

1. AQS介绍

AQS全名是AbstractQueuedSynchronizer，是JUC（java并发包）的一个同步工具类，是很多并发类的基础，比如ReentrantLock，CountDownLatch，ReentrantReadWriteLock等都是继承AQS，然后再自己实现相应的tryAcquire和tryRelease方法的，所以了解AQS的原理以及结构有助于了解其他并发类。

2. AQS和Synchronized

3. 设计AQS的思路

AQS也可以当作是一个同步器，通过同步器来实现锁的功能，控制线程获取资源以及挂起或唤醒线程，从最基础角度来说，AQS定义了如何获取以及释放锁，而获取以及释放锁的操作离不开三个基本点

原子操作：CAS

等待对列：CLH Lock queue（实现自旋锁，不用挂起到OS，节省资源），通过Node来实现

挂起或唤醒：LockSupport.park(),LockSupport.unpark();

4. AQS的类结构

state：用volatile修饰，保证可见性，表示当前锁的状态
1.1 在ReentrantLock中，state是已经重入（同一线程多次获取锁）的次数
1.2 在CountDownLatch中，state是锁存器的计数，也就是需要都少次countDown

Node:实现等待队列，将未能获取到锁的线程通过队列形式存储起来

ConditionObject：实现Condition的功能（等待，唤醒）

5. AQS的两种模式：独占和共享

acquire()的实现

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&//尝试获取锁，获取得到的话， &&后则不执行
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//获取不到时添加到等待队列并挂起等待的线程
            selfInterrupt();
    }

tryAcquire()是由子类实现，当前讨论ReentrantLock中的FairSync.tryAcquire()

可以画图表示逻辑关系

protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();//得到当前锁的状态，得到AQS中的state变量
            if (c == 0) {//当前锁未被使用
                if (!hasQueuedPredecessors() &&//当前等待队列为空：在非公平锁（NonFairSync）中则没有这行代码，这是公平锁和非公平锁的一个区别
                compareAndSetState(0, acquires)) {//通过CAS获取锁
                    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;
        }

addWaiter()

/**
     * Creates and enqueues node for current thread and given mode.
     *
     * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
     Node.EXCLUSIVE:独占模式
     Node.SHARED:共享模式
     * @return the new node
     */
    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;
        //tail！=null，则有等待队列，尝试添加到尾部
        if (pred != null) {
            node.prev = pred;//设置尾节点与新添加节点的关系
            if (compareAndSetTail(pred, node)) {//CAS设置尾节点，失败则enq(node)
                pred.next = node;
                return node;
            }
        }
        //tail==null时，则等待队列还没形成
        enq(node);//存在并发，自旋形式修改
        return 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;
                }
            }
        }
    }

acquireQueued()

final boolean acquireQueued(final Node node, int arg) {//通过addWaiter方法返回的node是尾节点
        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);
        }
    }