Java并发编程——AQS

分手后的思念是犯贱 2022-05-09 05:56 175阅读 0赞

### 本文转至 [链接太长][Link 1] ，如果页面失效，请直接关注微信公众号 “小孩子4919 我们都是小青蛙”查找文章—java并发性能（五）之牛逼的AQS（上）。特此申明！ ###

我们之前都是直接在线程中使用各种同步机制，我们可以把相关的同步问题抽象出来单独定义一些工具，这些工具可以在合适的并发场景下复用，下边我们看如何自定义我们自己的`同步工具`。

设计java的大叔们为了我们方便的自定义各种同步工具，为我们提供了大杀器`AbstractQueuedSynchronizer`类，这是一个抽象类，以下我们会简称`AQS`，翻译成中文就是`抽象队列同步器`。这家伙老有用了，封装了各种底层的同步细节，我们程序员想自定义自己的同步工具的时候，只需要定义这个类的子类并覆盖它提供的一些方法就好了。我们前边用到的显式锁`ReentrantLock`就是借助了`AQS`的神力实现的，现在马上来看看这个类的实现原理以及如何使用它自定义同步工具。

### 1.同步状态 ###

在`AQS`中维护了一个名叫`state`的字段，是由`volatile`修饰的，它就是所谓的`同步状态`：

private volatile int state;

并且提供了几个访问这个字段的方法：

<table> 
 <tbody> 
  <tr> 
   <td>方法名</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; &nbsp;final&nbsp; &nbsp;int&nbsp; &nbsp;getState( )</td> 
   <td>获取state的值</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; &nbsp;final&nbsp; &nbsp;void&nbsp; &nbsp;setState( int newState )</td> 
   <td>设置state的值</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; &nbsp;final&nbsp; &nbsp;boolean&nbsp; &nbsp;compareAndSetState( int&nbsp; expect, int&nbsp; update )</td> 
   <td>使用CAS方式更新state的值</td> 
  </tr> 
 </tbody> 
</table>

可以看到这几个方法都是`final`修饰的，说明子类中无法重写它们。另外它们都是`protected`修饰的，说明只能在子类中使用这些方法。

在一些线程协调的场景中，一个线程在进行某些操作的时候其他的线程都不能执行该操作，比如持有锁时的操作，在同一时刻只能有一个线程持有锁，我们把这种情景称为`独占模式`；在另一些线程协调的场景中，可以同时允许多个线程同时进行某种操作，我们把这种情景称为`共享模式`。

我们可以通过**修改`state`字段**代表的`同步状态`来实现多线程的**`独占模式`**或者**`共享模式`**。

**比如在`独占模式`下**，我们可以把`state`的初始值设置成`0`，每当某个线程要进行某项`独占`操作前，都需要判断`state`的值是不是`0`，如果不是`0`的话意味着别的线程已经进入该操作，则本线程需要阻塞等待；如果是`0`的话就把`state`的值设置成`1`，自己进入该操作。这个先判断再设置的过程我们可以通过`CAS`操作保证原子性，我们把这个过程称为`尝试获取同步状态`。如果一个线程`获取同步状态`成功了，那么在另一个线程`尝试获取同步状态`的时候发现`state`的值已经是`1`了就一直阻塞等待，直到`获取同步状态`成功的线程执行完了需要同步的操作后`释放同步状态`，也就是把`state`的值设置为`0`，并通知后续等待的线程。

**在`共享模式`下**的道理也差不多，比如说某项操作我们允许`10`个线程同时进行，超过这个数量的线程就需要阻塞等待。那么我们就可以把`state`的初始值设置为`10`，一个线程`尝试获取同步状态`的意思就是先判断`state`的值是否大于`0`，如果不大于`0`的话意味着当前已经有10个线程在同时执行该操作，本线程需要阻塞等待；如果`state`的值大于`0`，那么可以把`state`的值减`1`后进入该操作，每当一个线程完成操作的时候需要`释放同步状态`，也就是把`state`的值加`1`，并通知后续等待的线程。

所以对于我们自定义的同步工具来说，需要自定义获取同步状态与释放同步状态的方式，而`AQS`中的几个方法正是用来做这个事儿的：

<table> 
 <tbody> 
  <tr> 
   <td>方法名</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; boolean&nbsp; tryAcquire( int&nbsp; arg )</td> 
   <td>独占式的获取同步状态，获取成功返回true，否则false</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; boolean&nbsp; tryRelease( int&nbsp; arg )</td> 
   <td>独占式的释放同步状态，释放成功返回true，否则false</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; boolean&nbsp; tryAcquireShared( int&nbsp; arg )</td> 
   <td>共享式的获取同步状态，获取成功返回true，否则false</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; boolean&nbsp; tryAcquireShared( int&nbsp; arg )</td> 
   <td>共享式的释放同步状态，释放成功返回true，否则false</td> 
  </tr> 
  <tr> 
   <td>protected&nbsp; boolean&nbsp; isHeldExclusively(&nbsp; )</td> 
   <td> <p>在独占模式下，如果当前线程已经获取到同步状态，则返回true；</p> <p>其他情况则返回fasle</p> </td> 
  </tr> 
 </tbody> 
