AQS原理及源码浅析

作者: silence_J | 来源:发表于2020-05-14 17:49 被阅读0次

什么是AQS？

AQS 即 AbstractQueuedSynchronizer 抽象队列同步器，它是Java并发用来构建锁和其他同步工具的基础框架，是一个抽象类。AQS定义了一套多线程访问共享资源的同步器框架，许多同步类实现都依赖于它，如常用的ReentrantLock、Semaphore、CountDownLatch。

ReentrantLock是如何加锁的？AQS是如何被使用的？

ReentrantLock 通过调用对象的lock()和unlock()方法来加锁解锁，下面分析一下源码中加锁的过程，在加锁过程中使用了AQS类。

lock方法的内部调用关系如图：

Lock调用关系图.png
首先看一下默认使用非公平锁的时候的调用过程：

ReentrantLock调用lock()方法的时候，它会调用Sync的acquire(1)方法；
因为 Sync 是 ReentrantLock 内抽象的静态内部类，它继承了AQS抽象类，Sync 内又没重写 acquire()，所以直接调用的是AQS的 acquire(1)；
AQS的 acquire(1)，会调AQS内的 tryAcquire(1)；
AQS内的 tryAcquire(1) 方法内只是抛出了个异常，实际上调的是 ReentrantLock 中 NonfairSync 类重写的 tryAcquire(1)；如果采用的是公平锁的话，这里会调用 FairSync类(继承Sync) 重写的 tryAcquire 方法
NonfairSync 继承了 Sync，tryAcquire(1) 调用的是父类 Sync 的nonfairTryAcquire(1)。（FairSync 直接实现了 tryAcquire 方法，无其它调用）

至此就是整个调用过程，下是源码中的Sync和它的子类：

ReentrantLock部分源码.png

AQS原理及源码解读

上面读到nonfairTryAcquire(acquires)，就必须要了解实现了。下面先分析一下AQS的原理。

AQS队列结构浅析

AQS内部维护着一个 volatile int state 和一个FIFO（双向队列）线程等待队列（多线程争用资源被阻塞时会进入此队列），即CLH队列。这两个是AQS的核心。

注：CLH 是论文《Building FIFO and Priority-Queuing Spin Locks from Atomic Swap》三位作者的缩写，AQS就是根据 CLH 锁的变种实现的

AQS的同步机制就是依靠CLH队列实现的。state所代表的意义由子类来定，它的值是根据子类不同的实现取不同的意义。拿ReentrantLock来说，state初始为0，当线程得到了这把锁后state会变为1，重入一次会+1，释放一次-1，减到0完全释放，所以在ReentrantLock中state为0和1就代表了加锁和解锁。

双向队列由链表结构实现，AQS内部定义了

     /**
     * Wait queue node class.
     */
    static final class Node {
       
        volatile int waitStatus;

        volatile Node prev;

        volatile Node next;

        // Node中保存的是一个个线程
        volatile Thread thread;

        Node nextWaiter;

        // 只截取了部分源码，完整源码请参照jdk11
    }

    private transient volatile Node head;

    private transient volatile Node tail;

    private volatile int state;

CLH队列是FIFO的双端双向队列。线程通过AQS获取锁失败，就会将线程封装成一个Node节点，插入队列尾。当有线程释放锁时，会尝试把
head.next节点的线程唤醒（LockSupport的unpark方法）。被唤醒的线程获得锁后node节点随之出队。

AQS同步队列.png

当线程acquire(1)以后，如果state的值是0，那就直接拿到锁，默认是非公平的上来就抢，抢不着就进队列里acquireQueued()。

