专栏首页JAVA葵花宝典『并发包入坑指北』之向大佬汇报任务

『并发包入坑指北』之向大佬汇报任务

前言

目录如下:

在面试过程中聊到并发相关的内容时,不少面试官都喜欢问这类问题:

当 N 个线程同时完成某项任务时,如何知道他们都已经执行完毕了。

这也是本次讨论的话题之一,所以本篇为『并发包入坑指北』的第二篇;来聊聊常见的并发工具。

自己实现

其实这类问题的核心论点都是:如何在一个线程中得知其他线程是否执行完毕。

假设现在有 3 个线程在运行,需要在主线程中得知他们的运行结果;可以分为以下几步:

  • 定义一个计数器为 3。
  • 每个线程完成任务后计数减一。
  • 一旦计数器减为 0 则通知等待的线程。

所以也很容易想到可以利用等待通知机制来实现,和上文的『并发包入坑指北』之阻塞队列的类似。

按照这个思路自定义了一个 MultipleThreadCountDownKit 工具,构造函数如下:

考虑到并发的前提,这个计数器自然需要保证线程安全,所以采用了 AtomicInteger

所以在初始化时需要根据线程数量来构建对象。

计数器减一

当其中一个业务线程完成后需要将这个计数器减一,直到减为0为止。

/**
     * 线程完成后计数 -1
     */
    public void countDown(){

        if (counter.get() <= 0){
            return;
        }

        int count = this.counter.decrementAndGet();
        if (count < 0){
            throw new RuntimeException("concurrent error") ;
        }

        if (count == 0){
            synchronized (notify){
                notify.notify();
            }
        }

    }

利用 counter.decrementAndGet() 来保证多线程的原子性,当减为 0 时则利用等待通知机制来 notify 其他线程。

等待所有线程完成

而需要知道业务线程执行完毕的其他线程则需要在未完成之前一直处于等待状态,直到上文提到的在计数器变为 0 时得到通知。

/**
     * 等待所有的线程完成
     * @throws InterruptedException
     */
    public void await() throws InterruptedException {
        synchronized (notify){
            while (counter.get() > 0){
                notify.wait();
            }

            if (notifyListen != null){
                notifyListen.notifyListen();
            }

        }
    }

原理也很简单,一旦计数器还存在时则会利用 notify 对象进行等待,直到被业务线程唤醒。

同时这里新增了一个通知接口可以自定义实现唤醒后的一些业务逻辑,后文会做演示。

并发测试

主要就是这两个函数,下面来做一个演示。

  • 初始化了三个计数器的并发工具 MultipleThreadCountDownKit
  • 创建了三个线程分别执行业务逻辑,完毕后执行 countDown()
  • 线程 3 休眠了 2s 用于模拟业务耗时。
  • 主线程执行 await() 等待他们三个线程执行完毕。

通过执行结果可以看出主线程会等待最后一个线程完成后才会退出;从而达到了主线程等待其余线程的效果。

MultipleThreadCountDownKit multipleThreadKit = new MultipleThreadCountDownKit(3);
    multipleThreadKit.setNotify(() -> LOGGER.info("三个线程完成了任务"));

也可以在初始化的时候指定一个回调接口,用于接收业务线程执行完毕后的通知。

当然和在主线程中执行这段逻辑效果是一样的(和执行 await() 方法处于同一个线程)。

CountDownLatch

当然我们自己实现的代码没有经过大量生产环境的验证,所以主要的目的还是尝试窥探官方的实现原理。

所以我们现在来看看 juc 下的 CountDownLatch 是如何实现的。

通过构造函数会发现有一个 内部类 Sync,他是继承于 AbstractQueuedSynchronizer ;这是 Java 并发包中的基础框架,都可以单独拿来讲了,所以这次重点不是它,今后我们再着重介绍。

这里就可以把他简单理解为提供了和上文类似的一个计数器及线程通知工具就行了。

countDown

其实他的核心逻辑和我们自己实现的区别不大。

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

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

利用这个内部类的 releaseShared 方法,我们可以理解为他想要将计数器减一。

看到这里有没有似曾相识的感觉。

没错,在 JDK1.7 中的 AtomicInteger 自减就是这样实现的(利用 CAS 保证了线程安全)。

只是一旦计数器减为 0 时则会执行 doReleaseShared 唤醒其他的线程。

这里我们只需要关心红框部分(其他的暂时不用关心,这里涉及到了 AQS 中的队列相关),最终会调用 LockSupport.unpark 来唤醒线程;就相当于上文调用 object.notify()

