Java CountDownLatch的源码硬核解析

目录
  • 前言
  • 介绍和使用
    • 例子
    • 概述
  • 实现思路
  • 源码解析
    • 类结构图
    • await() 实现原理
    • countDown()实现原理

前言

对于并发执行,Java中的CountDownLatch是一个重要的类,简单理解, CountDownLatchcount down是倒数的意思,latch则是“门闩”的含义。在数量倒数到0的时候,打开“门闩”, 一起走,否则都等待在“门闩”的地方。

为了更好的理解CountDownLatch这个类,本文通过例子和源码带领大家深入解析这个类的原理。

介绍和使用

例子

我们先通过一个例子快速理解下CountDownLatch的妙处。

最近LOL S12赛如火如荼举行,比如我们玩王者荣耀的时候,10个万玩家登入游戏,每个玩家的网速可能不一样,只有每个人进度条走完,才会一起来到王者峡谷,网速快的要等网速慢的。我们通过例子模拟下这个过程。

@Slf4j(topic = "a.CountDownLatchTest")
public class CountDownLatchTest {

    public static void main(String[] args) throws InterruptedException {
        // 创建一个倒时器,默认10个数量
        CountDownLatch latch = new CountDownLatch(10);
        ExecutorService service = Executors.newFixedThreadPool(10);
        // 设置进度数据
        String[] personProcess = new String[10];
        Random random = new Random();

        for (int i = 0; i < 10; i++) {
            int finalJ = i;
            service.submit(() -> {
                // 模拟10个人的进度条
                for (int j = 0; j <= 100; j++) {
                    // 模拟网速快慢,随机生成
                    try {
                        Thread.sleep(random.nextInt(100));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    // 设置进度数据
                    personProcess[finalJ] = j + "%";
                   log.info("{}", Arrays.toString(personProcess));
                }

                // 运行结束,倒时器 - 1
                latch.countDown();
            });
        }
        // 打开"阀门"
        latch.await();
       log.info("王者峡谷到了");
        service.shutdown();
    }
}

运行结果:

概述

CountDownLatch一般用作多线程倒计时计数器,强制它们等待其他一组(CountDownLatch的初始化决定)任务执行完成。

构造器:

public CountDownLatch(int count):设置倒数器需要倒数的数量

常用API:

  • public void await() throws InterruptedException:调用await()方法的线程会被挂起,等待直到count值为0再继续执行。
  • public boolean await(long timeout, TimeUnit unit) throws InterruptedException:同await(),若等待timeout时长后,count值还是没有变为0,不再等待,继续执行。时间单位如下常用的毫秒、天、小时、微秒、分钟、纳秒、秒。
  • public void countDown(): count值递减1
  • public long getCount():获取当前count值

常见使用场景:

一个程序中有N个任务在执行,我们可以创建值为N的CountDownLatch,当每个任务完成后,调用一下countDown()方法进行递减count值,再在主线程中使用await()方法等待任务执行完成,主线程继续执行。

实现思路

通过前面的例子和介绍我们知道CountDownLatch的大致使用流程:

  • 创建CountDownLatch并设置计数器值。
  • 启动多线程并且调用CountDownLatch实例的countDown()方法。
  • 主线程调用 await() 方法,这样主线程的操作就会在这个方法上阻塞,直到其他线程完成各自的任务,count值为0,停止阻塞,主线程继续执行。

不妨我们先思考下,它是怎么实现的呢?我们可以问自己几个问题?

  • 如何做到可以让主线程阻塞等待在那里?是不是可以调用LockSupport.park()方法进行阻塞。
  • 那么什么时候该阻塞呢?我们需要有个变量,比如state, 如果state大于0,就阻塞主线程。
  • 那么什么时候该唤醒呢,又如何唤醒呢?如果任务执行完成后,我们让state 减去1,也就是调用countDown()方法,如果发现state是0,那么就调用LockSupport.unpark()唤醒此前阻塞的地方,继续执行。

是不是很熟悉,这就是我们的AQS共享模式的实现原理啊,不了解AQS共享模式的可以参考本篇文章:深入浅出理解Java并发AQS的共享锁模式

我们把思路理清楚后,直接看CountDownLatch的源码。

源码解析

类结构图

以上是CountDownLatch的类结构图,

  • SyncCountDownLatch的内部类,被成员变量sync持有。
  • Sync继承了AbstractQueuedSynchronizer,也就是我们大名鼎鼎的AQS。

await() 实现原理

1.线程调用 await()会阻塞等待其他线程完成任务

// CountDownLatch#await
public void await() throws InterruptedException {
    // 调用AbstractQueuedSynchronizer的acquireSharedInterruptibly方法
    sync.acquireSharedInterruptibly(1);
}
// AbstractQueuedSynchronizer#acquireSharedInterruptibly
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    // 判断线程是否被打断,抛出打断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 尝试获取共享锁
    // 条件成立说明 state > 0,此时线程入队阻塞等待,等待其他线程获取共享资源
    // 条件不成立说明 state = 0,此时不需要阻塞线程,直接结束函数调用
    if (tryAcquireShared(arg) < 0)
        // 阻塞当前线程的逻辑
        doAcquireSharedInterruptibly(arg);
}
// CountDownLatch.Sync#tryAcquireShared
protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

