CountDownLatch源码解析之await()

CountDownLatch 源码解析—— await(),具体内容如下

上一篇文章说了一下CountDownLatch的使用方法。这篇文章就从源码层面说一下await() 的原理。

我们已经知道await 能够让当前线程处于阻塞状态,直到锁存器计数为零(或者线程中断)。

下面是它的源码。

end.await();
  ↓
public void await() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);
}

sync 是CountDownLatch的内部类。下面是它的定义。

private static final class Sync extends AbstractQueuedSynchronizer {
  ...
}

它继承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer 这个类在java线程中属于一个非常重要的类。

它提供了一个框架来实现阻塞锁,以及依赖FIFO等待队列的相关同步器(比如信号、事件等)。

继续走下去,就跳到 AbstractQueuedSynchronizer 这个类中。

sync.acquireSharedInterruptibly(1);
  ↓
public final void acquireSharedInterruptibly(int arg) //AbstractQueuedSynchronizer
      throws InterruptedException {
  if (Thread.interrupted())
    throw new InterruptedException();
  if (tryAcquireShared(arg) < 0)
    doAcquireSharedInterruptibly(arg);
}

这里有两个判断,首先判断线程是否中断,然后再进行下一个判断,这里我们主要看看第二个判断。

protected int tryAcquireShared(int acquires) {
  return (getState() == 0) ? 1 : -1;
}

需要注意的是 tryAcquireShared 这个方法是在Sync 中实现的。

AbstractQueuedSynchronizer 中虽然也有对它的实现,但是默认的实现是抛一个异常。

tryAcquireShared 这个方法是用来查询当前对象的状态是否能够被允许获取锁。

我们可以看到Sync 中是通过判断state 是否为0 来返回对应的 int 值的。

那么 state 又代表什么?

/**
 * The synchronization state.
 */
  private volatile int state;

上面代码很清楚的表明 state 是表示同步的状态 。

需要注意的是 state 使用 volatile 关键字修饰。

volatile 关键字能够保证 state 的修改立即被更新到主存,当有其他线程需要读取时,会去内存中读取新值。

也就是保证了state的可见性。是最新的数据。

走到这里 state 是多少呢?

这里我们就需要看一看CountDownLatch 的 构造函数了。

CountDownLatch end = new CountDownLatch(2);
  ↓
public CountDownLatch(int count) {
  if (count < 0) throw new IllegalArgumentException("count < 0");
  this.sync = new Sync(count);
}
  ↓
Sync(int count) {
  setState(count);
}

原来构造函数中的数字就是这个作用啊,用来set state 。

所以我们这里state == 2 了。tryAcquireShared 就返回 -1。进入到下面

doAcquireSharedInterruptibly(arg);
  ↓
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
      for (;;) {
        final Node p = node.predecessor();
        if (p == head) {
          int r = tryAcquireShared(arg);
          if (r >= 0) {
            setHeadAndPropagate(node, r);
            p.next = null; // help GC
            failed = false;
            return;
          }
        }
        if (shouldParkAfterFailedAcquire(p, node) &&
          parkAndCheckInterrupt())
          throw new InterruptedException();
      }
    } finally {
      if (failed)
        cancelAcquire(node);
    }
  }

OK,这段代码有点长,里面还调用了几个函数。我们一行一行的看。

第一行 出现了一个新的类 Node。

Node 是AQS(AbstractQueuedSynchronizer)类中的内部类,定义了一种链式结构。如下所示。

   +------+ prev +-----+    +-----+
head |   | <---- |   | <---- |   | tail
   +------+    +-----+    +-----+

千万记住这个结构。

第一行代码中还有一个方法 addWaiter(Node.SHARED) 。

addWaiter(Node.SHARED) //Node.SHARED 表示该结点处于共享模式
  ↓
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; // private transient volatile Node tail;
  if (pred != null) {
    node.prev = pred;
    if (compareAndSetTail(pred, node)) {
      pred.next = node;
      return node;
    }
  }
  enq(node);
  return node;
}

