并发编程的基石——AQS类

GitBook安装部署实操手册

本博客系列是进修并发编程历程当中的纪录总结。由于文章比较多，写的时候也比较散，所以我整理了个目次贴（传送门），轻易查阅。

本文参考了[]和两篇文章。

AQS 简介

AbstractQueuedSynchronizer （简称AQS）类是全部 JUC包的中心类。JUC 中的ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步东西都是基于AQS完成的。

AQS 星散出了构建同步器时的通用关注点，这些关注点重要包括以下：

• 资本是能够被同时接见？照样在同一时候只能被一个线程接见？（同享/独有功用）
• 接见资本的线程怎样举行并发治理？（守候行列）
• 如果线程等不及资本了，怎样从守候行列退出？（超时/中断）

这些关注点都是围绕着资本——同步状况（synchronization state）来睁开的，AQS将这些通用的关注点封装成了一个个模板要领，让子类能够直接运用。

AQS 留给用户的只需两个问题：

• 什么是资本
• 什么情况下资本是能够被接见的

如许一来，定义同步器的难度就大大降低了。用户只需处理好上面两个问题，就可以构建出一个机能优异的同步器。

下面是几个罕见的同步器对资本的定义：

AQS 道理

上面一节中引见到 AQS 笼统出了三个关注点，下面就详细看下 AQS 是如果处理这三个问题的。

同步状况的治理

同步状况，实在就是资本。AQS运用单个int（32位）来保留同步状况，并暴露出getState、setState以及compareAndSetState操纵来读取和更新这个状况。

private volatile int state;
  
protected final int getState() {
    return state;
}

protected final void setState(int newState) {
    state = newState;
}

protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

线程的壅塞和叫醒

在JDK1.5之前，除了内置的监视器机制外，没有别的要领能够平安且便利得壅塞和叫醒当前线程。

JDK1.5今后，java.util.concurrent.locks包供应了类来作为线程壅塞和叫醒的东西。

守候行列

守候行列，是AQS框架的中心，全部框架的症结实在就是怎样在并发状况下治理被壅塞的线程。

守候行列是严厉的FIFO行列，是Craig，Landin和Hagersten锁（CLH锁）的一种变种，采纳双向轮回链表完成，因而也叫CLH行列。

1. 节点定义

CLH行列中的结点是对线程的包装，结点一共有两种范例：独有（EXCLUSIVE）和同享（SHARED）。

每种范例的结点都有一些状况，个中独有结点运用个中的CANCELLED(1)、SIGNAL(-1)、CONDITION(-2)，同享结点运用个中的CANCELLED(1)、SIGNAL(-1)、PROPAGATE(-3)。

AQS运用CLH行列完成线程的构造治理，而CLH构造恰是用前一结点某一属性示意当前结点的状况，之所以这类做是由于在双向链表的构造下，如许更轻易完成作废和超时功用。

next指针：用于保护行列次序，当临界区的资本被开释时，头结点经由历程next指针找到队首结点。

prev指针：用于在结点（线程）被作废时，让当前结点的先驱直接指向当前结点的后驱完成出队行动。

static final class Node {
    
    // 同享形式结点
    static final Node SHARED = new Node();
    
    // 独有形式结点
    static final Node EXCLUSIVE = null;

    static final int CANCELLED =  1;

    static final int SIGNAL    = -1;

    static final int CONDITION = -2;

    static final int PROPAGATE = -3;

    /**
    * INITAL：      0 - 默许，新结点会处于这类状况。
    * CANCELLED：   1 - 作废，示意后续结点被中断或超时，须要移出行列；
    * SIGNAL：      -1- 发信号，示意后续结点被壅塞了；（当前结点在入队后、壅塞前，应确保将其prev结点范例改成SIGNAL，以便prev结点作废或开释时将当前结点叫醒。）
    * CONDITION：   -2- Condition专用，示意当前结点在Condition行列中，由于守候某个前提而被壅塞了；
    * PROPAGATE：   -3- 流传，适用于同享形式。（比方一连的读操纵结点能够顺次进入临界区，设为PROPAGATE有助于完成这类迭代操纵。）
    * 
    * waitStatus示意的是后续结点状况，这是由于AQS中运用CLH行列完成线程的构造治理，而CLH构造恰是用前一结点某一属性示意当前结点的状况，如许更轻易完成作废和超时功用。
    */
    volatile int waitStatus;

