JAVA JVM 多线程

JAVA多线程之AQS分析(2)

CountDownLatch的分析来解读AQS的共享功能。

Posted by Jason Lee on 2019-07-23

AQS 共享功能的实现

CountDownLatch 的用法

CountDownLatch，CountDownLatch 为 java.util.concurrent包下的计数器工具类，常被用在多线程环境下，它在初始时需要指定一个计数器的大小，然后可被多个线程并发的实现减 1 操作，并在计数器为 0 后调用 await 方法的线程被唤醒，从而实现多线程间的协作。它在多线程环境下的基本使用方式为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
//main thread 新建一个 CountDownLatch，并指制定一个初始大小 
CountDownLatch countDownLatch = new CountDownLatch(3);
// 调用 await 方法后，main 线程将阻塞在这里，直到 countDownLatch 中的计数为 0 
countDownLatch.await();
System.out.println("over");
//thread1
// do something 
// 调用 countDown 方法，将计数减 1
  countDownLatch.countDown();

//thread2
// do something 
// 调用 countDown 方法，将计数减 1
  countDownLatch.countDown();

//thread3
// do something 
// 调用 countDown 方法，将计数减 1
  countDownLatch.countDown();

注意，线程 thread 1,2,3 各自调用 countDown 后，countDownLatch的计数为 0，await 方法返回，控制台输入“over”, 在此之前 main thread 会一直沉睡。
可以看到 CountDownLatch 的作用类似于一个“栏栅”，在 CountDownLatch 的计数为 0 前，调用 await 方法的线程将一直阻塞，直到 CountDownLatch 计数为 0，await 方法才会返回，
而 CountDownLatch 的 countDown() 方法则一般由各个线程调用，实现 CountDownLatch 计数的减 1。

CountDownLatch 的实现

构造函数

1
2
3
4
public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

和 ReentrantLock 类似，CountDownLatch 内部也有一个叫做 Sync 的内部类，同样也是用它继承了 AQS。

await

1
2
3
4
5
6
7
8
9
10
11
12
// CountDownLatch
public void await() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);
}
//AbstractQueuedSynchronizer
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  if (Thread.interrupted())
      throw new InterruptedException();
      // 同理  tryAcquireShared 是实现类实现的 这一点和读写锁一样
  if (tryAcquireShared(arg) < 0)
      doAcquireSharedInterruptibly(arg);
}

Sync 共享锁