首先是构造了一个Node,将当前的线程存进去了,模式是共享模式。

tail 表示 这个等待队列的队尾,此刻是null. 所以 pred == null ,进入到enq(node) ;

enq(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;
      }
    }
  }
}

同样tail 为 null , 进入到 compareAndSetHead 。

compareAndSetHead(new Node())
  ↓
/**
 * CAS head field. Used only by enq.
 */
private final boolean compareAndSetHead(Node update) {
  return unsafe.compareAndSwapObject(this, headOffset, null, update);
}

这是一个CAS操作,如果head 是 null 的话,等待队列的 head 就会被设置为 update 的值,也就是一个新的结点。

tail = head;  那么此时 tail 也不再是null了。进入下一次的循环。

这次首先将node 的 prev 指针指向 tail ,然后通过一个CAS 操作将node 设置为尾部,并返回了队列的 tail ,也就是 node 。

等待队列的模型变化如下

      +------+ prev   +----------------+
head(tail) |   | <---- node | currentThread |
      +------+      +----------------+

          ↓

    +------+ prev      +----------------+
head  |   | <---- node(tail) | currentThread |
    +------+         +----------------+

ok,到了这里await 方法 就返回了,是一个 thread 等于当前线程的Node。

返回到 doAcquireSharedInterruptibly(int arg) 中,进入下面循环。

for (;;) {
  final Node p = node.predecessor();
  if (p == head) {
    int r = tryAcquireShared(arg);
    if (r >= 0) {
      setHeadAndPropagate(node, r);
      p.next = null; // help GC
      failed = false;
      return;
    }
  }
  if (shouldParkAfterFailedAcquire(p, node) &&
    parkAndCheckInterrupt())
    throw new InterruptedException();
}

这个时候假设state 仍然大于0,那么此时 r < 0,所以进入到 shouldParkAfterFailedAcquire 这个方法 。

shouldParkAfterFailedAcquire(p, node)
  ↓
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  int ws = pred.waitStatus;
  if (ws == Node.SIGNAL) //static final int SIGNAL  = -1;
    /*
     * This node has already set status asking a release
     * to signal it, so it can safely park.
     */
    return true;
  if (ws > 0) {
    /*
     * Predecessor was cancelled. Skip over predecessors and
     * indicate retry.
     */
    do {
      node.prev = pred = pred.prev;
    } while (pred.waitStatus > 0);
    pred.next = node;
  } else {
    /*
     * waitStatus must be 0 or PROPAGATE. Indicate that we
     * need a signal, but don't park yet. Caller will need to
     * retry to make sure it cannot acquire before parking.
     */
    compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  }
  return false;
}
  ↓
/**
 * CAS waitStatus field of a node.
 */
private static final boolean compareAndSetWaitStatus(Node node,
                           int expect,
                           int update) {
  return unsafe.compareAndSwapInt(node, waitStatusOffset,
                  expect, update);
}

可以看到 shouldParkAfterFailedAcquire  也是一路走,走到 compareAndSetWaitStatus。

compareAndSetWaitStatus 将 prev 的 waitStatus 设置为 Node.SIGNAL 。

Node.SIGNAL 表示后续结点中的线程需要被unparking(类似被唤醒的意思)。该方法返回false。

经过这轮循环,队列模型变成下面状态

    +--------------------------+  prev      +------------------+
head  | waitStatus = Node.SIGNAL | <---- node(tail) | currentThread  |
    +--------------------------+         +------------------+

因为shouldParkAfterFailedAcquire返回的是false,所以后面这个条件就不再看了。继续 for (;;)  中的循环。

如果state仍然大于0,再次进入到 shouldParkAfterFailedAcquire。

这次因为head 中的waitStatus 为 Node.SIGNAL ,所以 shouldParkAfterFailedAcquire 返回true。

这次就需要看parkAndCheckInterrupt 这个方法了。

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

ok,线程没有被中断,所以,返回false。继续 for (;;)  中的循环。

