Java并发系列之CountDownLatch源码分析

CountDownLatch(闭锁)是一个很有用的工具类,利用它我们可以拦截一个或多个线程使其在某个条件成熟后再执行。它的内部提供了一个计数器,在构造闭锁时必须指定计数器的初始值,且计数器的初始值必须大于0。另外它还提供了一个countDown方法来操作计数器的值,每调用一次countDown方法计数器都会减1,直到计数器的值减为0时就代表条件已成熟,所有因调用await方法而阻塞的线程都会被唤醒。这就是CountDownLatch的内部机制,看起来很简单,无非就是阻塞一部分线程让其在达到某个条件之后再执行。但是CountDownLatch的应用场景却比较广泛,只要你脑洞够大利用它就可以玩出各种花样。最常见的一个应用场景是开启多个线程同时执行某个任务,等到所有任务都执行完再统计汇总结果。下图动态演示了闭锁阻塞线程的整个过程。

上图演示了有5个线程因调用await方法而被阻塞,它们需要等待计数器的值减为0才能继续执行。计数器的初始值在构造闭锁时被指定,后面随着每次countDown方法的调用而减1。下面代码贴出了CountDownLatch的构造方法。

//构造器
public CountDownLatch(int count) {
   if (count < 0) throw new IllegalArgumentException("count < 0");
   this.sync = new Sync(count);
 }

CountDownLatch只有一个带参构造器,必须传入一个大于0的值作为计数器初始值,否则会报错。可以看到在构造方法中只是去new了一个Sync对象并赋值给成员变量sync。和其他同步工具类一样,CountDownLatch的实现依赖于AQS,它是AQS共享模式下的一个应用。CountDownLatch实现了一个内部类Sync并用它去继承AQS,这样就能使用AQS提供的大部分方法了。下面我们就来看一下Sync内部类的代码。

//同步器
private static final class Sync extends AbstractQueuedSynchronizer {

  //构造器
  Sync(int count) {
    setState(count);
  }

  //获取当前同步状态
  int getCount() {
    return getState();
  }

  //尝试获取锁
  //返回负数:表示当前线程获取失败
  //返回零值:表示当前线程获取成功, 但是后继线程不能再获取了
  //返回正数:表示当前线程获取成功, 并且后继线程同样可以获取成功
  protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
  }

  //尝试释放锁
  protected boolean tryReleaseShared(int releases) {
    for (;;) {
      //获取同步状态
      int c = getState();
      //如果同步状态为0, 则不能再释放了
      if (c == 0) {
        return false;
      }
      //否则的话就将同步状态减1
      int nextc = c-1;
      //使用CAS方式更新同步状态
      if (compareAndSetState(c, nextc)) {
        return nextc == 0;
      }
    }
  }
}

可以看到Sync的构造方法会将同步状态的值设置为传入的参数值。之后每次调用countDown方法都会将同步状态的值减1,这也就是计数器的实现原理。在平时使用CountDownLatch工具类时最常用的两个方法就是await方法和countDown方法。调用await方法会阻塞当前线程直到计数器为0,调用countDown方法会将计数器的值减1直到减为0。下面我们来看一下await方法是怎样调用的。

//导致当前线程等待, 直到门闩减少到0, 或者线程被打断
public void await() throws InterruptedException {
  //以响应线程中断方式获取
  sync.acquireSharedInterruptibly(1);
}

//以可中断模式获取锁(共享模式)
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  //首先判断线程是否中断, 如果是则抛出异常
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
  //1.尝试去获取锁
  if (tryAcquireShared(arg) < 0) {
    //2. 如果获取失败则进人该方法
    doAcquireSharedInterruptibly(arg);
  }
}

当线程调用await方法时其实是调用到了AQS的acquireSharedInterruptibly方法,该方法是以响应线程中断的方式来获取锁的,上面同样贴出了该方法的代码。我们可以看到在acquireSharedInterruptibly方法首先会去调用tryAcquireShared方法尝试获取锁。我们看到Sync里面重写的tryAcquireShared方法的逻辑,方法的实现逻辑很简单,就是判断当前同步状态是否为0,如果为0则返回1表明可以获取锁,否则返回-1表示不能获取锁。如果tryAcquireShared方法返回1则线程能够不必等待而继续执行,如果返回-1那么后续就会去调用doAcquireSharedInterruptibly方法让线程进入到同步队列里面等待。这就是调用await方法会阻塞当前线程的原理,下面看看countDown方法是怎样将阻塞的线程唤醒的。

//减少门闩的方法
public void countDown() {
  sync.releaseShared(1);
}