从方法名上看，这个方法的调用是响应线程的打断的，所以在前两行会检查下线程是否被打断。接着，尝试着获取共享锁，小于 0，表示获取失败，通过本系列的上半部分的解读， 我们知道 AQS 在获取锁的思路是，先尝试直接获取锁，如果失败会将当前线程放在队列中，按照 FIFO 的原则等待锁。而对于共享锁也是这个思路，如果和独占锁一致，这里的 tryAcquireShared 应该是个空方法，留给子类去判断，我们来看 Sync 的具体实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;
    Sync(int count) { setState(count);}
    int getCount() { return getState();} 
    /**
     * 如果 state 变成 0 了，则返回 1，表示获取成功，否则返回 -1 则表示获取失败。
     * 看到这里，读者可能会发现， await 方法的获取方式更像是在获取一个独占锁，那为什么这里还会用 tryAcquireShared 呢？
     */
    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }
    protected boolean tryReleaseShared(int releases) {
        // Decrement count; signal when transition to zero
        for (;;) {
            int c = getState();
            if (c == 0)
                return false;
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
}

看到这里，读者可能会发现， await 方法的获取方式更像是在获取一个独占锁，那为什么这里还会用 tryAcquireShared 呢？

回想下 CountDownLatch 的 await 方法是不是只能在主线程中调用？答案是否定的，CountDownLatch 的 await 方法可以在多个线程中调用，当 CountDownLatch 的计数器为 0 后，调用 await 的方法都会依次返回。 也就是说可以多个线程同时在等待 await 方法返回，所以它被设计成了实现 tryAcquireShared 方法，获取的是一个共享锁，锁在所有调用 await 方法的线程间共享，所以叫共享锁。

如果获取共享锁失败（返回了 -1，说明 state 不为 0，也就是 CountDownLatch 的计数器还不为 0），进入调用 doAcquireSharedInterruptibly 方法中，按照我们上述的猜想，应该是要将当前线程放入到队列中去。

在这之前，我们再回顾一下 AQS 队列的数据结构：AQS 是一个双向链表，通过节点中的 next，pre 变量分别指向当前节点后一个节点和前一个节点。其中，每个节点中都包含了一个线程和一个类型变量：表示当前节点是独占节点还是共享节点，头节点中的线程为正在占有锁的线程，而后的所有节点的线程表示为正在等待获取锁的线程。

黄色节点为头节点，表示正在获取锁的节点，剩下的蓝色节点（Node1、Node2、Node3）为正在等待锁的节点，他们通过各自的 next、pre 变量分别指向前后节点，形成了 AQS 中的双向链表。每个线程被加上类型（共享还是独占）后便是一个 Node， 也就是本文中说的节点。
我再来看 doAcquireSharedInterruptibly 方法

Sync doAcquireSharedInterruptibly 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
  final Node node = addWaiter(Node.SHARED); 
// 将当前线程包装为类型为 Node.SHARED 的节点，标示这是一个共享节点。
  boolean failed = true;
  try {
      for (;;) {
          final Node p = node.predecessor();
          if (p == head) {
            // 如果新建节点的前一个节点，就是 Head，说明当前节点是 AQS 队列中等待获取锁的第一个节点，
            // 按照 FIFO 的原则，可以直接尝试获取锁。
              int r = tryAcquireShared(arg);
              if (r >= 0) {
                  setHeadAndPropagate(node, r); 
              // 获取成功，需要将当前节点设置为 AQS 队列中的第一个节点，这是 AQS 的规则
              // 队列的头节点表示正在获取锁的节点 
                  p.next = null; // help GC
                  failed = false;
                  return;
              }
          }
          if (shouldParkAfterFailedAcquire(p, node) && // 检查下是否需要将当前节点挂起 
              parkAndCheckInterrupt()) 
              throw new InterruptedException();
      }
  } finally {
      if (failed)
          cancelAcquire(node);
  }
}

这里有几点需要说明的：

  1. setHeadAndPropagate 方法：
    调用 setHeadAndPropagate 的方法的前提是 当前线程已经获取到了锁，且是第一个排队的节点，意味着和他的head 为空节点，
    当第一个排队的线程获取到了 锁之后， 首要的任务就是
    1、将自己弄成头结点，然后清空自己的节点为null
    2、然后判断排队自己后边排队的节点，如果是共享模式的，那么唤醒这个节点
1
2
3
4
5
6
7
8
9
10
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);  //将自己弄成头结点，然后清空自己的节点为null
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

Sync doReleaseShared 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
private void doReleaseShared() {
    for (;;) {
      // 这里为什么要 拿到head 因为共享锁是可以传播的，意思是如果某一个共享锁阻塞的线程被唤醒了
      // 那么意味着 排队链上的所有线程都应该被唤醒
      // 加入 当前线程是在 排队链的第n 个那么当唤醒的时候，我们要找到头部链
      // 如果 然后将head 链 唤醒，然后在 一次传播 
      // 如果 头结点是当前线程，那就意味着头结点已经被唤醒了或者已经持有锁了，那么意味着已经做过
      // 唤醒其他的操作了 详情看上面的代码 所有退出
      // 如果头结点是阻塞状态，那么自旋的去获取 ，知道解锁成功
      Node h = head;
      if (h != null && h != tail) {
          int ws = h.waitStatus;
          if (ws == Node.SIGNAL) { 
              if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 
            // 如果当前节点是 SIGNAL 意味着，它正在等待一个信号 可以理解为在已经是阻塞状态了，  
              // 或者说，它在等待被唤醒，因此做两件事，1 是重置 waitStatus 标志位，2 是重置成功后, 唤醒下一个节点。
                  continue;            // loop to recheck cases
              unparkSuccessor(h);
          }
          else if (ws == 0 &&
                  !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))  
// 如果本身头节点的 waitStatus 是出于重置状态（waitStatus==0）的，将其设置为“传播”状态。
// 意味着需要将状态向后一个节点传播。
              continue;                // loop on failed CAS
      }
      if (h == head)                   // loop if head changed
          break;
  }
}

