Java并发之-队列同步器AQS

Myth丶恋晨 2022-03-09 15:27 325阅读 0赞

## 前言 ##

AQS是AbstractQueuedSynchronizer的简称，是用来构建锁或者其他同步组建的基础框架，它使用一个 int 类型的成员变量来表示同步状态，通过内置的FIFO（先进先出）队列来完成资源获取和排队的。

在前面我讲了很多JUC中的同步工具，例如CountDownLatch、ReentrantLock等。其实我们知道这些同步工具都是通过继承AQS来实现的，所以AQS是这些同步工具的父类。所谓，了解一个人就要了解他的身世，爱一个人就要接受他的过去……

参考文献

《Java并发编程艺术》

## 正文 ##

同步器提供的模版方法基本分为3类：独占式获取与释放同步状态、共享式获取与释放同步状态和查询同步队列中的等待情况。

本文以独享式获取与释放同步状态为主，让大家了解获取与释放的流程。

## ##

### 同步器接口示例 ###

同步器提供3个方法来访问或修改同步状态。

*  getState():获取当前同步状态。
 *  setState(int newState):设置当前同步状态。
 *  compareAndSetState(int expect,int update):使用CAS设置当前状态，可以保证原子性。

<table>  同步器可重写的方法  
 <tbody> 
  <tr> 
   <td style="width:311px;">方法名称</td> 
   <td style="width:477px;">描述</td> 
  </tr> 
  <tr> 
   <td style="width:311px;">protected boolean&nbsp;tryAcquire(int arg)</td> 
   <td style="width:477px;"><strong>独占式获取同步状态</strong>，实现该方法需要查询当前状态并判断同步状态是否符合预期，然后再进行CAS设置同步状态。</td> 
  </tr> 
  <tr> 
   <td style="width:311px;">protected boolean&nbsp;tryRelease(int arg)</td> 
   <td style="width:477px;"><strong>独占式释放同步状态</strong>，等待获取同步状态的线程将有机会获取同步状态</td> 
  </tr> 
  <tr> 
   <td style="width:311px;">protected boolean&nbsp;tryAcquireShared(int arg)</td> 
   <td style="width:477px;"><strong>共享式获取同步状态</strong>，返回大于等于0表示获取成功，反之失败。</td> 
  </tr> 
  <tr> 
   <td style="width:311px;">protected boolean&nbsp;tryReleaseShared(int arg)</td> 
   <td style="width:477px;"><strong>共享式释放同步状态</strong></td> 
  </tr> 
  <tr> 
   <td style="width:311px;">protected boolean&nbsp;isHeldExclusively(int arg)</td> 
   <td style="width:477px;">当前同步器是否在独占模式下被线程一直占用，该方法返回是否被当前线程占用，true为持有占用</td> 
  </tr> 
 </tbody> 
</table>

### 同步队列结构 ###

同步器内部依赖一个同步队列，该队列遵循FIFO来完成同步状态的管理，当前线程获取同步失败时，同步器会把当前线程以及等待状态等信息构造成一个节点（Node）并加入同步队列尾部，同时阻塞当前线程，当同步状态释放时，把首节点线程唤醒，使其再次尝试获取同步状态。

节点是同步器的基础，同步器拥有首节点（head），和尾节点（tail）。如图是同步队列的结构。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70][]

**节点加入到队列尾节点**

一个线程获取到来同步状态（或锁），其他线程没有获取到，然后就被加入到队列的尾节点，这个过程要求必须线程安全，所以用CAS设置尾节点，只有设置成功后，当前节点才正式与之前尾节点建立关联。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 1][]

**首节点获取**

首节点是获取同步状态成功的节点，也就是出列队的线程是首节点，当首节点释放后，将唤醒后继节点，后继节点将会在获取同步状态时成功将自己设置为首节点，然后等待下次被释放。

说明：设置首节点是通过获取同步状态的线程完成的，由于只有一个线程可以获取同步状态，所以设置首节点不用CAS来保证线程安全。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 2][]

### 独占式同步状态获取 ###

获取同步状态通过acquire（int arg），该方法失败会进入同步队列。

**讲解代码：**

该方法完成了同步状态的获取、节点构造、加入队列以及在同步队列中自旋等操作，主要逻辑是：首先使用 tryAcquire 方法安全的获取线程的同步状态，如果失败则通过 addWaiter 方法构造尾节点加入队列中，最后调用 acquireQueued 方法使得该节点无限循环的方式获取同步状态，获取不到则阻塞节点的线程，解除阻塞只有唤醒前驱节点或阻塞线程中断来实现。enq 方法中，通过无限循环来保证节点正确添加。

节点进入同步队列后，就进入一个自旋的过程，每个线程都在观察，当满足条件，获取到同步状态就会从自旋过程退出，否则一直自旋。

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    
    
    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)) {
                    pred.next = node;
                    return node;
                }
            }
            enq(node);
            return node;
        }
    
    
    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;
                    }
                }
            }
        }
    
    
      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);
            }
        }

**自旋获取同步示意图**

首节点成功获取同步状态后将会唤醒后继节点，后继节点线程被唤醒后需要检查自己前驱节点是否是头节点。

白话：你要我继承你的位置做老大，首先我看你是不是我老大。

可以看出节点和节点之间在循环检查的过程中基本不互相通信，只是简单的判断自己的前驱节点是不是头节点而已，这样做符合FIFO。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 3][]

**独占式同步状态获取流程图**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 4][]

### 独占式同步状态释放 ###

独占式同步器释放同步状态使用 release 方法。

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

执行该方式时，会唤醒头节点的后继节点线程，在释放同步状态时，同步器调用 tryRelease 方法释放同步状态，然后唤醒头节点的后继节点。

### 共享式同步状态获取与释放 ###

这里不再贴代码讲解，我只做简单的介绍。

共享式获取与独占式获取主要区别在于同一时间能否有多个线程同时获取同步状态。举个例子，文件读写时，既保证高效有保证不被脏读的方法就是，写操作对资源独占访问，读操作可以共享访问。所以大家更好理解为什么 ReentrantReadWriteLock 的读可以共享了。关于共享方法在文章前面我已经列出了共享式方法的介绍。

个人微信，加微信可拉入java技术交流群。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 5][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70]: /images/20220309/08269e82e4584bca9de06fb52abe458d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 1]: /images/20220309/ca9bcfd1afef4de3a25cfab3d9ee0df7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 2]: /images/20220309/a81f52dad7b64488b15a147249373882.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 3]: /images/20220309/ed1c19af6fdc4cf49f102a71ae7e6898.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 4]: /images/20220309/956afd48ece74808854707a89c759a1b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70 5]: /images/20220309/f793e4654d324cae946d7f9b33b631a6.png