来看一下源码：

    // AQS类方法
    public final void acquire(int arg) {
        // 先tryAcquire尝试获取锁，若返回true表示获取成功，
        // 由于&&符号逻辑短路，符号后部分不会执行
        if (!tryAcquire(arg) &&
            // 若获取失败，线程排队等待
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

tryAcquire失败会调用 acquireQueued() 方法：

final boolean acquireQueued(final Node node, int arg) {
        boolean interrupted = false;
        try {
            // 节点进入等待队列，执行死循环自旋
            for (;;) {
                // 拿到前一个节点
                final Node p = node.predecessor();
                // 如果前一个节点是head，并且当前线程成功获得了锁
                if (p == head && tryAcquire(arg)) {
                    // 把当前节点设置为头节点（head指针指向当前节点）
                    // 并把thread和prev置空（因为现在已经获取到锁，不再需要记录这个节点所对应的线程了）
                    setHead(node);
                    // 把旧的head的next置空，使旧的head出队
                    p.next = null; // help GC
                    return interrupted;
                }
                // 判定前一个节点的状态是否为 Node.SIGNAL
                if (shouldParkAfterFailedAcquire(p, node))
                    interrupted |= parkAndCheckInterrupt();
            }
        } catch (Throwable t) {
            cancelAcquire(node);
            if (interrupted)
                selfInterrupt();
            throw t;
        }
    }

这是一个死循环，除非进入if(p == head && tryAcquire(arg))判定条件，获取node节点的前一个节点p。当p为头节点并且当前线程tryAcquire成功时，用setHead(node)把当前节点设置为头节点，同时把节点属性thread、prev置空。然后再把p的next节点置空使使旧的head出队。
总结起来就是把旧的头部断开，head指针指向一个新的头部，这个新head内thread属性也是空的，因为现在已经获取到锁，不再需要记录这个节点所对应的线程了。

图解如下：

acquireQueued示意图.png

如果上面判定条件失败
会首先判定：shouldParkAfterFailedAcquire(p , node)，这个方法内部会判定前一个节点的状态是否为：Node.SIGNAL，若是则返回true，不是返回false，不过会再做一些操作：判定节点的状态是否大于0，若大于0则认为被 CANCELLED 掉了（大于0的只可能被CANCELLED的状态），因此会从前一个节点开始逐步循环找到一个没有被 CANCELLED 节点，然后与这个节点的next、prev的引用相互指向；如果前一个节点的状态不是大于0的，则通过CAS尝试将状态修改为“Node.SIGNAL”，自然的如果下一轮循环的时候会返回值应该会返回true。
如果这个方法返回了true，则会执行：parkAndCheckInterrupt()方法，它是通过LockSupport.park(this)将当前线程挂起到WATING状态，它需要等待一个中断、unpark方法来唤醒它，通过这样一种FIFO的机制的等待，来实现了Lock的操作。

AQS队列初始化

程序第一个线程直接抢到锁后并不会入队，那队列是什么时候初始化的呢？

AQS队列初始化.png

    public final void acquire(int arg) {
        // 先tryAcquire尝试获取锁，若返回true表示获取成功，
        // 由于&&符号逻辑短路，符号后部分不会执行
        if (!tryAcquire(arg) &&
            // 若获取失败，线程排队等待
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

可以看到在第二个线程获取锁失败入队时会调用 addWaiter(Node.EXCLUSIVE)，默认EXCLUSIVE排它锁，下面再来看一下这个方法：

    private Node addWaiter(Node mode) {
        Node node = new Node(mode);
        // 死循环，直到执行完毕
        for (;;) {
            Node oldTail = tail;
            // 第一次入队未初始化时tail为空
            if (oldTail != null) {
                // 把新节点的前置节点设置在等待队列的末端（prev指针指向oldTail）
                node.setPrevRelaxed(oldTail);
                // cas操作，把新节点设置为tail末端
                if (compareAndSetTail(oldTail, node)) {
                    oldTail.next = node;
                    return node;
                }
            } else {
                // tail为空则初始化队列
                initializeSyncQueue();
            }
        }
    }

可以看到在addWaiter方法中进行了初始化。

如何使用AQS获取锁？tryAcquire()

ReentrantLock 可以实现公平锁和非公平锁，在NonfairSync、FairSync静态内部类中分别重写了tryAcquire()

    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }

看一下非公平时如何获取锁：
NonfairSync 中 tryAcquire() 调用了父类Sync 的 nonfairTryAcquire(acquires)

    // jdk11 ReentrantLock 部分源码
    final boolean nonfairTryAcquire(int acquires) {
        // 获取当前线程
        final Thread current = Thread.currentThread();
        // 获取state状态
        int c = getState();
        // 如果状态值为0
        if (c == 0) {
            // 尝试用cas操作将0变成1，获取锁
            if (compareAndSetState(0, acquires)) {
                // cas操作成功说明抢到锁，把这个线程设置为独占这把锁
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        // 如果这个线程是 独占同步锁的线程
        else if (current == getExclusiveOwnerThread()) {
            // 改变状态值，指锁重入机制
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

先得到当前线程，然后获取state的值，如果state的值等于0，用compareAndSetState(0,acquire)方法尝试把state的值改为1，假如改成了setExclusiveOwnerThread()把当前线程设置为独占statie这个把锁的状态，说明我已经得到这把锁，而且这把锁是互斥的，我得到以后，别人是得不到的，因为别人再来的时候这个state的值已经变成1了，如果说当前线程已经是独占state这把锁了，就往后加个1就表示可重入了。

公平时如何获取锁：
FairSync自己直接重写了tryAcquire。源码跟nonfairTryAcquire差不多，
区别是它会先查询是否已经有线程在队列中等待

protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 先判断队列中是否有等待线程
        if (!hasQueuedPredecessors() &&
            // 没有等待线程的话进行cas操作尝试加锁，改变state
            compareAndSetState(0, acquires)) {
            // 设置当前线程独占这把锁
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    // 下面代码跟nonfairTryAcquire的一样
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

如何释放锁，并唤醒队列中其它线程？

ReentrantLock 的 unlock() 被调用时会释放锁

public void unlock() {
    sync.release(1);
}

它实际调用的AQS的release：

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            // 唤醒head的next节点
            // 调用的LockSupport.unpark()
            unparkSuccessor(h);
        return true;
    }
    return false;
}

protected final boolean tryRelease(int releases) {
    // 改变状态
    int c = getState() - releases;
    // 当前线程不是独占锁的线程时，抛异常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    // 定义是否被释放的标记
    boolean free = false;
    // 判断state状态是否为0
    if (c == 0) {
        // 状态为0表示锁被释放，标记设为true
        free = true;
        // 把AQS中独占锁的线程设置为空，表示当前没有任何线程占用锁
        setExclusiveOwnerThread(null);
    }
    // 更新状态信息
    setState(c);
    return free;
}

AQS中的细节：VarHandle存在的意义

上面讲AQS队列初始化时看的 addWaiter 源码中调用了 node.setPrevRelaxed(oldTail) ，点进去看一下：

final void setPrevRelaxed(Node p) {
    PREV.set(this, p);
}

有这个调用：PREV.set(this, p)，这个PREV是这样定义的：

// AQS的Node类 部分源码 jdk11
private static final VarHandle PREV;
static {
    try {
        // Lookup 对象是一个创建handles的工厂
        MethodHandles.Lookup l = MethodHandles.lookup();
        // 返回某个类的Var Handle对象
        PREV = l.findVarHandle(Node.class, "prev", Node.class);
    } catch (ReflectiveOperationException e) {
        throw new ExceptionInInitializerError(e);
    }
}

它是VarHandle类型，VarHandle是在JDK1.9开始有的。看它的名字，Var 变量，Handle 句柄，意思是方便操纵变量的工具。

举个例子，Object o = new Object(); 引用o指向内存中Object对象，用VarHandle的话它也是指的这个引用。

既然两个引用都指向同一个对象，为什么还需要VarHandle？

下面写个小程序来说明：

public class T01_VarHandle {

    int var = 10;
    private static VarHandle varHandle;

    static {
        // 初始化一个handle工厂
        MethodHandles.Lookup l = MethodHandles.lookup();
        try {
            // 初始化var的handle
            varHandle = l.findVarHandle(T01_VarHandle.class, "var", int.class);
        } catch (NoSuchFieldException e) {
            e.printStackTrace();
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        T01_VarHandle instance = new T01_VarHandle();

        // 通过handle读取变量（读）
        System.out.println(varHandle.get(instance));

        // 通过handle改变变量（写）
        varHandle.set(instance, 20);
        System.out.println(instance.var);

        //进行线程安全的累加操作
        varHandle.getAndAdd(instance, 10);
        System.out.println(instance.var);

        // 通过handle进行CAS操作
        varHandle.compareAndSet(instance, 30, 40);
        System.out.println(instance.var);
    }

}

输出结果为10 20 30 40。

也就是说可以用VarHandle对变量进行读写，这点跟普通引用没太大区别。重要的是VarHandle可以做一些compareAndSet操作(原子操作)，还可以handle.getAndAdd()操作这也是原子操作，这些是线程安全的。

所以VarHandle除了可以完成普通属性的原子操作，还可以完成原子性的线程安全的操作，这也是VarHandle的含义。

AQS效率高的原因

AQS中线程进入等待队列，节点插入队尾时需要保证线程安全，如果对整个队列加锁效率就太低了，所以AQS用VarHandle的特性，采用CAS的方法插入节点，这样效率就会很高。

所以也可以说volatile和CAS是AQS的核心。

AQS队列为什么是双向链表？

要添加一个线程节点的时候，需要看一下前面这个节点的状态，如果前面的节点是持有线程的过程中，这个时候就得在后面等着，如果说前面这个节点已经取消掉了，那就应该越过这个节点，不去考虑它的状态，所以有需要看前面节点状态的时候，必须是双向的。

比如说前面看过的acquireQueued()方法的for死循环中，等待队列中的节点需要关注前面节点的状态，如果前面成head了，就尝试获取锁。这个过程就要求必须是双向链表。

后记

关于AQS其实还有很多可分析的地方，比如读写锁中的排他锁和共享锁是如何实现的，ReentrantLock的newCondition()等。本文暂时就写这些了，其它源码有空再读......

AQS原理及源码浅析

什么是AQS？

ReentrantLock是如何加锁的？AQS是如何被使用的？

AQS原理及源码解读

AQS队列结构浅析

AQS队列初始化

如何使用AQS获取锁？tryAcquire()

如何释放锁，并唤醒队列中其它线程？

AQS中的细节：VarHandle存在的意义

AQS效率高的原因

AQS队列为什么是双向链表？

后记

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读