为什么要这么做呢？这就是共享功能和独占功能最不一样的地方，对于独占功能来说，有且只有一个线程（通常只对应一个节点，拿 ReentantLock 举例，如果当前持有锁的线程重复调用 lock() 方法，那根据本系列上半部分我们的介绍，我们知道，会被包装成多个节点在 AQS 的队列中，所以用一个线程来描述更准确），能够获取锁，但是对于共享功能来说。
共享的状态是可以被共享的，也就是意味着其他 AQS 队列中的其他节点也应能第一时间知道状态的变化。因此，一个节点获取到共享状态流程图是这样的：

例如，当多个线程调用了await 方法的时候，这时候在Sycn 中会有一个等队列如下图：

Node1 变成了头节点然后调用 unparkSuccessor() 方法唤醒了 Node2、Node2 中持有的线程 A 出于上面流程图的 park node 的位置， 还记得 doAcquireSharedInterruptibly 里面的for循环吗，当Node2唤醒之后 会计入这个for循环当中，然后会判断当前node的 父节点，并唤醒。

线程 A 被唤醒后，重复黄色线条的流程，重新检查调用 tryAcquireShared 方法，看能否成功，如果成功，则又更改头节点，重复以上步骤，以实现节点自身获取共享锁成功后，唤醒下一个共享类型节点的操作，实现共享状态的向后传递。

  1. 其实对于 doAcquireShared 方法，AQS 还提供了集中类似的实现：

    分别对应了：
  • 带参数请求共享锁。 （忽略中断）
  • 带参数请求共享锁，且响应中断。（每次循环时，会检查当前线程的中断状态，以实现对线程中断的响应）
  • 带参数请求共享锁但是限制等待时间。（第二个参数设置超时时间，超出时间后，方法返回。）

Sync doAcquireSharedNanos方法

比较特别的为最后一个 doAcquireSharedNanos 方法，我们一起看下它怎么实现超时时间的控制的。
因为该方法和其余获取共享锁的方法逻辑是类似的，我用红色框圈出了它所不一样的地方，也就是实现超时时间控制的地方。
可以看到，其实就是在进入方法时，计算出了一个“deadline”，每次循环的时候用当前时间和“deadline”比较，大于“dealine”说明超时时间已到，直接返回方法。

注意，最后一个红框中的这行代码：

1
nanosTimeout > spinForTimeoutThreshold

从变量的字面意思可知，这是拿超时时间和超时自旋的最小作比较，在这里 Doug Lea 把超时自旋的阈值设置成了 1000ns, 即只有超时时间大于 1000ns 才会去挂起线程，否则，再次循环，以实现“自旋”操作。这是“自旋”在 AQS 中的应用之处。

countDown 方法

看完 await 方法，我们再来看下 countDown() 方法：

  • countDown() 方法
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    public void countDown() {
        sync.releaseShared(1);
    }
    //Sync 
    /*
      同样先尝试去释放锁，tryReleaseShared 同样为空方法，留给子类自己去实现，
    */
    public final boolean releaseShared(int arg) {
      if (tryReleaseShared(arg)) {
          doReleaseShared();
          return true;
      }
      return false;
    }

tryReleaseShared 方法

看一下 tryReleaseShared 的实现

