热门标签 | HotTags
当前位置:  开发笔记 > 运维 > 正文

Java并发编程之CountDownLatch源码解析

这篇文章主要介绍了Java并发编程之CountDownLatch源码解析,文中有非常详细的代码示例,对正在学习java并发编程的小伙伴们有很好的帮助,需要的朋友可以参考下

一、前言

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

二、使用

  • 一个线程等待其它线程执行完再继续执行
	......
	CountDownLatch cdl = new CountDownLatch(10);
	ExecutorService es = Executors.newFixedThreadPool(10);
	for (int i = 0; i <10; i++) {
		es.execute(() -> {
			//do something
			cdl.countDown();
		});
	}
	cdl.await();
	......
  • 实现类似CyclicBarrier的功能,先await,再countDown
	......
        CountDownLatch cdl = new CountDownLatch(1);
        ExecutorService es = Executors.newFixedThreadPool(10);
        for (int i = 0; i <10; i++) {
            es.execute(() -> {
                cdl.await();
                //do something
            });
        }
        Thread.sleep(10000L);
        cdl.countDown();
        ......

三、源码分析

CountDownLatch的结构和ReentrantLock、Semaphore的结构类似,也是使用的内部类Sync继承AQS的方式,并且重写了tryAcquireShared和tryReleaseShared方法。

还是首先来看构造函数:

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

需要传入一个大于0的count,代表CountDownLatch计数器的初始值,通过Sync的构造函数最终赋值给父类AQS的state字段。可一个看到这个state字段用法多多,在ReentrantLock中使用0和1来标识锁的状态,Semaphore中用来标识信号量,此处又用来表示计数器。

CountDownLatch要通过await方法完成阻塞,先来看看这个方法是如何实现的:

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

调用的是sync的acquireSharedInterruptibly方法,该方法定义在AQS中,Semaphore也调用的这个方法:

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

这个方法的逻辑前面在解析SemaPhore的时候细说过了,这里不再赘述,主要就是两个方法的调用,先通过tryAcquireShared方法尝试获取"许可",返回值代表此次获取后的剩余量,如果大于等于0表示获取成功,否则表示失败。如果失败,那么就会进入doAcquireSharedInterruptibly方法执行入队阻塞的逻辑。这里我们主要到CountDownLatch中看看tryAcquireShared方法的实现:

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

和Semaphore的实现中每次将state减去requires不同,这里直接判断state是否为0,如果为0那么返回1,表示获取"许可"成功;如果不为0,表示失败,则需要入队阻塞。从这个tryAcquireShared方法就能看出CountDownLatch的逻辑了:等到state变为了0,那么所有线程都能获取运行许可。

那么我们接下来来到countDown方法:

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

调用的是sync的releaseShared方法,该方法定义在父类AQS中,Semaphore使用的也是这个方法:

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
        	//当state从非
            doReleaseShared();
            return true;
        }
        return false;
    }

前面提到了CountDownLatch也重写了tryReleaseShared方法:

protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                	//如果state等于0了直接返回false
                	//保证在并发情况下,最多只会有一个线程返回true
                	//也包括调用countDown的次数超过state的初始值
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                	//如果返回true,表示state从非0变为了0
                	//那么后续需要唤醒阻塞线程
                    return nextc == 0;
            }
        }

Semaphore在释放信号量的时候,是将获取的许可归还到state中,但是CountDownLatch没有获取许可的逻辑(获取许可的时候是判断state是否等于0),所以在countDown的时候没有释放的逻辑,就是将state减1,然后根据state减1之后的值是否为0判断release是否成功,如果state本来大于0,经过减1之后变为了0,那么返回true。tryReleaseShared方法的返回值决定了后续需不需要调用doReleaseShared方法唤醒阻塞线程。

这里有个逻辑:如果state已经为0,那么返回false。这个主要应对两种情况:

  • 调用countDown的次数超过了state的初始值多
  • 线程并发调用的时候保证只有一个线程去完成阻塞线程的唤醒操作

可以看到CountDownLatch没有锁的概念,countDown方法可以被一个线程重复调用,只需要对state做reduce操作,而不用关心是谁做的reduce。如果tryReleaseShared返回true,那么表示需要在后面进入doReleaseShared方法,该方法和Semaphore中调用的方法是同一个,主要是唤醒阻塞线程或者设置PROPAGAGE状态,这里也不再赘述~