2.doAcquireSharedInterruptibly()方法是实现线程阻塞的核心逻辑

// AbstractQueuedSynchronizer#doAcquireSharedInterruptibly
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
    // 将调用latch.await()方法的线程 包装成 SHARED 类型的 node 加入到 AQS 的阻塞队列中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            // 获取当前节点的前驱节点
            final Node p = node.predecessor();
            // 前驱节点时头节点就可以尝试获取锁
            if (p == head) {
                // 再次尝试获取锁,获取成功返回 1
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 获取锁成功,设置当前节点为 head 节点,并且向后传播
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 阻塞在这里
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        // 阻塞线程被中断后抛出异常,进入取消节点的逻辑
        if (failed)
            cancelAcquire(node);
    }
}

3.parkAndCheckInterrupt()方法中会进行阻塞操作

private final boolean parkAndCheckInterrupt() {
    	// 阻塞线程
        LockSupport.park(this);
        return Thread.interrupted();
    }

countDown()实现原理

1.任务结束调用 countDown() 完成计数器减一(释放锁)的操作

public void countDown() {
    sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
    // 尝试释放共享锁
    if (tryReleaseShared(arg)) {
        // 释放锁成功开始唤醒阻塞节点
        doReleaseShared();
        return true;
    }
    return false;
}

2.调用tryReleaseShared()方法尝试释放锁,true表示state等于0,去唤醒阻塞线程。

protected boolean tryReleaseShared(int releases) {
    for (;;) {
        int c = getState();
        // 条件成立说明前面【已经有线程触发唤醒操作】了,这里返回 false
        if (c == 0)
            return false;
        // 计数器减一
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
            // 计数器为 0 时返回 true
            return nextc == 0;
    }
}

3.调用doReleaseShared()唤醒阻塞的节点

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        // 判断队列是否是空队列
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            // 头节点的状态为 signal,说明后继节点没有被唤醒过
            if (ws == Node.SIGNAL) {
                // cas 设置头节点的状态为 0,设置失败继续自旋
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 唤醒后继节点
                unparkSuccessor(h);
            }
            // 如果有其他线程已经设置了头节点的状态,重新设置为 PROPAGATE 传播属性
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        // 条件不成立说明被唤醒的节点非常积极,直接将自己设置为了新的head,
        // 此时唤醒它的节点(前驱)执行 h == head 不成立,所以不会跳出循环,会继续唤醒新的 head 节点的后继节点
        if (h == head)
            break;
    }
}

以上就是Java CountDownLatch的源码硬核解析的详细内容,更多关于Java CountDownLatch的资料请关注我们其它相关文章!

(0)