1
2
3
4
5
6
7
8
9
10
11
protected boolean tryReleaseShared(int releases) {
  // Decrement count; signal when transition to zero
  for (;;) {
      int c = getState();
      if (c == 0)
          return false;
      int nextc = c-1;
      if (compareAndSetState(c, nextc))
          return nextc == 0;
  }
}

总结

本文从 CountDownLatch 入手，深入分析了 AQS 关于共享锁方面的实现方式：

如果获取共享锁失败后，将请求共享锁的线程封装成 Node 对象放入 AQS 的队列中，并挂起 Node 对象对应的线程，实现请求锁线程的等待操作。
待共享锁可以被获取后，从头节点开始，依次唤醒头节点及其以后的所有共享类型的节点。实现共享状态的传播。

这里有几点值得注意：

  • 与 AQS 的独占功能一样，共享锁是否可以被获取的判断为空方法，交由子类去实现。
  • 与 AQS 的独占功能不同，当锁被头节点获取后，独占功能是只有头节点获取锁，其余节点的线程继续沉睡，等待锁被释放后，才会唤醒下一个节点的线程，而共享功能是只要头节点获取锁成功，就在唤醒自身节点对应的线程的同时，继续唤醒 AQS 队列中的下一个节点的线程，每个节点在唤醒自身的同时还会唤醒下一个节点对应的线程，以实现共享状态的“向后传播”，从而实现共享功能。

以上的分析都是从 AQS 子类的角度去看待 AQS 的部分功能的，而如果直接看待 AQS，或许可以这么去解读：

  • 首先，AQS 并不关心“是什么锁”，对于 AQS 来说它只是实现了一系列的用于判断“资源”是否可以访问的 API, 并且封装了在“访问资源”受限时将请求访问的线程的加入队列、挂起、唤醒等操作， AQS 只关心“资源不可以访问时，怎么处理？”、“资源是可以被同时访问，还是在同一时间只能被一个线程访问？”、“如果有线程等不及资源了，怎么从 AQS 的队列中退出？”等一系列围绕资源访问的问题，而至于“资源是否可以被访问？”这个问题则交给 AQS 的子类去实现。

  • 当 AQS 的子类是实现独占功能时，例如 ReentrantLock，“资源是否可以被访问”被定义为只要 AQS 的 state 变量不为 0，并且持有锁的线程不是当前线程，则代表资源不能访问。

  • 当 AQS 的子类是实现共享功能时，例如：CountDownLatch，“资源是否可以被访问”被定义为只要 AQS 的 state 变量不为 0，说明资源不能访问。

这是典型的将规则和操作分开的设计思路：规则子类定义，操作逻辑因为具有公用性，放在父类中去封装。

当然，正式因为 AQS 只是关心“资源在什么条件下可被访问”，所以子类还可以同时使用 AQS 的共享功能和独占功能的 API 以实现更为复杂的功能。

比如：ReentrantReadWriteLock，我们知道 ReentrantReadWriteLock 的中也有一个叫 Sync的内部类继承了 AQS，而 AQS 的队列可以同时存放共享锁和独占锁，对于 ReentrantReadWriteLock 来说分别代表读锁和写锁，当队列中的头节点为读锁时，代表读操作可以执行，而写操作不能执行，因此请求写操作的线程会被挂起，当读操作依次推出后，写锁成为头节点，请求写操作的线程被唤醒，可以执行写操作，而此时的读请求将被封装成 Node 放入 AQS 的队列中。如此往复，实现读写锁的读写交替进行。

而本系列文章上半部分提到的 FutureTask，其实思路也是：封装一个存放线程执行结果的变量 A, 使用 AQS 的独占 API 实现线程对变量 A 的独占访问，判断规则是，线程没有执行完毕：call() 方法没有返回前，不能访问变量 A，或者是超时时间没到前不能访问变量 A(这就是 FutureTask 的 get 方法可以实现获取线程执行结果时，设置超时时间的原因)。

参考


赏 支付宝打赏 微信打赏

赞赏一下



FEATURED TAGS
JAVA JVM 多线程
FRIENDS