AbstractQueuedSynchronizer超详细原理解析

时间 2019-11-16

标签 abstractqueuedsynchronizer 详细原理解析繁體版

原文原文链接

今天咱们来研究学习一下AbstractQueuedSynchronizer类的相关原理，java.util.concurrent包中不少类都依赖于这个类所提供队列式同步器，好比说经常使用的ReentranLock，Semaphore和CountDownLatch等。前端

为了方便理解，咱们以一段使用ReentranLock的代码为例，讲解ReentranLock每一个方法中有关AQS的使用。java

ReentranLock示例

咱们都知道ReentranLock的加锁行为和Synchronized相似，都是可重入的锁，可是两者的实现方式确实彻底不一样的，咱们以后也会讲解Synchronized的原理。除此以外，Synchronized的阻塞没法被中断，而ReentrantLock则提供了可中断的阻塞。下面的代码是ReentranLock的函数，咱们就以此为顺序，依次讲解这些函数背后的实现原理。node

ReentrantLock lock = new ReentrantLock();
lock.lock();
lock.unlock();
复制代码

公平锁和非公平锁

ReentranLock分为公平锁和非公平锁，两者的区别就在获取锁机会是否和排队顺序相关。咱们都知道，若是锁被另外一个线程持有，那么申请锁的其余线程会被挂起等待，加入等待队列。理论上，先调用lock函数被挂起等待的线程应该排在等待队列的前端，后调用的就排在后边。若是此时，锁被释放，须要通知等待线程再次尝试获取锁，公平锁会让最早进入队列的线程得到锁。而非公平锁则会唤醒全部线程，让它们再次尝试获取锁，因此可能会致使后来的线程先得到了锁，则就是非公平。算法

public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}
复制代码

咱们会发现FairSync和NonfairSync都继承了Sync类，而Sync的父类就是AbstractQueuedSynchronizer(后续简称AQS)。可是AQS的构造函数是空的,并无任何操做。bash

以后的源码分析，若是没有特别说明，就是指公平锁。多线程

lock操做

ReentranLock的lock函数以下所示，直接调用了sync的lock函数。也就是调用了FairSync的lock函数。函数

//ReentranLock
    public void lock() {
        sync.lock();
    }
    //FairSync
    final void lock() {
        //调用了AQS的acquire函数,这是关键函数之一
        acquire(1);
    }
复制代码

咱们接下来就正式开始AQS相关的源码分析了，acquire函数的做用是获取同一时间段内只能被一个线程获取的量，这个量就是抽象化的锁概念。咱们先分析代码，你慢慢就会明白其中的含义。工具

public final void acquire(int arg) {
	// tryAcquire先尝试获取"锁",获取了就不进入后续流程
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        //addWaiter是给当前线程建立一个节点,并将其加入等待队列
        //acquireQueued是当线程已经加入等待队列以后继续尝试获取锁.
        selfInterrupt();
}
复制代码

tryAcquire，addWaiter和acquireQueued都是十分重要的函数，咱们接下来依次学习一下这些函数，理解它们的做用。源码分析