如果state 一直大于0,并且线程一直未被中断,那么就一直在这个循环中。也就是我们上篇文章说的裁判一直不愿意宣布比赛结束的情况。

那么什么情况下跳出循环呢?也就是什么情况下state 会 小于0呢? 下一篇文章 我将说明。

总结一下,await()  方法 其实就是初始化一个队列,将需要等待的线程(state > 0)加入一个队列中,并用waitStatus 标记后继结点的线程状态。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • 详解ES6之async+await 同步/异步方案

    异步编程一直是JavaScript 编程的重大事项.关于异步方案, ES6 先是出现了 基于状态管理的 Promise,然后出现了 Generator 函数 + co 函数,紧接着又出现了 ES7 的 async + await 方案. 本文力求以最简明的方式来疏通 async + await. 异步编程的几个场景 先从一个常见问题开始:一个for 循环中,如何异步的打印迭代顺序? 我们很容易想到用闭包,或者 ES6 规定的 let 块级作用域来回答这个问题. for (let val of [

  • 深入理解ES7的async/await的用法

    在最开始学习ES6的Promise时,曾写过一篇博文 <promise和co搭配生成器函数方式解决js代码异步流程的比较> ,文章中对比了使用Promise和co模块搭配生成器函数解决js异步的异同. 在文章末尾,提到了ES7的async和await,只是当时只是简单的提了一下,并未做深入探讨. 在前两个月发布的Nodejs V7中,已添加了对async和await的支持,今天就来对这个东东做一下深入的探究.以更加优雅的方法写异步代码. async/await是什么 async/await可以

  • 关于async和await的一些误区实例详解

    微软官方的MSDN上说async和await是"异步",但是不少人(包括笔者自己)都有一些误区需要澄清:为什么await语句之后没有执行?不是异步吗? 先举一个示例代码如下: public partial class Form1 : Form { public async Task Processing() { await Task.Delay(5000); label1.Text = "Succuessful"; } public Form1() { Initia

  • 理解Koa2中的async&await的用法

    Koa是一款非常著名的Node服务端框架,有1.x版本和2.x版本.前者使用了generator来进行异步操作,后者则用了最新的async/await方案 一开始使用这种写法的时候,我遇到一个问题,代码如下: const Koa = require('koa'); const app = new Koa(); const doSomething = time => { return new Promise(resolve => { setTimeout(() => { resolve('

  • 浅谈Async和Await如何简化异步编程(几个实例让你彻底明白)

    引言 C#5.0中async和await两个关键字,这两个关键字简化了异步编程,之所以简化了,还是因为编译器给我们做了更多的工作,下面就具体看看编译器到底在背后帮我们做了哪些复杂的工作的. 同步代码存在的问题 对于同步的代码,大家肯定都不陌生,因为我们平常写的代码大部分都是同步的,然而同步代码却存在一个很严重的问题,例如我们向一个Web服务器发出一个请求时,如果我们发出请求的代码是同步实现的话,这时候我们的应用程序就会处于等待状态,直到收回一个响应信息为止,然而在这个等待的状态,对于用户不能操作

  • CountDownLatch源码解析之await()

    CountDownLatch 源码解析-- await(),具体内容如下 上一篇文章说了一下CountDownLatch的使用方法.这篇文章就从源码层面说一下await() 的原理. 我们已经知道await 能够让当前线程处于阻塞状态,直到锁存器计数为零(或者线程中断). 下面是它的源码. end.await(); ↓ public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } s

  • CountDownLatch源码解析之countDown()

    CountDownLatch 源码解析-- countDown() 上一篇文章从源码层面说了一下CountDownLatch 中 await() 的原理.这篇文章说一下countDown() . public void countDown() { //CountDownLatch sync.releaseShared(1); } ↓ public final boolean releaseShared(int arg) { //AQS if (tryReleaseShared(arg)) { d

  • Java并发编程之CountDownLatch源码解析

    一.前言 CountDownLatch维护了一个计数器(还是是state字段),调用countDown方法会将计数器减1,调用await方法会阻塞线程直到计数器变为0.可以用于实现一个线程等待所有子线程任务完成之后再继续执行的逻辑,也可以实现类似简易CyclicBarrier的功能,达到让多个线程等待同时开始执行某一段逻辑目的. 二.使用 一个线程等待其它线程执行完再继续执行 ...... CountDownLatch cdl = new CountDownLatch(10); Executor

  • CountDownLatch和Atomic原子操作类源码解析

    目录 引导语 1.CountDownLatch 1.1.await 1.2.countDown 1.3.示例 2.Atomic原子操作类 3.总结 引导语 本小节和大家一起来看看 CountDownLatch 和 Atomic 打头的原子操作类,CountDownLatch 的源码非常少,看起来比较简单,但 CountDownLatch 的实际应用却不是很容易:Atomic 原子操作类就比较好理解和应用,接下来我们分别来看一下. 1.CountDownLatch CountDownLatch 中

  • await 错误捕获实现方式源码解析

    目录 前言 Promise 的使用方法 await-to-js 源码 总结 前言 Promise 是一种在 JavaScript 中用于处理异步操作的机制.Promise 在开发中被广泛使用,这篇文章将学习如何优雅的捕获 await 的错误. 资源: 仓库地址:await-to-js 参考文章:How to write async await without try-catch blocks in Javascript (grossman.io) Promise 的使用方法 创建一个 Promi

  • JetCache 缓存框架的使用及源码解析(推荐)

    目录 一.简介 为什么使用缓存? 使用场景 使用规范 二.如何使用 引入maven依赖 添加配置 配置说明 注解说明 @EnableCreateCacheAnnotation @EnableMethodCache @CacheInvalidate @CacheUpdate @CacheRefresh @CachePenetrationProtect @CreateCache 三.源码解析 项目的各个子模块 常用注解与变量 缓存API Cache接口 AbstractCache抽象类 Abstra

  • java并发包工具CountDownLatch源码分析

    目录 一:简述 二:什么是CountDownLatch 三:CountDownLatch的使用 四:CountDownLatch原理分析 构造函数 await()方法: doAcquireSharedInterruptibly() 1. 当前节点的前置节点是head节点 2. 当前节点的前置节点不是head节点 addWaiter() setHeadAndPropagate() shouldParkAfterFailedAcquire() parkAndCheckInterrupt() coun

  • ReentrantLock从源码解析Java多线程同步学习

    目录 前言 管程 管程模型 MESA模型 主要特点 AQS 共享变量 资源访问方式 主要方法 队列 node节点等待状态 ReentrantLock源码分析 实例化ReentrantLock 加锁 A线程加锁成功 B线程尝试加锁 释放锁 总结 前言 如今多线程编程已成为了现代软件开发中的重要部分,而并发编程中的线程同步问题更是一道难以逾越的坎.在Java语言中,synchronized是最基本的同步机制,但它也存在着许多问题,比如可重入性不足.死锁等等.为了解决这些问题,Java提供了更加高级的

  • 详解ArrayBlockQueue源码解析

    今天要讲的是ArrayBlockQueue,ArrayBlockQueue是JUC提供的线程安全的有界的阻塞队列,一看到Array,第一反应:这货肯定和数组有关,既然是数组,那自然是有界的了,我们先来看看ArrayBlockQueue的基本使用方法,然后再看看ArrayBlockQueue的源码. ArrayBlockQueue基本使用 public static void main(String[] args) throws InterruptedException { ArrayBlocki

  • Tomcat源码解析之Web请求与处理

    前言 Tomcat最全UML类图 Tomcat请求处理过程: Connector对象创建的时候,会创建Http11NioProtocol的ProtocolHandler,在Connector的startInteral方法中,会启动AbstractProtocol,AbstractProtocol启动NioEndPoint进行监听客户端的请求,EndPoint接受到客户端的请求之后,会交给Container去处理请求.请求从Engine开始经过的所有容器都含有责任链模式,每经过一个容器都会调用该容

随机推荐