java高并发系列 - 第16天：JUC中等待多线程完成的工具类CountDownLatch，必备技

时间  2019-11-06
标签 java 并发 系列 16天 juc 等待 多线程 完成 工具 countdownlatch 必备 栏目 Java 繁體版
原文   原文链接

这是java高并发系列第16篇文章。java

本篇内容

  1. 介绍CountDownLatch及使用场景
  2. 提供几个示例介绍CountDownLatch的使用
  3. 手写一个并行处理任务的工具类

假若有这样一个需求，当咱们须要解析一个Excel里多个sheet的数据时，能够考虑使用多线程，每一个线程解析一个sheet里的数据，等到全部的sheet都解析完以后，程序须要统计解析总耗时。分析一下：解析每一个sheet耗时可能不同，总耗时就是最长耗时的那个操做。spring

咱们可以想到的最简单的作法是使用join，代码以下：微信

package com.itsoku.chat13;

import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo1 {

    public static class T extends Thread {
        //休眠时间（秒）
        int sleepSeconds;

        public T(String name, int sleepSeconds) {
            super(name);
            this.sleepSeconds = sleepSeconds;
        }

        @Override
        public void run() {
            Thread ct = Thread.currentThread();
            long startTime = System.currentTimeMillis();
            System.out.println(startTime + "," + ct.getName() + ",开始处理!");
            try {
                //模拟耗时操做，休眠sleepSeconds秒
                TimeUnit.SECONDS.sleep(this.sleepSeconds);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            long endTime = System.currentTimeMillis();
            System.out.println(endTime + "," + ct.getName() + ",处理完毕,耗时:" + (endTime - startTime));
        }
    }

    public static void main(String[] args) throws InterruptedException {
        long starTime = System.currentTimeMillis();
        T t1 = new T("解析sheet1线程", 2);
        t1.start();

        T t2 = new T("解析sheet2线程", 5);
        t2.start();

        t1.join();
        t2.join();
        long endTime = System.currentTimeMillis();
        System.out.println("总耗时:" + (endTime - starTime));

    }
}

输出：多线程

1563767560271,解析sheet1线程,开始处理!
1563767560272,解析sheet2线程,开始处理!
1563767562273,解析sheet1线程,处理完毕,耗时:2002
1563767565274,解析sheet2线程,处理完毕,耗时:5002
总耗时:5005

代码中启动了2个解析sheet的线程，第一个耗时2秒，第二个耗时5秒，最终结果中总耗时：5秒。上面的关键技术点是线程的join()方法，此方法会让当前线程等待被调用的线程完成以后才能继续。能够看一下join的源码，内部实际上是在synchronized方法中调用了线程的wait方法，最后被调用的线程执行完毕以后，由jvm自动调用其notifyAll()方法，唤醒全部等待中的线程。这个notifyAll()方法是由jvm内部自动调用的，jdk源码中是看不到的，须要看jvm源码，有兴趣的同窗能够去查一下。因此JDK不推荐在线程上调用wait、notify、notifyAll方法。并发

而在JDK1.5以后的并发包中提供的CountDownLatch也能够实现join的这个功能。jvm

CountDownLatch介绍

CountDownLatch称之为闭锁，它可使一个或一批线程在闭锁上等待，等到其余线程执行完相应操做后，闭锁打开，这些等待的线程才能够继续执行。确切的说，闭锁在内部维护了一个倒计数器。经过该计数器的值来决定闭锁的状态，从而决定是否容许等待的线程继续执行。ide

经常使用方法：高并发

public CountDownLatch(int count)：构造方法，count表示计数器的值，不能小于0，否者会报异常。工具

public void await() throws InterruptedException：调用await()会让当前线程等待，直到计数器为0的时候，方法才会返回，此方法会响应线程中断操做。this

public boolean await(long timeout, TimeUnit unit) throws InterruptedException：限时等待，在超时以前，计数器变为了0，方法返回true，否者直到超时，返回false，此方法会响应线程中断操做。

public void countDown()：让计数器减1

CountDownLatch使用步骤：

  1. 建立CountDownLatch对象
  2. 调用其实例方法await()，让当前线程等待
  3. 调用countDown()方法，让计数器减1
  4. 当计数器变为0的时候，await()方法会返回

示例1：一个简单的示例

咱们使用CountDownLatch来完成上面示例中使用join实现的功能，代码以下：

package com.itsoku.chat13;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo2 {

    public static class T extends Thread {
        //休眠时间（秒）
        int sleepSeconds;
        CountDownLatch countDownLatch;

        public T(String name, int sleepSeconds, CountDownLatch countDownLatch) {
            super(name);
            this.sleepSeconds = sleepSeconds;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            Thread ct = Thread.currentThread();
            long startTime = System.currentTimeMillis();
            System.out.println(startTime + "," + ct.getName() + ",开始处理!");
            try {
                //模拟耗时操做，休眠sleepSeconds秒
                TimeUnit.SECONDS.sleep(this.sleepSeconds);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
            long endTime = System.currentTimeMillis();
            System.out.println(endTime + "," + ct.getName() + ",处理完毕,耗时:" + (endTime - startTime));
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 start!");
        CountDownLatch countDownLatch = new CountDownLatch(2);

        long starTime = System.currentTimeMillis();
        T t1 = new T("解析sheet1线程", 2, countDownLatch);
        t1.start();

        T t2 = new T("解析sheet2线程", 5, countDownLatch);
        t2.start();

        countDownLatch.await();
        System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 end!");
        long endTime = System.currentTimeMillis();
        System.out.println("总耗时:" + (endTime - starTime));

    }
}

输出：

1563767580511,main线程 start!
1563767580513,解析sheet1线程,开始处理!
1563767580513,解析sheet2线程,开始处理!
1563767582515,解析sheet1线程,处理完毕,耗时:2002
1563767585515,解析sheet2线程,处理完毕,耗时:5002
1563767585515,main线程 end!
总耗时:5003

从结果中看出，效果和join实现的效果同样，代码中建立了计数器为2的CountDownLatch，主线程中调用countDownLatch.await();会让主线程等待，t一、t2线程中模拟执行耗时操做，最终在finally中调用了countDownLatch.countDown();,此方法每调用一次，CountDownLatch内部计数器会减1，当计数器变为0的时候，主线程中的await()会返回，而后继续执行。注意：上面的countDown()这个是必需要执行的方法，因此放在finally中执行。

示例2：等待指定的时间

仍是上面的示例，2个线程解析2个sheet，主线程等待2个sheet解析完成。主线程说，我等待2秒，大家仍是没法处理完成，就不等待了，直接返回。以下代码：

package com.itsoku.chat13;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo3 {

    public static class T extends Thread {
        //休眠时间（秒）
        int sleepSeconds;
        CountDownLatch countDownLatch;

        public T(String name, int sleepSeconds, CountDownLatch countDownLatch) {
            super(name);
            this.sleepSeconds = sleepSeconds;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            Thread ct = Thread.currentThread();
            long startTime = System.currentTimeMillis();
            System.out.println(startTime + "," + ct.getName() + ",开始处理!");
            try {
                //模拟耗时操做，休眠sleepSeconds秒
                TimeUnit.SECONDS.sleep(this.sleepSeconds);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
            long endTime = System.currentTimeMillis();
            System.out.println(endTime + "," + ct.getName() + ",处理完毕,耗时:" + (endTime - startTime));
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 start!");
        CountDownLatch countDownLatch = new CountDownLatch(2);

        long starTime = System.currentTimeMillis();
        T t1 = new T("解析sheet1线程", 2, countDownLatch);
        t1.start();

        T t2 = new T("解析sheet2线程", 5, countDownLatch);
        t2.start();

        boolean result = countDownLatch.await(2, TimeUnit.SECONDS);

        System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 end!");
        long endTime = System.currentTimeMillis();
        System.out.println("主线程耗时:" + (endTime - starTime) + ",result:" + result);

    }
}

输出：

1563767637316,main线程 start!
1563767637320,解析sheet1线程,开始处理!
1563767637320,解析sheet2线程,开始处理!
1563767639321,解析sheet1线程,处理完毕,耗时:2001
1563767639322,main线程 end!
主线程耗时:2004,result:false
1563767642322,解析sheet2线程,处理完毕,耗时:5002

从输出结果中能够看出，线程2耗时了5秒，主线程耗时了2秒，主线程中调用countDownLatch.await(2, TimeUnit.SECONDS);，表示最多等2秒，无论计数器是否为0，await方法都会返回，若等待时间内，计数器变为0了，当即返回true，不然超时后返回false。

示例3：2个CountDown结合使用的示例

有3我的参见跑步比赛，须要先等指令员发指令枪后才能开跑，全部人都跑完以后，指令员喊一声，你们跑完了。

示例代码：

package com.itsoku.chat13;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo4 {

    public static class T extends Thread {
        //跑步耗时（秒）
        int runCostSeconds;
        CountDownLatch commanderCd;
        CountDownLatch countDown;

        public T(String name, int runCostSeconds, CountDownLatch commanderCd, CountDownLatch countDown) {
            super(name);
            this.runCostSeconds = runCostSeconds;
            this.commanderCd = commanderCd;
            this.countDown = countDown;
        }

        @Override
        public void run() {
            //等待指令员枪响
            try {
                commanderCd.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            Thread ct = Thread.currentThread();
            long startTime = System.currentTimeMillis();
            System.out.println(startTime + "," + ct.getName() + ",开始跑!");
            try {
                //模拟耗时操做，休眠runCostSeconds秒
                TimeUnit.SECONDS.sleep(this.runCostSeconds);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDown.countDown();
            }
            long endTime = System.currentTimeMillis();
            System.out.println(endTime + "," + ct.getName() + ",跑步结束,耗时:" + (endTime - startTime));
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 start!");
        CountDownLatch commanderCd = new CountDownLatch(1);
        CountDownLatch countDownLatch = new CountDownLatch(3);

        long starTime = System.currentTimeMillis();
        T t1 = new T("小张", 2, commanderCd, countDownLatch);
        t1.start();

        T t2 = new T("小李", 5, commanderCd, countDownLatch);
        t2.start();

        T t3 = new T("路人甲", 10, commanderCd, countDownLatch);
        t3.start();

        //主线程休眠5秒,模拟指令员准备发枪耗时操做
        TimeUnit.SECONDS.sleep(5);
        System.out.println(System.currentTimeMillis() + ",枪响了，你们开始跑");
        commanderCd.countDown();

        countDownLatch.await();
        long endTime = System.currentTimeMillis();
        System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "全部人跑完了，主线程耗时:" + (endTime - starTime));

    }
}

输出：

1563767691087,main线程 start!
1563767696092,枪响了，你们开始跑
1563767696092,小张,开始跑!
1563767696092,小李,开始跑!
1563767696092,路人甲,开始跑!
1563767698093,小张,跑步结束,耗时:2001
1563767701093,小李,跑步结束,耗时:5001
1563767706093,路人甲,跑步结束,耗时:10001
1563767706093,main全部人跑完了，主线程耗时:15004

代码中，t一、t二、t3启动以后，都阻塞在commanderCd.await();，主线程模拟发枪准备操做耗时5秒，而后调用commanderCd.countDown();模拟发枪操做，此方法被调用之后，阻塞在commanderCd.await();的3个线程会向下执行。主线程调用countDownLatch.await();以后进行等待，每一个人跑完以后，调用countDown.countDown();通知一下countDownLatch让计数器减1，最后3我的都跑完了，主线程从countDownLatch.await();返回继续向下执行。

手写一个并行处理任务的工具类

package com.itsoku.chat13;

import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class TaskDisposeUtils {
    //并行线程数
    public static final int POOL_SIZE;

    static {
        POOL_SIZE = Integer.max(Runtime.getRuntime().availableProcessors(), 5);
    }

    /**
     * 并行处理，并等待结束
     *
     * @param taskList 任务列表
     * @param consumer 消费者
     * @param <T>
     * @throws InterruptedException
     */
    public static <T> void dispose(List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        dispose(true, POOL_SIZE, taskList, consumer);
    }

    /**
     * 并行处理，并等待结束
     *
     * @param moreThread 是否多线程执行
     * @param poolSize   线程池大小
     * @param taskList   任务列表
     * @param consumer   消费者
     * @param <T>
     * @throws InterruptedException
     */
    public static <T> void dispose(boolean moreThread, int poolSize, List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        if (CollectionUtils.isEmpty(taskList)) {
            return;
        }
        if (moreThread && poolSize > 1) {
            poolSize = Math.min(poolSize, taskList.size());
            ExecutorService executorService = null;
            try {
                executorService = Executors.newFixedThreadPool(poolSize);
                CountDownLatch countDownLatch = new CountDownLatch(taskList.size());
                for (T item : taskList) {
                    executorService.execute(() -> {
                        try {
                            consumer.accept(item);
                        } finally {
                            countDownLatch.countDown();
                        }
                    });
                }
                countDownLatch.await();
            } finally {
                if (executorService != null) {
                    executorService.shutdown();
                }
            }
        } else {
            for (T item : taskList) {
                consumer.accept(item);
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        //生成1-10的10个数字，放在list中，至关于10个任务
        List<Integer> list = Stream.iterate(1, a -> a + 1).limit(10).collect(Collectors.toList());
        //启动多线程处理list中的数据，每一个任务休眠时间为list中的数值
        TaskDisposeUtils.dispose(list, item -> {
            try {
                long startTime = System.currentTimeMillis();
                TimeUnit.SECONDS.sleep(item);
                long endTime = System.currentTimeMillis();

                System.out.println(System.currentTimeMillis() + ",任务" + item + "执行完毕，耗时:" + (endTime - startTime));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        //上面全部任务处理完毕完毕以后，程序才能继续
        System.out.println(list + "中的任务都处理完毕!");
    }
}

运行代码输出：

1563769828130,任务1执行完毕，耗时:1000
1563769829130,任务2执行完毕，耗时:2000
1563769830131,任务3执行完毕，耗时:3001
1563769831131,任务4执行完毕，耗时:4001
1563769832131,任务5执行完毕，耗时:5001
1563769833130,任务6执行完毕，耗时:6000
1563769834131,任务7执行完毕，耗时:7001
1563769835131,任务8执行完毕，耗时:8001
1563769837131,任务9执行完毕，耗时:9001
1563769839131,任务10执行完毕，耗时:10001
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]中的任务都处理完毕!

TaskDisposeUtils是一个并行处理的工具类，能够传入n个任务内部使用线程池进行处理，等待全部任务都处理完成以后，方法才会返回。好比咱们发送短信，系统中有1万条短信，咱们使用上面的工具，每次取100条并行发送，待100个都处理完毕以后，再取一批按照一样的逻辑发送。

java高并发系列连载中，总计估计会有四五十篇文章，能够关注公众号：javacode2018，获取最新文章。

java高并发系列 - 第16天：JUC中等待多线程完成的工具类CountDownLatch，必备技

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程