听说你想学Java并发编程？先把这个学了(2)--20210604

大家好，我是指北君。

俗话说，铁要趁热打，指北君在写完AQS第一篇文章后，就马不停蹄的输出第二篇了，这篇主要是讲AQS是如何解决互斥问题的，如果没看过AQS系列第一篇的童鞋，建议先把第一篇看完，它是后面两篇的基础。

说到互斥，我们第一个反应是什么？锁！对，AQS就是利用的锁来解决互斥的，那我们就来看看AQS是如何实现这个锁的。

AQS提供了两种锁，独占锁和共享锁。独占锁只有一把锁，同一时间只允许一个线程获得锁；而共享锁则有多把锁，同一时间允许多个线程获得锁。我们本文主要讲独占锁。

一. 独占锁的获取

AQS中对独占锁的获取一共有三个方法：

acquire：不响应中断获取独占锁
acquireInterruptibly：响应中断获取独占锁
tryAcquireNanos：响应中断+超时获取独占锁

由于篇幅，我们主要着眼于acquire方法，当然，只要你理解了acquire，acquireInterruptibly和tryAcquireNanos自然不在话下了，因为这两个方法只是在acquire的基础上增加了一些判断逻辑来处理中断和超时情况而已。

我们上源码

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

其acquire方法中一共有四个方法，其逻辑也分为4步：

tryAcquire：尝试获取锁，成功即acquire方法结束，否则调用addWaiter
addWaiter：获取锁失败即调用此方法入队，即将获取锁失败的线程包装成Node放入同步队列的队尾
acquireQueued：入队成功后即调用此方法，如果Node在队首则再次抢锁，否则挂起等待唤醒（唤醒后再去获取锁）
selfInterrupt：如果是被中断唤醒，则再次执行中断

粗略介绍完后，我们现在一个一个方法看。

1.1 tryAcquire

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

tryAcquire是钩子方法，是我们根据需要重写的。其功能就是在独占模式下去获取锁，获取成功则返回true，acquire方法直接结束；如果获取失败返回false，则后续会调用后面要讲的addWaiter方法将线程入队。

因为AQS是模板类，不同的子类只需要重写不同的钩子方法，因此，tryAcquire不能设置成抽象方法，不然一些不需要此钩子方法的子类也要实现这个方法。所以作者对tryAcquire的默认实现是抛了一个异常（当然我认为直接写个return也是ok的）。

1.2 addWaiter

如果tryAcquire获取锁失败后，我们就会调用addWaiter将线程包装成Node入队挂起。addWaiter的大致逻辑是：先将线程包装成Node，然后入队，如果队列未初始化或者入队失败，则会调用子方法enq，enq来进行初始化队列和自旋入队，我们看下具体代码：

private Node addWaiter(Node mode) {
   // 将此线程包装成Node
   Node node = new Node(Thread.currentThread(), mode);
   // 将pred指向尾结点
   Node pred = tail;
   // 如果pred 即尾结点不为null,说明同步队列初始化完成了。
   if (pred != null) {
       // 尾插法
       // 步骤一：将node的前驱指针指向当前尾结点
       node.prev = pred;
       // 步骤二：通过CAS将尾结点指向当前节点
       if (compareAndSetTail(pred, node)) {
           pred.next = node;
           return node;
       }
   }
   // 走到这一步有两个原因
   // 1是队列未初始化，2是尾结点插入失败
   enq(node);
   return node;
}

下面是enq方法，当执行到这个方法时，说明线程获取锁已经失败了，然后入队过程又失败了，入队过程失败有两个原因：

同步队列未初始化
入队过程中CAS操作失败

private Node enq(final Node node) {
   for (;;) {
       Node t = tail;
       // 队列为空, 初始化队列操作，即将head和tail指向一个空节点
       if (t == null) { 
           if (compareAndSetHead(new Node()))
               tail = head;
       } else {  // 队列不为空
           // 并发下，cas操作可能会失败，所以通过for循环不断进行入队，直到成功为止
           node.prev = t;
           if (compareAndSetTail(t, node)) {
               t.next = node;
               return t;
           }
       }
   }
}

CAS节点入队失败的原因，我们看到enq源码中执行完尾插法的步骤一，即将Node的前驱指针指向当前尾结点，如果是并发情况下，应该是如下图所示（紫色节点代表我们关注的Node）：

此时，可能有多个Node都准备入队，所以此时可能有多个Node的前驱节点都指向尾结点，所以我们在执行步骤二将尾结点指向Node时，采用的是CAS，即只有一个Node能成功，假设我们关注的Node入队成功了，如下图：

则另外两个CAS操作肯定会失败，即它们将要进入enq方法重新自旋入队。

1.3 acquireQueued

执行完addWaiter方法后，说明我们已经入队成功了，此时我们需要将Node中的线程挂起，等待下次被唤醒。

但在挂起之前，我们需要再次检查下我们此时的Node是否是在队首，如果在队首，我们又会再次去抢锁。否则我们会通过shouldParkAfterFailedAcquire判断是否要挂起（shouldParkAfterFailedAcquire不仅仅是判断此线程是否可以被挂起，还会将同步队列中属性为CANCELLED的Node移除队列），如果需要挂起，则调用parkAndCheckInterrupt将线程挂起。具体源码如下：