//释放锁的操作(共享模式)
public final boolean releaseShared(int arg) {
  //1.尝试去释放锁
  if (tryReleaseShared(arg)) {
    //2.如果释放成功就唤醒其他线程
    doReleaseShared();
    return true;
  }
  return false;
}

可以看到countDown方法里面调用了releaseShared方法,该方法同样是AQS里面的方法,我们在上面也贴出了它的代码。releaseShared方法里面首先是调用tryReleaseShared方法尝试释放锁,tryReleaseShared方法在AQS里面是一个抽象方法,它的具体实现逻辑在子类Sync类里面,我们在上面贴出的Sync类代码里可以找到该方法。tryReleaseShared方法如果返回true表示释放成功,返回false表示释放失败,只有当将同步状态减1后该同步状态恰好为0时才会返回true,其他情况都是返回false。那么当tryReleaseShared返回true之后就会马上调用doReleaseShared方法去唤醒同步队列的所有线程。这样就解释了为什么最后一次调用countDown方法将计数器减为0后就会唤醒所有被阻塞的线程。CountDownLatch基本的原理大致就是这些,下面我们看一个它的使用示例。

应用场景:在玩欢乐斗地主时必须等待三个玩家都到齐才可以进行发牌。

public class Player extends Thread {

  private static int count = 1;
  private final int id = count++;
  private CountDownLatch latch;

  public Player(CountDownLatch latch) {
    this.latch = latch;
  }

  @Override
  public void run() {
    System.out.println("【玩家" + id + "】已入场");
    latch.countDown();
  }

  public static void main(String[] args) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(3);
    System.out.println("牌局开始, 等待玩家入场...");
    new Player(latch).start();
    new Player(latch).start();
    new Player(latch).start();
    latch.await();
    System.out.println("玩家已到齐, 开始发牌...");
  }

}

运行结果显示发牌操作一定是在所有玩家都入场后才进行。我们将23行的latch.await()注释掉,对比下看看结果。

可以看到在注释掉latch.await()这行之后,就不能保证在所有玩家入场后才开始发牌了。

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

您可能感兴趣的文章:

  • Java并发系列之Semaphore源码分析
  • Java并发系列之CyclicBarrier源码分析
  • Java并发系列之ConcurrentHashMap源码分析
  • Java并发系列之AbstractQueuedSynchronizer源码分析(共享模式)
  • Java并发系列之AbstractQueuedSynchronizer源码分析(独占模式)
  • Java并发系列之AbstractQueuedSynchronizer源码分析(概要分析)
  • Java并发系列之ReentrantLock源码分析
  • Java并发系列之AbstractQueuedSynchronizer源码分析(条件队列)
  • Java并发系列之Semaphore源码分析
  • Java并发系列之ReentrantLock源码分析
(0)

