【Java并发编程】AQS(2)——独占锁的获取

谁践踏了优雅 2023-07-22 08:00 29阅读 0赞

今天是4月4日，清明节第一天，互联网一片灰白，大家都在缅怀逝者，致敬英烈。所以今天我也没有过多的娱乐，一天都在鼓捣这篇文章。今天这篇主要说说AQS独占锁的获取。

AQS中对独占锁的获取一共有三个方法，今天主要说第一个

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

acquire方法，即在独占模式下获取锁，并且忽略中断。它至少调用一次tryAcquire方法去获取锁，如果成功则直接返回，否则线程将被包装成节点(即AQS内部类Node) 进入同步队列，并且其可能反复阻塞和解除阻塞，并调用tryAcquire去获取锁，直到最后成功。

上面这段话详细介绍了acquire方法的执行过程，如果不理解没关系，等看完下面的源码解读后，一切就清晰了

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

上面的源码不太直观，也不方便我后续讲解，所以在不改变逻辑的前提下我将源码改写如下：

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

我们可以看到，就四个方法，看上去是不是很简单，所以接下来我们就将依次解读上面四个方法

## **1  tryAcquire** ##

我在"[并发三板斧][Link 1]"已经说过，tryAcquire方法是AQS中的钩子方法，是需要子类重写实现的，**其主要功能就是获取锁**。当我们获取到锁时，返回true，acquire方法就直接return结束了；如果没拿到锁，返回false，调用入队方法addWaiter。下面是tryAcquire的源码

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

我们需要注意的是tryAcquire并不是一个抽象方法，而是抛了一个不支持运行的异常，这是为什么呢？其实很简单，还记得"[并发三板斧][Link 1]"文章中说的，不同的模式只需要重写特定的钩子方法吗？继承AQS的子类并不是所有的基本方法都需要重写，而是按需重写，如果钩子方法都定义成抽象方法，则我们在实现AQS子类时就要重写一些并不需要用到的方法

## **2  addWaiter** ##

当获取锁失败后，我们就会调用此方法，**此方法主要功能是将获取锁失败的线程包装成Node放入等待队列中**，即入队操作（ps：等待队列是FIFO队列，出队在head端，入队在tail端），下面是源码

private Node addWaiter(Node mode) {
       Node node = new Node(Thread.currentThread(), mode);
       // Try the fast path of enq; backup to full enq on failure
       Node pred = tail;
       if (pred != null) {
           node.prev = pred;
           if (compareAndSetTail(pred, node)) {
               pred.next = node;
               return node;
           }
       }
       enq(node);
       return node;
    }

我们首先会将当前线程和Node.EXCLUSIVE标志位构造成一个Node。然后将pre指向尾结点tail，然后判断pre是否为null

如果为null，说明同步队列未初始化，则此时是没有Node(准确的说是Node里的线程，后面统一用Node来表示)在等待锁，则我们会调用enq方法进行队列初始化，然后重新入队

如果不为null，说明队列不为空，则我们就会进入第一个if分支内，尝试将节点放入同步队列的队尾

首先，node.prev = pred 我们将要入队的Node的前驱Node指向原来的尾结点pred，此时的队列结构图如下：

![format_png][]

真的是如上图所示吗？不一定哈，上图是在没有并发的情况下，如果在并发的情况下，则可能如下图所示：

![format_png 1][]

此时可能有几个节点都在进行入队，且都走到了node.prev = pred 这一步，将自己的前驱节点指向了尾结点pred，所以下面就到了CAS发挥作用的时候啦

compareAndSetTail(pred, node)

此时只有一个节点会操作成功，我们假设中间的Node成功执行了这个操作，则此时变为

![format_png 2][]

则中间这个Node持有的线程会进入到第二个if分支内，完成Node入队的剩余操作，即pred的后继Node指向tail，然后返回此Node。最上和最下这两个操作失败的节点则不会进入if分支内，而是调用enq方法，进行重新入队操作

## **2.1  enq** ##

在addWaiter方法解读中我们看到了，有两种情况我们会进入enq方法，第一种是同步队列未初始化，第二种是在并发情况下入队失败。我们来看源码

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

if分支是处理队列为空的情况，即初始化队列

else分支是处理入队失败的情况，这个入队和addWaiter中的入队是一模一样的，所以也是会失败的，但是我们可以看到，这里有个for循环自旋，所以当我们入队失败后会再次尝试，一直到入队成功

这里还需要特别注意的是，enq返回的是入队Node的前驱节点，这里大家有个印象就行了，这里并没涉及这个注意点，因为addAwaiter方法中调用enq是没有接受返回值的

## **3  acquireQueued** ##

我们通过addWaiter入队成功后，就会调用此方法，**此方法的主要功能是挂起刚入队Node中的线程，然后等待被唤醒再去获取锁**。但需要注意的是，在挂起线程之前，如果满足一定条件，此线程还会再次去获取锁，失败后才挂起线程。满足的是什么条件呢？我们通过源码来分析

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

acquireQueued方法中有两个标志变量：failed和interrupted，他们分别代表拿锁失败标志和线程被中断标志，两者为true时分别代表拿锁失败和中断成功。

我们首先会将这两个标志位初始化，然后我们就会进入for循环自旋。首先我们会找到入队Node的前驱Node，然后进入第一个if判断：判断前驱Node是否是头结点head，如果是，则说明入队的Node在同步队列的第一个，前面没有等待的Node了，此时我们会调用tryAcquire方法来再一次获取锁，获取成功后，我们进入第一个if分支中（现在大家应该知道在什么情况下会再次去尝试拿锁了吧）

