【搞定Java并发编程】第16篇：队列同步器AQS源码分析之独占模式

以你之姓@ 2022-04-05 14:45 120阅读 0赞

**AQS系列文章：**

**[1、队列同步器AQS源码分析之概要分析][1_AQS]**

**[2、队列同步器AQS源码分析之独占模式][2_AQS]**

**[3、队列同步器AQS源码分析之共享模式][3_AQS]**

**[4、队列同步器AQS源码分析之Condition接口、等待队列][4_AQS_Condition]**

--------------------

本文主要讲解队列同步器AQS的独占模式：主要分为**独占式同步状态获取（不响应中断）、独占式同步状态释放、独占式获取同步状态（响应中断）、独占式超时获取同步状态**。

## **目  录：** ##

**1、独占式同步状态获取：不响应中断**

**2、独占式同步状态的释放**

**3、独占式获取同步状态：可响应中断**

**4、独占式超时获取同步状态**

--------------------

# 1、独占式同步状态获取：不响应中断 #

通过调用同步器的acquire(int  arg)方法可以获取同步状态，该方法对中断不敏感，也是由于线程获取同步状态失败后进入同步队列中，后续对线程进行中断操作时，线程不会从同步队列中移除。

// 独占式同步状态获取与释放（不响应中断方式获取）
    public final void acquire(int arg) {
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
            selfInterrupt();
        }
    }

上诉代码主要完成了同步状态获取、节点构造、加入同步队列以及在同步队列中自旋等待的相关工作。

其主要逻辑是：首先调用自定义同步器实现的tryAcquire(int  arg)方法，该方法保证线程安全的获取同步状态，如果失败则构造同步节点（独占式Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过addWriter(Node node)方法将该节点加入到同步队列的尾部，最后调用acquireQueued(Node node, int arg)方法，使得该节点以“死循环”的方式获取同步状态。

如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出队或阻塞线程被中断来实现。

![20181212213342851.png][]

*  **第一步：tryAcquire(int arg)：尝试去获取同步状态 / 锁 **

// 尝试去获取同步状态(独占模式)
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

tryAcquire(int arg)方法需要子类去覆盖，重写里面的判断逻辑。如果获取到了同步状态则退出返回，否则生成节点加入同步队列尾部。

*  **第二步：addWaiter(Node mode)：将当前线程包装成结点并添加到同步队列尾部**

// 将当前线程包装成结点并添加到同步队列尾部
    private Node addWaiter(Node mode) {
        // 指定持有锁的模式
        Node node = new Node(Thread.currentThread(), mode);
        // 获取同步队列尾结点引用
        Node pred = tail;
        // 如果尾结点不为空, 表明同步队列已存在结点
        if (pred != null) {
            // 1.指向当前尾结点
            node.prev = pred;
            // 2.设置当前结点为尾结点
            if (compareAndSetTail(pred, node)) {
                // 3.将旧的尾结点的后继指向新的尾结点
                pred.next = node;
                return node;
            }
        }
        // 否则表明同步队列还没有进行初始化
        enq(node);
        return node;
    }
    
    // 结点入队操作
    private Node enq(final Node node) {
        for (;;) {
            // 获取同步队列尾结点引用
            Node t = tail;
            // 如果尾结点为空说明同步队列还没有初始化
            if (t == null) {
                // 初始化同步队列
                if (compareAndSetHead(new Node())) {
                    tail = head;
                }
            } else {
                // 1.指向当前尾结点
                node.prev = t;
                // 2.设置当前结点为尾结点
                if (compareAndSetTail(t, node)) {
                    // 3.将旧的尾结点的后继指向新的尾结点
                    t.next = node;
                    return t;
                }
            }
        }
    }

上诉代码通过使用compareAndSetTail(Node expect, Node update)方法来确保节点能够被线程安全添加。

在enq(final  Node  node)方法中，同步器通过“死循环”来保证节点的正确添加，在“死循环”中只有通过CAS将节点设置成为尾结点之后，当前线程才能从该方法返回，否则当前线程不断地尝试设置。