final boolean acquireQueued(final Node node, int arg) {
   // 获取失败标签，默认ture，如果获取到锁了后则会置为false
   boolean failed = true;
   try {
       // 中断标签，默认false
       boolean interrupted = false;
       for (;;) {
           // 获取此节点的前驱节点
           final Node p = node.predecessor();
           // 如果前驱节点是头结点,则会再次调用tryAcquire抢锁
           // 如果抢锁成功了，则进入if语句，然后return
           if (p == head && tryAcquire(arg)) {
               // 将此节点设置为头结点
               setHead(node);
               p.next = null; // help GC
               // 获取失败标志置为false,因为拿到锁了
               failed = false;
               // 返回中断标志
               return interrupted;
           }
           //  shouldParkAfterFailedAcquire判断是否要挂起
           //  如果要挂起，则调用parkAndCheckInterrupt将线程挂起
           if (shouldParkAfterFailedAcquire(p, node) &&
               parkAndCheckInterrupt())
               interrupted = true;
       }
   } finally {
       if (failed)
           cancelAcquire(node);
   }
}

shouldParkAfterFailedAcquire源码如下。其主要作用有2：

决定获取锁失败后，是否将线程挂起
清除同步队列中所有状态为CANCELLED的节点

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
   int ws = pred.waitStatus;
   // 如果此节点的前驱节点为SIGNAL,则说明此节点需要挂起，返回true
   if (ws == Node.SIGNAL)
       return true;
   // 如果此节点的前驱节点状态大于0，即状态为CANCELLED则移除前驱节点，然后再往前遍历，直到清除完所有CANCELLED的节点
   if (ws > 0) {
       do {
           node.prev = pred = pred.prev;
       } while (pred.waitStatus > 0);
       pred.next = node;
   } else {
       // 将前驱节点置为SIGNAL
       compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
   }
   return false;
}

这是acquireQueued中的最后一步，即将线程挂起，然后静静的等待被唤醒。除非该线程被其他线程unpark或者被中断，否则该线程的程序将一直停止在这。

private final boolean parkAndCheckInterrupt() {
   // 通过LockSupport挂起线程
   LockSupport.park(this);
   // 返回线程的标志位，true表示此线程被中断过
   return Thread.interrupted();
}

1.4 selfInterrupt

通过我们前面的分析可以知道，当线程被中断过，则会进入到此方法。

而interrupte这个方法也只是将当前线程的中断标志置为true，至于会不会被中断，这个是由系统决定的。

static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

二. 独占锁的的释放

相比独占锁的获取，独占锁的释放逻辑就简单多了。独占锁释放只做了两件事情：

释放锁
唤醒head结点后最近需要被唤醒的节点。

其释放逻辑的实现是通过release方法，而做的两件事分别对应了其子方法tryRelease和unparkSuccessor：

public final boolean release(int arg) {
    // 如果释放锁成功，则进入if去唤醒同步队列中的线程
    if (tryRelease(arg)) {
        Node h = head;
        // head节点不为空(即同步队列不为空) 且 状态不为0(初始化队列时，head结点waitStatus为0，此时等待队列中是没有节点的)
        // 则唤醒head结点后继节点
        if (h != null && h.waitStatus != 0)
            // 唤醒离head最近需要被唤醒的节点
            unparkSuccessor(h);
        return true;
    }
    return false;
}

2.1 tryRelease

这个方法和tryAcquire一样，也是钩子方法，是留给子类重写的，作用是用来释放锁，如果释放成功则返回true，失败返回false，这个具体的实现我们也放在后续AQS的子类中讲解，这里就不过多阐述了。

2.2 unparkSuccessor

此方法的作用是唤醒后继Node，我们看代码：

private void unparkSuccessor(Node node) {
 
    int ws = node.waitStatus;
    // waitStatus<0，说明此时waitStatus为SIGNAL
    if (ws < 0)
      	// 此时需要将waitStatus置为0，待会唤醒后继节点
        compareAndSetWaitStatus(node, ws, 0);
 		
    Node s = node.next;
    // 此Node的后继节点如果是null或者状态为CANCELLED，则此Node已经不存在或者取消
    // 则我们需要从尾结点往前遍历找到离head最近的需要被唤醒的Node
    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)
        // 唤醒Node中的线程
        LockSupport.unpark(s.thread);
}

这里需要注意的是，我们在找需要被唤醒的节点时，为什么是从后往前遍历呢？

其实这和获取锁时的尾结点入队有关，我们再看下入队方法addWaiter中插入尾结点的相关代码：

node.prev = pred;   //step1
if (compareAndSetTail(pred, node))   // step2
     pred.next = node;  // step3

假设我们此时有个Node正在入队，执行完step2，还未执行step3，unparkSuccessor中如果采用从head往后遍历，是找不到这个新插入的Node的；但如果是采用从后往前遍历，则不会出现这个问题。

三. 总结

对于独占锁的获取与释放，指北君就分析完了，这里我再总结一下：

获取独占锁是通过acquire来实现的，首先通过tryAcquire获取锁，如果获取成功，则直接返回，如果失败，则会调用addWaiter方法进行入队，如果入队过程中发现队列未初始化，则会初始化队列再进行入队，入队不成功则会一直自旋直到成功；入队成功后就会挂起，直到被其他线程或者中断唤醒；唤醒后会检查线程的中断标志位，如果被中断过，会再次调用中断方法，告诉系统自己需要被中断。

释放独占锁是通过release方法实现的，其首先通过tryRelease释放锁，如果失败则直接返回false，如果成功则会调用unparkSuccessor唤醒后继节点。

通过上面的分析，大家应该了解了AQS是如何解决互斥问题的。后面指北君将会讨论AQS如何解决线程间通信协作问题，敬请期待~