</table>

我们说`AQS`是一个抽象类，我们以`tryAcquire`为例看看它在`AQS`中的实现：

protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

我的天，竟然只是抛出个异常，这不科学。是的，在`AQS`中的确没有实现这个方法，不同的同步工具针对的具体并发场景不同，所以如何获取同步状态和如何释放同步状态是需要我们在自定义的`AQS`子类中实现的，如果我们自定义的同步工具需要在`独占模式`下工作，那么我们就重写`tryAcquire`、`tryRelease`和`isHeldExclusively`方法，如果是在`共享模式`下工作，那么我们就重写`tryAcquireShared`和`tryReleaseShared`方法。比如在独占模式下我们可以这样定义一个`AQS`子类：

public class Sync extends AbstractQueuedSynchronizer {
    
        @Override
        protected boolean tryAcquire(int arg) {
            return compareAndSetState(0, 1);
        }
    
        @Override
        protected boolean tryRelease(int arg) {
            setState(0);
            return true;
        }
    
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
    }

`tryAcquire`表示`尝试获取同步状态`，我们这里定义了一种极其简单的获取方式，就是使用`CAS`的方式把`state`的值设置成1，如果成功则返回`true`，失败则返回`false`，`tryRelease`表示`尝试释放同步状态`，这里同样采用了一种极其简单的释放算法，直接把`state`的值设置成`0`就好了。`isHeldExclusively`就表示当前是否有线程已经获取到了同步状态。如果你有更复杂的场景，可以使用更复杂的获取和释放算法来重写这些方法。

通过上边的唠叨，我们只是了解了啥是个`同步状态`，学会了如何通过继承`AQS`来自定义独占模式和共享模式下获取和释放同步状态的各种方法，但是你会惊讶的发现会了这些仍然没有什么卵用。我们期望的效果是一个线程获取同步状态成功会立即返回`true`，并继续执行某些需要同步的操作，在操作完成后释放同步状态，如果获取同步状态失败的话会立即返回`false`，并且进入阻塞等待状态，那线程是怎么进入等待状态的呢？不要走开，下节更精彩。

## ##

### 2. 同步队列 ###

`AQS`中还维护了一个所谓的`同步队列`，这个队列的`节点类`被定义成了一个静态内部类，它的主要字段如下

static final class Node {
        volatile int waitStatus;
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
        Node nextWaiter;
    
        static final int CANCELLED =  1;
        static final int SIGNAL    = -1;
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;
    }

`AQS`中定义一个`头节点引用`，一个`尾节点引用`：

private transient volatile Node head;
    private transient volatile Node tail;

通过这两个节点就可以控制到这个队列，也就是说可以在队列上进行诸如插入和移除操作。可以看到`Node`类中有一个`Thread`类型的字段，这**表明每一个节点都代表一个线程**。我们期望的效果是当一个线程获取同步状态失败之后，就把这个线程阻塞并包装成`Node`节点插入到这个`同步队列`中，当获取同步状态成功的线程释放同步状态的时候，同时通知在队列中下一个未获取到同步状态的节点，让该节点的线程再次去获取同步状态。

这个`节点类`的其他字段的意思我们之后遇到会详细唠叨，我们先看一下`独占模式`和`共享模式`下在什么情况下会往这个`同步队列`里添加节点，什么情况下会从它里边移除节点，以及线程阻塞和恢复的实现细节。

### 3. 独占式同步状态获取与释放 ###

在`独占模式`下，同一个时刻只能有一个线程获取到同步状态，其他同时去获取同步状态的线程会被包装成一个`Node`节点放到`同步队列`中，直到获取到同步状态的线程释放掉同步状态才能继续执行。初始状态的`同步队列`是一个空队列，里边一个节点也没有，就长这样：

![70][]

接下来我们就要详细看一下获取同步状态失败的线程是如何被包装成`Node`节点插入到队列中同时阻塞等待的。

前边说过，获取和释放同步状态的方式是由我们自定义的，在`独占模式`需要我们定义`AQS`的子类并且重写下边这些方法：

protected boolean tryAcquire(int arg)
    protected boolean tryRelease(int arg)
    protected boolean isHeldExclusively()

在定义了这些方法后，谁去调用它们呢？`AQS`里定义了一些调用它们的方法，这些方法都是由`public final`修饰的：

![70 1][]

忽然摆了这么多方法可能有点突兀哈，我们先看一下`acquire`方法的源代码：

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