    // 先驱指针
    volatile Node prev;

    // 后驱指针
    volatile Node next;

    // 结点所包装的线程
    volatile Thread thread;

    // Condition行列运用，存储condition行列中的后继节点
    Node nextWaiter;

    Node() {
    }

    Node(Thread thread, Node mode) { 
        this.nextWaiter = mode;
        this.thread = thread;
    }
}

2. 行列定义

关于CLH行列，当线程请求资本时，如果请求不到，会将线程包装成结点，将其挂载在行列尾部。
CLH行列的示意图以下：

初始状况，行列head和tail都指向空

首个线程入队，先建立一个空的头结点，然后以自旋的体式格局不停尝试插进去一个包括当前线程的新结点

再次到场新节点，新到场的节点会被安排到行列的尾部。（PS：看下了代码，AQS 的线程治理行列好像是一个双向轮回行列，这边这个图是不是是有点问题？？？）

AQS 的要领引见

用户须要本身重写的要领

上面引见到 AQS 已帮用户处理了同步器定义历程当中的大部份问题，只将下面两个问题丢给用户处理：

• 什么是资本
• 什么情况下资本是能够被接见的

详细的，AQS 是经由历程暴露以下 API 来让用户处理上面的问题的。

如果你须要完成一个本身的同步器，平常情况下只需继承 AQS ，并重写 AQS 中的这个几个要领就好了。至于详细线程守候行列的保护（如猎取资本失利入队/叫醒出队等），AQS已在顶层完成好了。要不怎样说知心呢。

须要注重的是：如果你没在子类中重写这几个要领就直接挪用了，会直接抛出非常。所以，在你挪用这些要领之前必需重写他们。不运用的话能够不重写。

AQS 供应的一系列模板要领

检察 AQS 的源码我们就可以够发明这个类供应了许多要领，看起来让人“头昏眼花”的。然则最重要的两类要领就是猎取资本的要领和开释资本的要领。因而我们捉住重要矛盾就好了：

• public final void acquire(int arg) // 独有形式的猎取资本
• public final boolean release(int arg) // 独有形式的开释资本
• public final void acquireShared(int arg) // 同享形式的猎取资本
• public final boolean releaseShared(int arg) // 同享形式的开释资本

acquire(int)要领

该要领以独有体式格局猎取资本，如果猎取到资本，线程继承往下实行，不然进入守候行列，直到猎取到资本为止，且全部历程疏忽中断的影响。该要领是独有形式下线程猎取同享资本的顶层进口。

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

下面剖析下这个acquire要领的详细实行流程：

step1：起首这个要领挪用了用户本身完成的要领tryAcquire要领尝试猎取资本，如果这个要领返回true，也就是示意猎取资本胜利，那末全部acquire要领就实行完毕了，线程继承往下实行；

step2：如果tryAcquir要领返回false，也就示意尝试猎取资本失利。这时候acquire要领会先挪用addWaiter要领将当前线程封装成Node类并到场一个FIFO的双向行列的尾部。

step3：再看acquireQueued这个症结要领。起首要注重的是这个要领中哪一个无前提的for轮回，这个for轮回申明acquireQueued要领一直在自旋尝试猎取资本。进入for轮回后，起首推断了当前节点的前继节点是不是是头节点，如果是的话就再次尝试猎取资本，猎取资本胜利的话就直接返回false（示意未被中断过）