节点进入同步队列以后，就进入了一个自旋的过程，每个节点（或者说每个线程）都在自省地观察，当条件满足，获取到了同步状态，就可以从这个自旋过程中退出，否则依旧留在这个自旋过程中。

执行到这一步表示，获取同步状态失败，进入同步队列中排队去了，需要说明是当前线程获取同步状态是独占模式还是共享模式，然后将这个线程挂起。

*  **第三步：acquireQueued(addWaiter(Node.EXCLUSIVE), arg)：以独占式获取同步状态 / 锁**

// 独占式同步状态获取
    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)) {
                    // 将给定结点设置为head结点
                    setHead(node);
                    // 为了帮助垃圾收集, 将上一个head结点的后继清空
                    p.next = null;
                    // 设置获取成功状态
                    failed = false;
                    // 返回中断的状态, 整个循环执行到这里才是出口
                    return interrupted;
                }
                // 否则说明锁的状态还是不可获取, 这时判断是否可以挂起当前线程
                // 如果判断结果为真则挂起当前线程, 否则继续循环, 在这期间线程不响应中断
                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
                    interrupted = true;
                }
            }
        } finally {
            // 在最后确保如果获取失败就取消获取
            if (failed) {
                cancelAcquire(node);
            }
        }
    }
    
    // 判断是否可以将当前结点挂起
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 获取前继结点的等待状态
        int ws = pred.waitStatus;
        // 如果前继结点状态为SIGNAL, 表明前继结点会唤醒当前结点, 所以当前结点可以安心的挂起了
        if (ws == Node.SIGNAL) {
            return true;
        }
    
        if (ws > 0) {
            // 下面的操作是清理同步队列中所有已取消的前继结点
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 到这里表示前继结点状态不是SIGNAL, 很可能还是等于0, 这样的话前继结点就不会去唤醒当前结点了 
            // 所以当前结点必须要确保前继结点的状态为SIGNAL才能安心的挂起自己
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    
    // 挂起当前线程
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

在acquireQueued(final  Node node, int arg)方法中，当前线程在“死循环”中尝试获取同步状态，而只有前驱节点是头节点才能够尝试获取同步状态。当头节点的线程释放了同步状态后，将会唤醒其后继节点，后继节点的线程被唤醒后需要检查自己的前驱节点是否是头节点。

整个for循环就只有一个出口，那就是等线程成功的获取到同步状态 / 锁之后才能出去，在没有获取到同步状态 / 锁之前就一直是挂在for循环的parkAndCheckInterrupt()方法里。线程被唤醒后也是从这个地方继续执行for循环。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70][]**节点自旋获取同步状态**

从上图中可以看出，节点与节点之间的循环检查的过程中基本上不相互通信，而是简单地判断自己的前驱是否为头节点，这样使得节点的释放规则符合FIFO。

*  **第四步：selfInterrupt()：中断当前线程【可选项】**

// 当前线程将自己中断
    private static void selfInterrupt() {
        Thread.currentThread().interrupt();
    }

由于上面整个线程一直是挂在for循环的parkAndCheckInterrupt()方法里，没有成功获取到锁之前不会响应任何形式的线程中断，只有当线程成功获取到锁并从for循环出来后，才会查看在这期间是否有人要求中断线程，如果是的话再去调用selfInterrupt()方法将自己挂起。

独占式同步状态的获取流程，也就是acquire(int  arg)方法的调用流程如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 1][]

当前线程获取同步状态并执行了相应逻辑后，就需要释放同步状态，使得后继节点能够继续获取同步状态。通过调用同步器的release(int  arg)方法可以释放同步状态，该方法在释放了同步状态之后，会唤醒其后继节点。

--------------------

# **2、独占式同步状态的释放** #