代码显示`acquire`方法实际上是通过`tryAcquire`方法来获取同步状态的，如果`tryAcquire`方法返回`true`则结束，如果返回`false`则继续执行。这个`tryAcquire`方法就是我们自己规定的获取同步状态的方式。假设现在有一个线程已经获取到了同步状态，而线程`t1`同时调用`tryAcquire`方法尝试获取同步状态，结果就是获取失败，会先执行`addWaiter`方法，我们一起来看一下这个方法：

private Node addWaiter(Node mode) {
    
        Node node = new Node(Thread.currentThread(), mode);  //构造一个新节点
        Node pred = tail;
        if (pred != null) { //尾节点不为空，插入到队列最后
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {       //更新tail，并且把新节点插入到列表最后
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }
    
    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) {    //tail节点为空，初始化队列
                if (compareAndSetHead(new Node()))  //设置head节点
                    tail = head;
            } else {    //tail节点不为空，开始真正插入节点
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

可以看到，这个`addWaiter`方法就是向队列中插入节点的方法。首先会构造一个`Node`节点，假设这个节点为`节点1`，它的`thread`字段就是当前线程`t2`，这个节点被刚刚创建出来的样子就是这样：

![70 2][]

然后我们再分析一下具体的插入过程。如果`tail`节点不为空，直接把新节点插入到队列后边就返回了，如果`tail`节点为空，调用`enq`方法先初始化一下`head`和`tail`节点之后再把新节点插入到队列后边。`enq`方法的这几行初始化队列的代码需要特别注意：

if (t == null) {    //tail节点为空，初始化队列
        if (compareAndSetHead(new Node()))  //设置head节点
            tail = head;
    } else {
        //真正插入节点的过程
    }

也就是说在队列为空的时候会先让`head`和`tail`引用指向同一个节点后再进行插入操作，而这个节点竟然就是简简单单的`new Node()`，真是没有任何添加剂呀～ 我们先把这个节点称为`0号节点吧`，这个节点的任何一个字段都没有被赋值，所以在第一次节点插入后，`队列`其实长这样：

![70 3][]

其中的`节点1`才是我们真正插入的节点，代表获取同步状态失败的线程，`0号节点`是在初始化过程中创建的，我们之后再看它有什么用。

`addWaiter`方法调用完会返回新插入的那个节点，也就是`节点1`，`acquire`方法会接着调用`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);
        }
    }

可以看到，如果新插入的节点的前一个节点是`头节点`的话，会再次调用`tryAcquire`尝试获取同步状态，这个主要是怕获取同步状态的线程很快就把同步状态给释放了，所以在当前线程阻塞之前抱着侥幸的心理再试试能不能成功获取到同步状态，如果侥幸可以获取，那就调用`setHead`方法把`头节点`换成自己：

private void setHead(Node node) {
        head = node;
        node.thread = null;
        node.prev = null;
    }

同时把本`Node`节点的`thread`字段设置为`null`，意味着自己成为了`0号节点`。

如果当前`Node`节点不是头节点或者已经获取到同步状态的线程并没有释放同步状态，那就乖乖的往下执行`shouldParkAfterFailedAcquire`方法：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;   //前一个节点的状态
        if (ws == Node.SIGNAL)  //Node.SIGNAL的值是-1
            return true;
        if (ws > 0) {   //当前线程已被取消操作，把处于取消状态的节点都移除掉
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {    //设置前一个节点的状态为-1
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

这个方法是对`Node`节点中的`waitStatus`的各种操作。如果当前节点的前一个节点的`waitStatus`是`Node.SIGNAL`，也就是-1，那么意味着当前节点可以被阻塞，如果前一个节点的`waitStatus`大于`0`，意味着该节点代表的线程已经被取消操作了，需要把所有`waitStatus`大于`0`的节点都移除掉，如果前一个节点的`waitStatus`既不是`-1`，也不大于`0`，就把如果前一个节点的`waitStatus`设置成`Node.SIGNAL`。我们知道`Node`类里定义了一些代表`waitStatus`的静态变量，我们来看看`waitStatus`的各个值都是什么意思吧：

![70 4][]

未完待续！

[Link 1]: https://mp.weixin.qq.com/s?t=123&__biz=MzIxNTQ3NDMzMw==&mid=2247483910&idx=1&sn=f42489db76a6a1d7ff3a2ae48c2e0754&chksm=97968bf4a0e102e2c262af8ed5381f80d0aa2dc1a082d8a407bd1845008144db20e48f580dba&mpshare=1&scene=24&srcid=0923qDiCr8rPF9gsVpKkVx3o#rd
[70]: /images/20220509/d907f097d0e54b1db37d3f15c2cacd78.png
[70 1]: /images/20220509/352691fcaf5d45828450019007c72d2a.png
[70 2]: /images/20220509/60128a7459ac41a7989abc8aed09b64b.png
[70 3]: /images/20220509/24eaf409228f40d18b8e60ddd124d598.png
[70 4]: /images/20220509/5d3daf0de75e44bda8a5b60e313c6283.png