所以其实本质上还是相同的。

await

其中的 await() 也是借用 Sync 对象的方法实现的。

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

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        //判断计数器是否还未完成    
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

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

一旦还存在未完成的线程时,则会调用 doAcquireSharedInterruptibly 进入阻塞状态。

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

同样的由于这也是 AQS 中的方法,我们只需要关心红框部分;其实最终就是调用了 LockSupport.park 方法,也就相当于执行了 object.wait()

  • 所有的业务线程执行完毕后会在计数器减为 0 时调用 LockSupport.unpark 来唤醒线程。
  • 等待线程一旦计数器 > 0 时则会利用 LockSupport.park 来等待唤醒。

这样整个流程也就串起来了,它的使用方法也和上文的类似。

就不做过多介绍了。

实际案例

同样的来看一个实际案例。

在上一篇《一次分表踩坑实践的探讨》提到了对于全表扫描的情况下,需要利用多线程来提高查询效率。

比如我们这里分为了 64 张表,计划利用 8 个线程来分别处理这些表的数据,伪代码如下:

CountDownLatch count = new CountDownLatch(64);
ConcurrentHashMap total = new ConcurrentHashMap();
for(Integer i=0;i<=63;i++){
    executor.execute(new Runnable(){
        @Override
        public void run(){
            List value = queryTable(i);
            total.put(value,NULL);
            count.countDown();
        }
    }) ;

}

count.await();
System.out.println("查询完毕");

这样就可以实现所有数据都查询完毕后再做统一汇总;代码挺简单,也好理解(当然也可以使用线程池的 API)。

总结

CountDownLatch 算是 juc 中一个高频使用的工具,学会和理解他的使用会帮助我们更容易编写并发应用。

文中涉及到的源码:

https://github.com/crossoverJie/JCSprout/blob/master/src/main/java/com/crossoverjie/concurrent/communication/MultipleThreadCountDownKit.java

本文分享自微信公众号 - JAVA葵花宝典(Javakhbd)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-10-23

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 36 个JS 面试题为你助力金九银十(面试必读)

    在现代js中,let&const是创建变量的不同方式。 在早期的js中,咱们使用var关键字来创建变量。 let&const关键字是在ES6版本中引入的,其目的...

    Javanx
  • Node.js 服务 Docker 容器化应用实践

    本篇不会讲解 Docker 命令的使用、安装等,因为在之前一篇文章 【一文零基础教你学会 Docker 入门到实践

    五月君
  • 浅谈MVC--Node中如何使用ORM?

    在正常的开发中,大部分都会使用MVC为主要的系统架构模式。而Model一般包含了复杂的业务逻辑以及数据逻辑,因为Model中逻辑的复杂度,所以我们有必要降低系统...

    逆月翎
  • 防抖、节流、call、bind、apply、new、柯里化实现

    前端迷
  • 4399AT UI自动化CI与CD

    看完了这三者的区别,目前4399AT是达到了CI和CD(持续交付)阶段,接下来我们说下4399AT的CI与CD; 以下均以安卓项目APP为基础来进行讲解: 软件...

    八音弦
  • 更轻量级的 V8 引擎[每日前端夜话0xC8]

    作者:Mythri Alle, Dan Elphick, and Ross McIlroy

    疯狂的技术宅
  • 11 种在大多数教程中找不到的JavaScript技巧

    当我开始学习JavaScript时,我把我在别人的代码、code challenge网站以及我使用的教程之外的任何地方发现的每一个节省时间的技巧都列了一个清单。

    Javanx
  • Flutter完整开发实战详解(十八、 神奇的ScrollPhysics与Simulation)

    作为系列文章的第十八篇,本篇将通过 ScrollPhysics 和 Simulation ,带你深入走进 Flutter 的滑动新世界,为你打开 Flutter...

    恋猫
  • 军备竞赛:DDoS攻击防护体系构建

    ? 前言:DDoS攻击(Distributed Denial of Service,分布式拒绝服务攻击)的历史可以追溯到1996年(还记得经典的Ping of...

    腾讯技术工程官方号
  • JavaScript是如何工作的:引擎,运行时和调用堆栈的概述!

    本文是旨在深入研究JavaScript及其实际工作原理的系列文章中的第一篇:我们认为通过了解JavaScript的构建块以及它们是如何工作的,将能够编写更好的代...

    Javanx

扫码关注云+社区

领取腾讯云代金券