相关推荐

  • Java多线程CountDownLatch的实现

    介绍 CountDownLatch是一个同步辅助类,它允许一个或多个线程一直等待直到其他线程执行完毕才开始执行. 用给定的计数初始化CountDownLatch,其含义是要被等待执行完的线程个数. 每次调用CountDown(),计数减1 主程序执行到await()函数会阻塞等待线程的执行,直到计数为0 场景 1:多线程读取批量文件, 并且读取完成之后汇总处理 2:多线程读取Excel多个sheet,读取完成之后获取汇总获取的结果 3:多个人一起一起来吃饭,主人等待客人到来,客人一个个从不同地方

  • Java并发编程同步器CountDownLatch

    CountDownLatch 在日常开发中经常会遇到需要在主线程中开启多个线程去并行执行任务,并且主线程需要等待所有子线程执行完毕后再进行汇总的场景.在 CountDownLatch 出现之前般都使用线程的join()方法来实现这一点,但是 join 方法不够灵活,不能够满足不同场景的需要,所以 JDK 开发组提供了 CountDownLatch 这个类,我们前面介绍的例子使用 CoumtDownLatch 会更优雅. 使用CountDownLatch 的代码如下: package LockSu

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

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

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

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

  • Java CountDownLatch的源码硬核解析

    目录 前言 介绍和使用 例子 概述 实现思路 源码解析 类结构图 await() 实现原理 countDown()实现原理 前言 对于并发执行,Java中的CountDownLatch是一个重要的类,简单理解, CountDownLatch中count down是倒数的意思,latch则是“门闩”的含义.在数量倒数到0的时候,打开“门闩”, 一起走,否则都等待在“门闩”的地方. 为了更好的理解CountDownLatch这个类,本文通过例子和源码带领大家深入解析这个类的原理. 介绍和使用 例子

  • Java Lambda 表达式源码解析

    Java Lambda 源码分析 问题: Lambda 表达式是什么?JVM 内部究竟是如何实现 Lambda 表达式的?为什么要这样实现? 一.基本概念 1.Lambda 表达式 下面的例子中,() -> System.out.println("1") 就是一个 Lambda 表达式.Java 8 中每一个 Lambda 表达式必须有一个函数式接口与之对应.Lambda 表达式就是函数式接口的一个实现. @Test public void test0() { Runnable

  • java中CopyOnWriteArrayList源码解析

    目录 简介 继承体系 源码解析 属性 构造方法 add(Ee)方法 add(intindex,Eelement)方法 addIfAbsent(Ee)方法 get(intindex) remove(intindex)方法 size()方法 提问 总结 简介 CopyOnWriteArrayList是ArrayList的线程安全版本,内部也是通过数组实现,每次对数组的修改都完全拷贝一份新的数组来修改,修改完了再替换掉老数组,这样保证了只阻塞写操作,不阻塞读操作,实现读写分离. 继承体系 public

  • Java OkHttp框架源码深入解析

    目录 1.OkHttp发起网络请求 可以通过OkHttpClient发起一个网络请求 通过Retrofit发起一个OkHttp请求 2.OkHttp的连接器 1.OkHttp发起网络请求 可以通过OkHttpClient发起一个网络请求 //创建一个Client,相当于打开一个浏览器 OkHttpClient okHttpClient = new OkHttpClient.Builder().build(); //创建一个请求. Request request = new Request.Bui

  • Java struts2请求源码分析案例详解

    Struts2是Struts社区和WebWork社区的共同成果,我们甚至可以说,Struts2是WebWork的升级版,他采用的正是WebWork的核心,所以,Struts2并不是一个不成熟的产品,相反,构建在WebWork基础之上的Struts2是一个运行稳定.性能优异.设计成熟的WEB框架. 我这里的struts2源码是从官网下载的一个最新的struts-2.3.15.1-src.zip,将其解压即可.里面的目录页文件非常的多,我们只需要定位到struts-2.3.15.1\src\core

  • Java并发 结合源码分析AQS原理

    前言: 如果说J.U.C包下的核心是什么?那我想答案只有一个就是AQS.那么AQS是什么呢?接下来让我们一起揭开AQS的神秘面纱 AQS是什么? AQS是AbstractQueuedSynchronizer的简称.为什么说它是核心呢?是因为它提供了一个基于FIFO的队列和state变量来构建锁和其他同步装置的基础框架.下面是其底层的数据结构. AQS的特点 1.其内使用Node实现FIFO(FirstInFirstOut)队列.可用于构建锁或者其他同步装置的基础框架 2.且利用了一个int类表示

  • 深入剖析Java ReentrantLock的源码

    目录 1. ReentrantLock的使用 2. ReentrantLock类结构 3. ReentrantLock源码解析 3.1 ReentrantLock构造方法 3.2 非公平锁源码 3.3 公平锁源码 4. 总结 ReentrantLock和Synchronized都是Java开发中最常用的锁,与Synchronized这种JVM内置锁不同的是,ReentrantLock提供了更丰富的语义.可以创建公平锁或非公平锁.响应中断.超时等待.按条件唤醒等.在某些场景下,使用Reentran

  • java 中Buffer源码的分析

    java 中Buffer源码的分析 Buffer Buffer的类图如下: 除了Boolean,其他基本数据类型都有对应的Buffer,但是只有ByteBuffer才能和Channel交互.只有ByteBuffer才能产生Direct的buffer,其他数据类型的Buffer只能产生Heap类型的Buffer.ByteBuffer可以产生其他数据类型的视图Buffer,如果ByteBuffer本身是Direct的,则产生的各视图Buffer也是Direct的. Direct和Heap类型Buff

  • Java集合框架源码分析之LinkedHashMap详解

    LinkedHashMap简介 LinkedHashMap是HashMap的子类,与HashMap有着同样的存储结构,但它加入了一个双向链表的头结点,将所有put到LinkedHashmap的节点一一串成了一个双向循环链表,因此它保留了节点插入的顺序,可以使节点的输出顺序与输入顺序相同. LinkedHashMap可以用来实现LRU算法(这会在下面的源码中进行分析). LinkedHashMap同样是非线程安全的,只在单线程环境下使用. LinkedHashMap源码剖析 LinkedHashM

  • Java 中模仿源码自定义ArrayList

    Java 中模仿源码自定义ArrayList 最近看了下ArrayList的源码,抽空根据ArrayList的底层结构写了一个功能简单无泛型的自定义ArrayLsit,帮助自己更好理解ArrayList:,其实现的底层数据结构为数Object组,代码如下: /** * 自己实现一个ArrayList * */ public class MyArrayList { private Object[] elementData; private int size; public int size(){

随机推荐