public final boolean release(int arg) {
        // 拨动密码锁, 看看是否能够开锁
        if (tryRelease(arg)) {
            // 获取head结点
            Node h = head;
            // 如果head结点不为空并且等待状态不等于0就去唤醒后继结点
            if (h != null && h.waitStatus != 0) {
                // 唤醒后继结点
                unparkSuccessor(h);
            }
            return true;
        }
        return false;
    }
    
    // 唤醒后继结点
    private void unparkSuccessor(Node node) {
        // 获取给定结点的等待状态
        int ws = node.waitStatus;
        // 将等待状态更新为0
        if (ws < 0) {
            compareAndSetWaitStatus(node, ws, 0);
        }
        // 获取给定结点的后继结点
        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);
        }
    }

该方法执行时会唤醒头节点的后继节点线程，unparkSuccessor(Node  node)方法使用LockSupport来唤醒处于等待状态的线程。

以上分析了独占式同步状态的获取和释放过程，做个总结：

**在获取同步状态时，同步器维护一个同步队列，获取状态失败的线程都会被加入到队列中进行自旋；移除队列（或停止自旋）的条件是前驱节点为头节点且成功获取了同步状态。在释放同步状态时，同步器调用tryRelease(int  arg)方法释放同步状态，然后唤醒头节点的后继节点。**

--------------------

# 3、独占式获取同步状态：可响应中断 #

上诉过程是独占式获取同步状态的过程，这个过程是不响应线程的中断的。那怎样响应线程中断获取同步状态呢？

// 以可中断模式获取同步状态 / 锁(独占模式)
    private void doAcquireInterruptibly(int arg) throws InterruptedException {
        // 将当前线程包装成结点添加到同步队列中
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                // 获取当前结点的前继结点
                final Node p = node.predecessor();
                // 如果p是head结点, 那么当前线程就再次尝试获取锁
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    //获取锁成功后返回
                    return;
                }
                // 如果满足条件就挂起当前线程, 此时响应中断并抛出异常
                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
                    // 线程被唤醒后如果发现中断请求就抛出异常
                    throw new InterruptedException();
                }
            }
        } finally {
            if (failed) {
                cancelAcquire(node);
            }
        }
    }

响应线程中断方式和不响应线程中断方式获取锁流程上大致上是相同的。唯一的一点区别就是线程从parkAndCheckInterrupt方法中醒来后会检查线程是否中断，如果是的话就抛出InterruptedException异常，而不响应线程中断获取锁是在收到中断请求后只是设置一下中断状态，并不会立马结束当前获取锁的方法，一直到结点成功获取到锁之后才会根据中断状态决定是否将自己挂起。

--------------------

# 4、独占式超时获取同步状态 #

通过调用同步器的doAcquireNanos(int arg, long nanosTimeout)方法可以超时获取同步状态，即在指定的时间段内获取同步状态，如果获取到同步状态则返回true，否则返回false。该方法提供了传统Java同步操作（比如synchronized关键字）所不具备的特性。

针对超时获取，主要需要计算出需要睡眠的时间间隔nanosTimeout，为了防止过早通知，nanosTimeout计算公式为：nanosTimeout -= now - lastTime，其中now为当前唤醒时间，lastTime为上次唤醒时间，如果nanosTime大于0则表示超时时间未到，需要继续睡眠nanosTimeout纳秒，反之，则已经超时。

// 以限定超时时间获取同步状态/ 锁(独占模式)
    private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
        // 获取系统当前时间
        long lastTime = System.nanoTime();
        // 将当前线程包装成结点添加到同步队列中
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                // 获取当前结点的前继结点
                final Node p = node.predecessor();
                //如果前继是head结点, 那么当前线程就再次尝试获取锁
                if (p == head && tryAcquire(arg)) {
                    // 更新head结点
                    setHead(node);
                    p.next = null;
                    failed = false;
                    return true;
                }
                // 超时时间用完了就直接退出循环
                if (nanosTimeout <= 0) {
                    return false;
                }
                // 如果超时时间大于自旋时间, 那么等判断可以挂起线程之后就会将线程挂起一段时间
                if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) {
                    // 将当前线程挂起一段时间, 之后再自己醒来
                    LockSupport.parkNanos(this, nanosTimeout);
                }
                // 获取系统当前时间
                long now = System.nanoTime();
                //超时时间每次都减去获取锁的时间间隔
                nanosTimeout -= now - lastTime;
                // 再次更新lastTime
                lastTime = now;
                // 在获取锁的期间收到中断请求就抛出异常
                if (Thread.interrupted()) {
                    throw new InterruptedException();
                }
            }
        } finally {
            if (failed) {
                cancelAcquire(node);
            }
        }
    }