相关推荐

  • Java并发之ReentrantLock类源码解析

    ReentrantLock内部由Sync类实例实现. Sync类定义于ReentrantLock内部. Sync继承于AbstractQueuedSynchronizer. AbstractQueuedSynchronizer继承于AbstractOwnableSynchronizer. AbstractOwnableSynchronizer类中只定义了一个exclusiveOwnerThread变量,表示当前拥有的线程. 除了Sync类,ReentrantLock内部还定义了两个实现类. No

  • Java并发系列之AbstractQueuedSynchronizer源码分析(独占模式)

    在上一篇<Java并发系列[1]----AbstractQueuedSynchronizer源码分析之概要分析>中我们介绍了AbstractQueuedSynchronizer基本的一些概念,主要讲了AQS的排队区是怎样实现的,什么是独占模式和共享模式以及如何理解结点的等待状态.理解并掌握这些内容是后续阅读AQS源码的关键,所以建议读者先看完我的上一篇文章再回过头来看这篇就比较容易理解.在本篇中会介绍在独占模式下结点是怎样进入同步队列排队的,以及离开同步队列之前会进行哪些操作.AQS为在独占模

  • Java并发系列之ConcurrentHashMap源码分析

    我们知道哈希表是一种非常高效的数据结构,设计优良的哈希函数可以使其上的增删改查操作达到O(1)级别.Java为我们提供了一个现成的哈希结构,那就是HashMap类,在前面的文章中我曾经介绍过HashMap类,知道它的所有方法都未进行同步,因此在多线程环境中是不安全的.为此,Java为我们提供了另外一个HashTable类,它对于多线程同步的处理非常简单粗暴,那就是在HashMap的基础上对其所有方法都使用synchronized关键字进行加锁.这种方法虽然简单,但导致了一个问题,那就是在同一时间

  • Java并发系列之AbstractQueuedSynchronizer源码分析(条件队列)

    通过前面三篇的分析,我们深入了解了AbstractQueuedSynchronizer的内部结构和一些设计理念,知道了AbstractQueuedSynchronizer内部维护了一个同步状态和两个排队区,这两个排队区分别是同步队列和条件队列.我们还是拿公共厕所做比喻,同步队列是主要的排队区,如果公共厕所没开放,所有想要进入厕所的人都得在这里排队.而条件队列主要是为条件等待设置的,我们想象一下如果一个人通过排队终于成功获取锁进入了厕所,但在方便之前发现自己没带手纸,碰到这种情况虽然很无奈,但是它

  • Java并发系列之AbstractQueuedSynchronizer源码分析(概要分析)

    学习Java并发编程不得不去了解一下java.util.concurrent这个包,这个包下面有许多我们经常用到的并发工具类,例如:ReentrantLock, CountDownLatch, CyclicBarrier, Semaphore等.而这些类的底层实现都依赖于AbstractQueuedSynchronizer这个类,由此可见这个类的重要性.所以在Java并发系列文章中我首先对AbstractQueuedSynchronizer这个类进行分析,由于这个类比较重要,而且代码比较长,为了

  • Java并发系列之AbstractQueuedSynchronizer源码分析(共享模式)

    通过上一篇的分析,我们知道了独占模式获取锁有三种方式,分别是不响应线程中断获取,响应线程中断获取,设置超时时间获取.在共享模式下获取锁的方式也是这三种,而且基本上都是大同小异,我们搞清楚了一种就能很快的理解其他的方式.虽然说AbstractQueuedSynchronizer源码有一千多行,但是重复的也比较多,所以读者不要刚开始的时候被吓到,只要耐着性子去看慢慢的自然能够渐渐领悟.就我个人经验来说,阅读AbstractQueuedSynchronizer源码有几个比较关键的地方需要弄明白,分别是

  • Java并发系列之CyclicBarrier源码分析

    现实生活中我们经常会遇到这样的情景,在进行某个活动前需要等待人全部都齐了才开始.例如吃饭时要等全家人都上座了才动筷子,旅游时要等全部人都到齐了才出发,比赛时要等运动员都上场后才开始.在JUC包中为我们提供了一个同步工具类能够很好的模拟这类场景,它就是CyclicBarrier类.利用CyclicBarrier类可以实现一组线程相互等待,当所有线程都到达某个屏障点后再进行后续的操作.下图演示了这一过程. 在CyclicBarrier类的内部有一个计数器,每个线程在到达屏障点的时候都会调用await

  • Java并发系列之ReentrantLock源码分析

    在Java5.0之前,协调对共享对象的访问可以使用的机制只有synchronized和volatile.我们知道synchronized关键字实现了内置锁,而volatile关键字保证了多线程的内存可见性.在大多数情况下,这些机制都能很好地完成工作,但却无法实现一些更高级的功能,例如,无法中断一个正在等待获取锁的线程,无法实现限定时间的获取锁机制,无法实现非阻塞结构的加锁规则等.而这些更灵活的加锁机制通常都能够提供更好的活跃性或性能.因此,在Java5.0中增加了一种新的机制:Reentrant

  • Java并发系列之Semaphore源码分析

    Semaphore(信号量)是JUC包中比较常用到的一个类,它是AQS共享模式的一个应用,可以允许多个线程同时对共享资源进行操作,并且可以有效的控制并发数,利用它可以很好的实现流量控制.Semaphore提供了一个许可证的概念,可以把这个许可证看作公共汽车车票,只有成功获取车票的人才能够上车,并且车票是有一定数量的,不可能毫无限制的发下去,这样就会导致公交车超载.所以当车票发完的时候(公交车以满载),其他人就只能等下一趟车了.如果中途有人下车,那么他的位置将会空闲出来,因此如果这时其他人想要上车

  • Java并发系列之CountDownLatch源码分析

    CountDownLatch(闭锁)是一个很有用的工具类,利用它我们可以拦截一个或多个线程使其在某个条件成熟后再执行.它的内部提供了一个计数器,在构造闭锁时必须指定计数器的初始值,且计数器的初始值必须大于0.另外它还提供了一个countDown方法来操作计数器的值,每调用一次countDown方法计数器都会减1,直到计数器的值减为0时就代表条件已成熟,所有因调用await方法而阻塞的线程都会被唤醒.这就是CountDownLatch的内部机制,看起来很简单,无非就是阻塞一部分线程让其在达到某个条

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

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

随机推荐