//AQS类中的变量.
private volatile int state;
//这是FairSync的实现,AQS中未实现,子类按照本身的须要实现该函数
protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    //获取AQS中的state变量,表明抽象概念的锁.
    int c = getState();
    if (c == 0) { //值为0,那么当前独占性变量还未被线程占有
        //若是当前阻塞队列上没有先来的线程在等待,UnfairSync这里的实现就不一致
        if (!hasQueuedPredecessors() && 
            compareAndSetState(0, acquires)) {
            //成功cas,那么表明当前线程得到该变量的全部权,也就是说成功得到锁
            setExclusiveOwnerThread(current);
            // setExclusiveOwnerThread将本线程设置为独占性变量全部者线程
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        //若是该线程已经获取了独占性变量的全部权,那么根据重入性
        //原理,将state值进行加1,表示屡次lock
        //因为已经得到锁,该段代码只会被一个线程同时执行,因此不须要
        //进行任何并行处理
        int nextc = c + acquires;
        if (nextc < 0)
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    //上述状况都不符合,说明获取锁失败
    return false;
}
复制代码

由上述代码咱们能够发现，tryAcquire就是尝试获取那个线程独占的变量state。state的值表示其状态：若是是0，那么当前尚未线程独占此变量；否在就是已经有线程独占了这个变量，也就是表明已经有线程得到了锁。可是这个时候要再进行一次判断，看是不是当前线程本身得到的这个锁，若是是，就增长state的值。学习

这里有几点须要说明一下，首先是compareAndSetState函数，这是使用CAS操做来设置state的值，并且state值设置了volatile修饰符，经过这两点来确保修改state的值不会出现多线程问题。而后是公平锁和非公平锁的区别问题，在UnfairSync的nonfairTryAcquire函数中不会在相同的位置上调用hasQueuedPredecessors来判断当前是否已经有线程在排队等待得到锁。

若是tryAcquire返回true，那么就是获取锁成功；若是返回false，那么就是未得到锁，须要加入阻塞等待队列。咱们下面就来看一下addWaiter的相关操做。

等待锁的阻塞队列

将保存当前线程信息的节点加入到等待队列的相关函数中涉及到了无锁队列的相关算法，因为在AQS中只是将节点添加到队尾，使用到的无锁算法也相对简单。真正的无锁队列的算法咱们等到分析ConcurrentSkippedListMap时在进行讲解。

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    //先使用快速入列法来尝试一下,若是失败,则进行更加完备的入列算法.
    //只有在必要的状况下才会使用更加复杂耗时的算法，也就是乐观的态度
    Node pred = tail; //列尾指针
    if (pred != null) {
        node.prev = pred; //步骤1:该节点的前趋指针指向tail
        if (compareAndSetTail(pred, node)){ //步骤二:cas将尾指针指向该节点
            pred.next = node;//步骤三:若是成果,让旧列尾节点的next指针指向该节点.
            return node;
        }
    }
    //cas失败,或在pred == null时调用enq
    enq(node);
    return node;
}
private Node enq(final Node node) {
    for (;;) { //cas无锁算法的标准for循环,不停的尝试
        Node t = tail;
        if (t == null) { //初始化
            if (compareAndSetHead(new Node())) 
              //须要注意的是head是一个哨兵的做用,并不表明某个要获取锁的线程节点
                tail = head;
        } else {
            //和addWaiter中一致,不过有了外侧的无限循环,不停的尝试,自旋锁
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
复制代码

经过调用addWaiter函数，AQS将当前线程加入到了等待队列，可是尚未阻塞当前线程的执行，接下来咱们就来分析一下acquireQueued函数。

等待队列节点的操做

因为进入阻塞状态的操做会下降执行效率，因此，AQS会尽力避免试图获取独占性变量的线程进入阻塞状态。因此，当线程加入等待队列以后，acquireQueued会执行一个for循环，每次都判断当前节点是否应该得到这个变量(在队首了)。若是不该该获取或在再次尝试获取失败，那么就调用shouldParkAfterFailedAcquire判断是否应该进入阻塞状态。若是当前节点以前的节点已经进入阻塞状态了，那么就能够断定当前节点不可能获取到锁，为了防止CPU不停的执行for循环，消耗CPU资源，调用parkAndCheckInterrupt函数来进入阻塞状态。

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) { //一直执行,直到获取锁,返回.
            final Node p = node.predecessor(); 
            //node的前驱是head,就说明,node是将要获取锁的下一个节点.
            if (p == head && tryAcquire(arg)) { //因此再次尝试获取独占性变量
                setHead(node); //若是成果,那么就将本身设置为head
                p.next = null; // help GC
                failed = false;
                return interrupted;
                //此时,尚未进入阻塞状态,因此直接返回false,表示不须要中断调用selfInterrupt函数
            }
            //判断是否要进入阻塞状态.若是`shouldParkAfterFailedAcquire`
            //返回true,表示须要进入阻塞
            //调用parkAndCheckInterrupt；不然表示还能够再次尝试获取锁,继续进行for循环
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                //调用parkAndCheckInterrupt进行阻塞,而后返回是否为中断状态
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL) //前一个节点在等待独占性变量释放的通知,因此,当前节点能够阻塞
        return true;
    if (ws > 0) { //前一个节点处于取消获取独占性变量的状态,因此,能够跳过去
        //返回false
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        //将上一个节点的状态设置为signal,返回false,
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this); //将AQS对象本身传入
    return Thread.interrupted();
}

复制代码

阻塞和中断

由上述分析，咱们知道了AQS经过调用LockSupport的park方法来执行阻塞当前进程的操做。其实，这里的阻塞就是线程再也不执行的含义，经过调用这个函数，线程进入阻塞状态，上述的lock操做也就阻塞了，等待中断或在独占性变量被释放。

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);//设置阻塞对象,用来记录线程被谁阻塞的,用于线程监控和分析工具来定位
    UNSAFE.park(false, 0L);//让当前线程再也不被线程调度,就是当前线程再也不执行.
    setBlocker(t, null);
}
复制代码

关于中断的相关知识，咱们之后再说，就继续沿着AQS的主线，看一下释放独占性变量的相关操做吧。

unlock操做

与lock操做相似，unlock操做调用了AQS的relase方法，参数和调用acquire时同样，都是1。

public final boolean release(int arg) {
    if (tryRelease(arg)) { 
    //释放独占性变量,起始就是将status的值减1,由于acquire时是加1
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);//唤醒head的后继节点
        return true;
    }
    return false;
}
复制代码

由上述代码可知，release就是先调用tryRelease来释放独占性变量。若是成功，那么就看一下是否有等待锁的阻塞线程，若是有，就调用unparkSuccessor来唤醒他们。

protected final boolean tryRelease(int releases) {
    //因为只有一个线程能够得到独占先变量,因此,全部操做不须要考虑多线程
    int c = getState() - releases; 
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) { //若是等于0,那么说明锁应该被释放了,不然表示当前线程有屡次lock操做.
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}
复制代码

咱们能够看到tryRelease中的逻辑也体现了可重入锁的概念，只有等到state的值为0时，才表明锁真正被释放了。因此独占性变量state的值就表明锁的有无。当state=0时，表示锁未被占有，否在表示当前锁已经被占有。

private void unparkSuccessor(Node node) {
    .....
     //通常来讲,须要唤醒的线程就是head的下一个节点,可是若是它获取锁的操做被取消,或在节点为null时
     //就直接继续日后遍历,找到第一个未取消的后继节点.
    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);
}
复制代码

调用了unpark方法后，进行lock操做被阻塞的线程就恢复到运行状态,就会再次执行acquireQueued中的无限for循环中的操做，再次尝试获取锁。

后记

有关AQS和ReentrantLock的分析就差很少结束了。不得不说，我第一次看到AQS的实现时真是震惊，之前都认为Synchronized和ReentrantLock的实现原理是一致的，都是依靠java虚拟机的功能实现的。没有想到还有AQS这样一个背后大Boss在提供帮助啊。学习了这个类的原理，咱们对JUC的不少类的分析就简单了不少。此外，AQS涉及的CAS操做和无锁队列的算法也为咱们学习其余无锁算法提供了基础。知识的海洋是无限的啊！