该方法在自旋过程中，当节点的前驱节点为头节点时尝试获取同步状态，如果获取成功则从方法中返回，这个过程和独占式同步获取的过程类似，但是在同步状态获取失败的处理上不同。如果当前线程获取同步状态失败，则判断是否超时（nanosTimeout小于等于0，就代表已经超时了），如果没有超时，重新计算nanosTimeout，然后使当前线程等待nanosTimeout纳秒（当已到设置的超时时间，该线程会从LockSupport.parkNanos(Object blocker, long  nanos)方法返回）。

注意在以超时时间获取锁的过程中是可以响应线程中断请求的。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 2][]**独占式超时获取同步状态的流程**

从上图可以看出：独占式超时获取同步状态doAcquireNanos(int arg, long  nanosTimeout)和独占式获取同步状态acquire(int  args)在流程上非常相似，其主要区别在于未获取到同步状态时的处理逻辑。acquire(int  args)在未获取同步状态时，将会使当前线程一直处于等待状态，而doAcquireNanos(int arg, long  nanosTimeout)会使当前线程等待nanosTimeout纳秒，如果当前线程在nanosTimeout纳秒没有获取到同步状态，将会从等待逻辑中自动返回。

--------------------

**AQS系列文章：**

**[1、队列同步器AQS源码分析之概要分析][1_AQS]**

**[2、队列同步器AQS源码分析之独占模式][2_AQS]**

**[3、队列同步器AQS源码分析之共享模式][3_AQS]**

**[4、队列同步器AQS源码分析之Condition接口、等待队列][4_AQS_Condition]**

## 参考及推荐： ##

**[1、AbstractQueuedSynchronizer源码分析之概要分析][1_AbstractQueuedSynchronizer]**

**2、[AbstractQueuedSynchronizer源码分析之独占模式][AbstractQueuedSynchronizer]**

**3、[AbstractQueuedSynchronizer源码分析之共享模式][AbstractQueuedSynchronizer 1]**

**4、[AbstractQueuedSynchronizer源码分析之条件队列][AbstractQueuedSynchronizer 2]**

[1_AQS]: https://blog.csdn.net/pcwl1206/article/details/84972609
[2_AQS]: https://blog.csdn.net/pcwl1206/article/details/84980812
[3_AQS]: https://blog.csdn.net/pcwl1206/article/details/84983118
[4_AQS_Condition]: https://blog.csdn.net/pcwl1206/article/details/84984713
[20181212213342851.png]: /images/20220405/29695d18f40f4e7c9afbcc6d65e85d18.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70]: /images/20220405/b0580167191447efb0c83db0b145c91c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 1]: /images/20220405/421eded65c294f74bb50862d26ce48a0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bjd2wxMjA2_size_16_color_FFFFFF_t_70 2]: /images/20220405/e6ad99ee7a6145e986381708c2e2e864.png
[1_AbstractQueuedSynchronizer]: https://mp.weixin.qq.com/s/RzD-_SAo3f68lS_05H4B2g
[AbstractQueuedSynchronizer]: https://mp.weixin.qq.com/s/ORoB7yj5oA35FCtuR8gYUA
[AbstractQueuedSynchronizer 1]: https://mp.weixin.qq.com/s/BGB1_UTujwmlQAFIwx4zXQ
[AbstractQueuedSynchronizer 2]: https://mp.weixin.qq.com/s/95wZVLSGbY9ubu3Ot-eAvQ