第一个if分支中的主要操作是更新head，即将head指向此时的Node，然后将Node中的前驱Node和线程置为null，使得此Node变为一个虚拟头结点 ，最后再更新两个标志位并返回interrupted

如果此入队Node的前驱Node不是head或者在第一个if判断中拿锁失败，我们就会进入第二个if判断，第二个if判断中我们会先执行shouldParkAfterFailedAcquire方法

## **3.1  shouldParkAfterFailedAcquire** ##

看这个方法名也能明白其功能：检查Node中的线程是否需要被挂起，如果返回true则说明需要挂起，然后执行后续挂起方法parkAndCheckInterrupt，否则重新自旋。我们需要注意两点

1.  走到这个方法的线程，都已经调用tryAcquire一次或多次失败了
2.  此方法不仅仅判断线程能否被挂起，它还有**将同步队列中属性为CANCELLED的Node移除队列**的功能

我们看源码：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
       int ws = pred.waitStatus;
       if (ws == Node.SIGNAL)
           return true;
       if (ws > 0) {
           do {
               node.prev = pred = pred.prev;
           } while (pred.waitStatus > 0);
           pred.next = node;
       } else {
           compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
       }
       return false;
    }

"[并发三板斧][Link 2]"中我们说过Node一共有五种状态，其中独占模式下会使用三种状态：CANCELLED、SIGNAL、初始状态0，我们可以看到上面的代码也是三段分支。首先我们会拿到此节点的**前驱节点状态**，为什么是前驱Node的状态？因为在独占模式下，Node是否能够被挂起的依据是它前驱节点是否为SIGNAL，为SIGNAL时才能被挂起

1.  前驱节点状态为SIGNAL，直接返回true
2.  前驱节点状态为CANCELLED（ws>0）,则移除这些节点，返回false
3.  其他情况则将前驱节点的状态改为SIGNAL，返回false

我们看到，只有当前驱节点为SIGNAL才返回true；其余情况都返回false，然后回到acquireQueued方法中自旋重新执行

## **3.2 parkAndCheckInterrupt** ##

如果shouldParkAfterFailedAcquire返回ture，我们则通过parkAndCheckInterrupt方法来挂起线程

private final boolean parkAndCheckInterrupt() {
       LockSupport.park(this);
       return Thread.interrupted();
    }

这里我们需要注意，当执行完LockSupport.park(this)后，此线程就被挂起了，除非当其他线程调用LockSupport.unpark唤醒当前线程或者当前线程被中断，否则后面代码是不会执行的

假设此时线程被唤醒了，我们此时是不知道线程是被unpark方法还是中断唤醒的，所以我们需要通过Thread类的interrupted方法来判断。interrupted方法会返回给我们当前线程的中断标志位，并将中断标志位复位，即置为false。如果我们是中断唤醒的，则返回true，然后会进入acquireQueued的第二个If分支中将interrupted置为true。然后再次进入for循环自旋，看是获取锁还是又被挂起。

最后acquiredQueued方法只会存在两种情况，第一种是获取锁然后返回interrupted标志位，第二种出现异常，执行finally中if分支的cancelAcquire方法（注意，获取锁成功是不会执行cancelAcquire的，因为failed标志位为false）

## **4  selfInterrupt** ##

我们最后返回到acquire方法，如果acquire返回的是true，说明Node是被中断唤醒的，则会调用selfInterrupt方法再一次调用中断

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

为啥还要执行中断呢？还记得文章开头我们是怎么介绍acquire这个方法的吗？acquire方法是在独占模式下不响应中断获取锁的方法。如果在parkAndCheckInterrupt方法中线程是被中断唤醒的，我们还是会继续回到acquiredQueued中去抢锁然后执行

当然interrupte这个方法也只是将当先线程的中断标志置为true，至于会不会被中断，我们也不知道

独占锁的获取我们就讲完了，最后我再将文章的脉络梳理下：

![format_png 3][]

需要注意的是上面改变Node状态的地方一共有两处：

1. Node的初始化

2. shouldParkAfterFailedAcquire方法中

(未完）

欢迎大家关注我的公众号 “程序员进阶之路”，里面记录了一个非科班程序员的成长之路

![format_png 4][]

[Link 1]: http://mp.weixin.qq.com/s?__biz=Mzg3ODE5OTgzOQ==&mid=2247483809&idx=1&sn=e1b01fb81a299c51cd023a28e41a5948&chksm=cf1614d1f8619dc7d6625178a8bc6e78b49b91d7a156c1083b64a3a06634e853e7547c5519e5&scene=21#wechat_redirect
[format_png]: /images/20230528/7887f017a588431cae10345faba5b898.png
[format_png 1]: /images/20230528/254fdfb62fb34b97b1ddd7a53c194f41.png
[format_png 2]: /images/20230528/6ef0fa1a5c97419d8d90ae365213db64.png
[Link 2]: http://mp.weixin.qq.com/s?__biz=Mzg3ODE5OTgzOQ==&mid=2247483780&idx=1&sn=df7323b85df2fb78cd267624e374f33a&chksm=cf1614f4f8619de2aedcb55010407615b4e05d9fe67fdb4261088f7292fd340c54a4ccf27e97&scene=21#wechat_redirect
[format_png 3]: /images/20230528/724c36503d3e4323921d593a32080878.png
[format_png 4]: /images/20230528/c62dee6b53884d3db497fb89bd90d957.png