如果照样没有猎取资本胜利，推断是不是须要让当前节点进入waiting状况，经由 shouldParkAfterFailedAcquire这个要领推断，如果须要让线程进入waiting状况的话，就挪用LockSupport的park要领让线程进入waiting状况。进入waiting状况后，这线程守候被interupt或许unpark（在release操纵中会举行如许的操纵，能够拜见背面的代码）。这个线程被叫醒后继承实行for轮回来尝试猎取资本。

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                //起首推断了当前节点的前继节点是不是是头节点，如果是的话就再次尝试猎取资本，
                //猎取资本胜利的话就直接返回false（示意未被中断过）
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                //推断是不是须要让当前节点进入waiting状况
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    // 如果在全部守候历程当中被中断过，则返回true，不然返回false。
                    // 如果线程在守候历程当中被中断过，它是不响应的。只是猎取资本后才再举行自我中断selfInterrupt()，将中断补上。
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

以上就是acquire要领的简朴剖析。

零丁看这个要领的话可能会不太清楚，连系ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步东西看这个代码的话就会好明白许多。

release(int)要领

release(int)要领是独有形式下线程开释同享资本的顶层进口。它会开释指定量的资本，如果完整开释了（即state=0）,它会叫醒守候行列里的其他线程来猎取资本。

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

//上面已讲过了，须要用户自定义完成
protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}

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

与acquire()要领中的tryAcquire()类似，tryRelease()要领也是须要独有形式的自定义同步器去完成的。一般来讲，tryRelease()都邑胜利的，由于这是独有形式，该线程来开释资本，那末它一定已拿到独有资本了，直接减掉响应量的资本即可(state-=arg)，也不须要斟酌线程平安的问题。

但要注重它的返回值，上面已提到了，release()是依据tryRelease()的返回值来推断该线程是不是已完成开释掉资本了！所以自义定同步器在完成时，如果已完整开释资本(state=0)，要返回true，不然返回false。

unparkSuccessor(Node)要领用于叫醒守候行列中下一个线程。这里要注重的是，下一个线程并不一定是当前节点的next节点，而是下一个能够用来叫醒的线程，如果这个节点存在，挪用unpark()要领叫醒。

总之，release()是独有形式下线程开释同享资本的顶层进口。它会开释指定量的资本，如果完整开释了（即state=0）,它会叫醒守候行列里的其他线程来猎取资本。

acquireShared(int)要领

acquireShared(int)要领是同享形式下线程猎取同享资本的顶层进口。它会猎取指定量的资本，猎取胜利则直接返回，猎取失利则进入守候行列，直到猎取到资本为止，全部历程疏忽中断。

public final void acquireShared(int arg) {
    //tryAcquireShared须要用户自定义完成
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

能够发明，这个要领的症结完成实际上是猎取资本失利后，怎样治理线程。也就是doAcquireShared的逻辑。

//不响应中断
private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

能够看出，doAcquireShared的逻辑和acquireQueued的逻辑差不多。将当前线程到场守候行列尾部歇息，直到其他线程开释资本叫醒本身，本身胜利拿到响应量的资本后才返回。

简朴总结下acquireShared的流程：

step1：tryAcquireShared()尝试猎取资本，胜利则直接返回；

step2：失利则经由历程doAcquireShared()进入守候行列park()，直到被unpark()/interrupt()并胜利猎取到资本才返回。全部守候历程也是疏忽中断的。

releaseShared(int)要领

releaseShared(int)要领是同享形式下线程开释同享资本的顶层进口。它会开释指定量的资本，如果胜利开释且许可叫醒守候线程，它会叫醒守候行列里的其他线程来猎取资本。

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

开释掉资本后，叫醒后继。跟独有形式下的release()类似，但有一点轻微须要注重：独有形式下的tryRelease()在完整开释掉资本（state=0）后，才会返回true去叫醒其他线程，这重如果基于独有下可重入的考量；而同享形式下的releaseShared()则没有这类请求，同享形式本质就是掌握一定量的线程并发实行，那末具有资本的线程在开释掉部份资本时就可以够叫醒后继守候结点。

参考

abp(net core)+easyui+efcore实现仓储管理系统——入库管理之一(三十七)