阻塞线程被唤醒之后,会在doAcquireSharedInterruptibly方法中继续循环,虽然和Semaphore调用的是同样的方法,但是这里有不一样的地方,所以还是提一句。我们首先回到doAcquireSharedInterruptibly方法:

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) {
                	//如果head.next被unpark唤醒,说明此时state==0
                	//那么tryAcquireShared会返回1
                    int r = tryAcquireShared(arg);
                    //r==1
                    if (r >= 0) {
                    	//node节点被唤醒后,还会继续唤醒node.next
                    	//这样依次传递,因为在这里的r一定为1
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

当head.next线程被unpark唤醒后,会进入tryAcquireShared方法判断,由于此时state已经为0(只有当state变为0时,才会unpark唤醒线程),而前面提到了在CountDownLatch重写的tryAcquireShared中,如果state==0,那么会返回1,所以会进入setHeadAndPropagate方法:

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus <0 ||
            (h = head) == null || h.waitStatus <0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

该方法在Semaphore中详细介绍过,这里我们就站在CountDownLatch的角度来看看。其实很简单了,注意此时该方法的propagate参数值是1,那么就会进入到下面的if逻辑里,继续唤醒下一个node。当下一个node对应的线程被唤醒后,同样会进入setHeadAndPropagate方法,propagage同样为1,那么继续唤醒下一个node,就这样依次将整个CLH队列的节点都唤醒。

四、总结

如果单独把CountDownLatch拿出来看其实是很复杂的,只是CountDownLatch(包括Semaphore和ReentrantLock)都高度共用了AQS提供的一些方法,而这些方法在前面介绍Semaphore和ReentrantLock的时候已经详细分析过,所以到本文分析CountDownLatch的时候,只需要关注它内部类Sync重写的两个方法:tryAcquireShared和tryReleaseShared,也就是"获取许可"和"释放许可"的逻辑。

CountDownLatch在await的逻辑里,如果当前state的值大于0,那么会进入CLH队列进行阻塞等待unpark唤醒(或者中断唤醒);在countDown的逻辑里,就是简单的将state-1,如果一个线程把state从1减为0,那么该线程就会负责唤醒head.next节点,head.next节点被唤醒后,又会在setHeadAndPropagate方法中唤醒next.next节点,这样依次唤醒所有CLH队列中的阻塞节点。当然,如果线程被中断唤醒,那么也会进入cancelAcquire中进行无效节点的移除逻辑。

到此这篇关于Java并发编程之CountDownLatch源码解析的文章就介绍到这了,更多相关Java中CountDownLatch源码解析内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持!


推荐阅读
  • 在现代网络环境中,两台计算机之间的文件传输需求日益增长。传统的FTP和SSH方式虽然有效,但其配置复杂、步骤繁琐,难以满足快速且安全的传输需求。本文将介绍一种基于Go语言开发的新一代文件传输工具——Croc,它不仅简化了操作流程,还提供了强大的加密和跨平台支持。 ... [详细]
  • 网络运维工程师负责确保企业IT基础设施的稳定运行,保障业务连续性和数据安全。他们需要具备多种技能,包括搭建和维护网络环境、监控系统性能、处理突发事件等。本文将探讨网络运维工程师的职业前景及其平均薪酬水平。 ... [详细]
  • 本文详细介绍如何在Linux系统中配置SSH密钥对,以实现从一台主机到另一台主机的无密码登录。内容涵盖密钥对生成、公钥分发及权限设置等关键步骤。 ... [详细]
  • 对象自省自省在计算机编程领域里,是指在运行时判断一个对象的类型和能力。dir能够返回一个列表,列举了一个对象所拥有的属性和方法。my_list[ ... [详细]
  • Python 工具推荐 | PyHubWeekly 第二十一期:提升命令行体验的五大工具
    本期 PyHubWeekly 为大家精选了 GitHub 上五个优秀的 Python 工具,涵盖金融数据可视化、终端美化、国际化支持、图像增强和远程 Shell 环境配置。欢迎关注并参与项目。 ... [详细]
  • 并发编程 12—— 任务取消与关闭 之 shutdownNow 的局限性
    Java并发编程实践目录并发编程01——ThreadLocal并发编程02——ConcurrentHashMap并发编程03——阻塞队列和生产者-消费者模式并发编程04——闭锁Co ... [详细]
  • 本文将详细介绍如何在没有显示器的情况下,使用Raspberry Pi Imager为树莓派4B安装操作系统,并进行基本配置,包括设置SSH、WiFi连接以及更新软件源。 ... [详细]
  • 本文探讨了如何通过一系列技术手段提升Spring Boot项目的并发处理能力,解决生产环境中因慢请求导致的系统性能下降问题。 ... [详细]
  • 本文详细介绍如何通过设置SSH密钥来获取连接GitHub远程仓库的权限,包括生成密钥、添加到GitHub账户以及验证连接等步骤。 ... [详细]
  • 本文介绍如何配置SecureCRT以正确显示Linux终端的颜色,并解决中文显示问题。通过简单的步骤设置,可以显著提升使用体验。 ... [详细]
  • 最新计算机专业原创毕业设计参考选题都有源码+数据库是近期作品ling取参考你的选题刚好在下面有,有时间看到机会给您发1ssm资源循环利用2springboot校园考勤系统3ssm防 ... [详细]
  • CentOS 7.2 配置防火墙端口开放
    本文介绍如何在 CentOS 7.2 系统上配置防火墙以开放特定的服务端口,包括 FTP 服务的临时与永久开放方法,以及如何验证配置是否生效。 ... [详细]
  • 本指南详细介绍了如何在同一台计算机上配置多个GitHub账户,并使用不同的SSH密钥进行身份验证,确保每个账户的安全性和独立性。 ... [详细]
  • WinSCP: 跨Windows与Linux系统的高效文件传输解决方案
    本文详细介绍了一款名为WinSCP的开源图形化SFTP客户端,该工具支持SSH协议,适用于Windows操作系统,能够实现与Linux系统之间的文件传输。对于从事嵌入式开发的技术人员来说,掌握WinSCP的使用方法将极大提高工作效率。 ... [详细]
  • 利用SSH隧道实现外网对局域网机器的安全访问
    本文探讨了一种常见的网络配置问题及其解决方案,即如何在外网环境下安全地访问位于局域网内的计算机。特别介绍了使用SSH反向隧道技术来实现这一目标的具体步骤和注意事项。 ... [详细]
author-avatar
平凡黯淡_551
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有