ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 编程语言> 文章详细

Java高并发专题之16、JUC中的CountDownLatch

2022-05-04 20:00:59  阅读：129  来源： 互联网

标签：JUC Java 16 countDownLatch System 耗时 线程 CountDownLatch public


本篇内容

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

思考一个需求

假如有这样一个需求，当我们需要解析一个Excel里多个sheet的数据时，可以考虑使用多线程，每个线程解析一个sheet里的数据，等到所有的sheet都解析完之后，程序需要统计解析总耗时。分析一下：解析每个sheet耗时可能不一样，总耗时就是最长耗时的那个操作。

我们能够想到的最简单的做法是使用join，代码如下：

  1. package com.itsoku.chat13;
  3. import java.util.concurrent.TimeUnit;
  5. /**
  6.  * 微信公众号：javacode2018，获取年薪50万课程
  7.  */
  8. public class Demo1 {
  10.     public static class T extends Thread {
  11.         //休眠时间（秒）
  12.         int sleepSeconds;
  14.         public T(String name, int sleepSeconds) {
  15.             super(name);
  16.             this.sleepSeconds = sleepSeconds;
  17.         }
  19.         @Override
  20.         public void run() {
  21.             Thread ct = Thread.currentThread();
  22.             long startTime = System.currentTimeMillis();
  23.             System.out.println(startTime + "," + ct.getName() + ",开始处理!");
  24.             try {
  25.                 //模拟耗时操作，休眠sleepSeconds秒
  26.                 TimeUnit.SECONDS.sleep(this.sleepSeconds);
  27.             } catch (InterruptedException e) {
  28.                 e.printStackTrace();
  29.             }
  30.             long endTime = System.currentTimeMillis();
  31.             System.out.println(endTime + "," + ct.getName() + ",处理完毕,耗时:" + (endTime - startTime));
  32.         }
  33.     }
  35.     public static void main(String[] args) throws InterruptedException {
  36.         long starTime = System.currentTimeMillis();
  37.         T t1 = new T("解析sheet1线程", 2);
  38.         t1.start();
  40.         T t2 = new T("解析sheet2线程", 5);
  41.         t2.start();
  43.         t1.join();
  44.         t2.join();
  45.         long endTime = System.currentTimeMillis();
  46.         System.out.println("总耗时:" + (endTime - starTime));
  48.     }
  49. }

输出：

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

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

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

CountDownLatch介绍

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

常用方法：

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

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

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实现的功能，代码如下：

  1. package com.itsoku.chat13;
  3. import java.util.concurrent.CountDownLatch;
  4. import java.util.concurrent.TimeUnit;
  6. /**
  7.  * 微信公众号：javacode2018，获取年薪50万课程
  8.  */
  9. public class Demo2 {
  11.     public static class T extends Thread {
  12.         //休眠时间（秒）
  13.         int sleepSeconds;
  14.         CountDownLatch countDownLatch;
  16.         public T(String name, int sleepSeconds, CountDownLatch countDownLatch) {
  17.             super(name);
  18.             this.sleepSeconds = sleepSeconds;
  19.             this.countDownLatch = countDownLatch;
  20.         }
  22.         @Override
  23.         public void run() {
  24.             Thread ct = Thread.currentThread();
  25.             long startTime = System.currentTimeMillis();
  26.             System.out.println(startTime + "," + ct.getName() + ",开始处理!");
  27.             try {
  28.                 //模拟耗时操作，休眠sleepSeconds秒
  29.                 TimeUnit.SECONDS.sleep(this.sleepSeconds);
  30.             } catch (InterruptedException e) {
  31.                 e.printStackTrace();
  32.             } finally {
  33.                 countDownLatch.countDown();
  34.             }
  35.             long endTime = System.currentTimeMillis();
  36.             System.out.println(endTime + "," + ct.getName() + ",处理完毕,耗时:" + (endTime - startTime));
  37.         }
  38.     }
  40.     public static void main(String[] args) throws InterruptedException {
  41.         System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 start!");
  42.         CountDownLatch countDownLatch = new CountDownLatch(2);
  44.         long starTime = System.currentTimeMillis();
  45.         T t1 = new T("解析sheet1线程", 2, countDownLatch);
  46.         t1.start();
  48.         T t2 = new T("解析sheet2线程", 5, countDownLatch);
  49.         t2.start();
  51.         countDownLatch.await();
  52.         System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 end!");
  53.         long endTime = System.currentTimeMillis();
  54.         System.out.println("总耗时:" + (endTime - starTime));
  56.     }
  57. }

输出：

  1. 1563767580511,main线程 start!
  2. 1563767580513,解析sheet1线程,开始处理!
  3. 1563767580513,解析sheet2线程,开始处理!
  4. 1563767582515,解析sheet1线程,处理完毕,耗时:2002
  5. 1563767585515,解析sheet2线程,处理完毕,耗时:5002
  6. 1563767585515,main线程 end!
  7. 总耗时:5003

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

示例2：等待指定的时间

还是上面的示例，2个线程解析2个sheet，主线程等待2个sheet解析完成。主线程说，我等待2秒，你们还是无法处理完成，就不等待了，直接返回。如下代码：

  1. package com.itsoku.chat13;
  3. import java.util.concurrent.CountDownLatch;
  4. import java.util.concurrent.TimeUnit;
  6. /**
  7.  * 微信公众号：javacode2018，获取年薪50万课程
  8.  */
  9. public class Demo3 {
  11.     public static class T extends Thread {
  12.         //休眠时间（秒）
  13.         int sleepSeconds;
  14.         CountDownLatch countDownLatch;
  16.         public T(String name, int sleepSeconds, CountDownLatch countDownLatch) {
  17.             super(name);
  18.             this.sleepSeconds = sleepSeconds;
  19.             this.countDownLatch = countDownLatch;
  20.         }
  22.         @Override
  23.         public void run() {
  24.             Thread ct = Thread.currentThread();
  25.             long startTime = System.currentTimeMillis();
  26.             System.out.println(startTime + "," + ct.getName() + ",开始处理!");
  27.             try {
  28.                 //模拟耗时操作，休眠sleepSeconds秒
  29.                 TimeUnit.SECONDS.sleep(this.sleepSeconds);
  30.             } catch (InterruptedException e) {
  31.                 e.printStackTrace();
  32.             } finally {
  33.                 countDownLatch.countDown();
  34.             }
  35.             long endTime = System.currentTimeMillis();
  36.             System.out.println(endTime + "," + ct.getName() + ",处理完毕,耗时:" + (endTime - startTime));
  37.         }
  38.     }
  40.     public static void main(String[] args) throws InterruptedException {
  41.         System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 start!");
  42.         CountDownLatch countDownLatch = new CountDownLatch(2);
  44.         long starTime = System.currentTimeMillis();
  45.         T t1 = new T("解析sheet1线程", 2, countDownLatch);
  46.         t1.start();
  48.         T t2 = new T("解析sheet2线程", 5, countDownLatch);
  49.         t2.start();
  51.         boolean result = countDownLatch.await(2, TimeUnit.SECONDS);
  53.         System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 end!");
  54.         long endTime = System.currentTimeMillis();
  55.         System.out.println("主线程耗时:" + (endTime - starTime) + ",result:" + result);
  57.     }
  58. }

输出：

  1. 1563767637316,main线程 start!
  2. 1563767637320,解析sheet1线程,开始处理!
  3. 1563767637320,解析sheet2线程,开始处理!
  4. 1563767639321,解析sheet1线程,处理完毕,耗时:2001
  5. 1563767639322,main线程 end!
  6. 主线程耗时:2004,result:false
  7. 1563767642322,解析sheet2线程,处理完毕,耗时:5002

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

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

有3个人参见跑步比赛，需要先等指令员发指令枪后才能开跑，所有人都跑完之后，指令员喊一声，大家跑完了。

示例代码：

  1. package com.itsoku.chat13;
  3. import java.util.concurrent.CountDownLatch;
  4. import java.util.concurrent.TimeUnit;
  6. /**
  7.  * 微信公众号：javacode2018，获取年薪50万课程
  8.  */
  9. public class Demo4 {
  11.     public static class T extends Thread {
  12.         //跑步耗时（秒）
  13.         int runCostSeconds;
  14.         CountDownLatch commanderCd;
  15.         CountDownLatch countDown;
  17.         public T(String name, int runCostSeconds, CountDownLatch commanderCd, CountDownLatch countDown) {
  18.             super(name);
  19.             this.runCostSeconds = runCostSeconds;
  20.             this.commanderCd = commanderCd;
  21.             this.countDown = countDown;
  22.         }
  24.         @Override
  25.         public void run() {
  26.             //等待指令员枪响
  27.             try {
  28.                 commanderCd.await();
  29.             } catch (InterruptedException e) {
  30.                 e.printStackTrace();
  31.             }
  32.             Thread ct = Thread.currentThread();
  33.             long startTime = System.currentTimeMillis();
  34.             System.out.println(startTime + "," + ct.getName() + ",开始跑!");
  35.             try {
  36.                 //模拟耗时操作，休眠runCostSeconds秒
  37.                 TimeUnit.SECONDS.sleep(this.runCostSeconds);
  38.             } catch (InterruptedException e) {
  39.                 e.printStackTrace();
  40.             } finally {
  41.                 countDown.countDown();
  42.             }
  43.             long endTime = System.currentTimeMillis();
  44.             System.out.println(endTime + "," + ct.getName() + ",跑步结束,耗时:" + (endTime - startTime));
  45.         }
  46.     }
  48.     public static void main(String[] args) throws InterruptedException {
  49.         System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "线程 start!");
  50.         CountDownLatch commanderCd = new CountDownLatch(1);
  51.         CountDownLatch countDownLatch = new CountDownLatch(3);
  53.         long starTime = System.currentTimeMillis();
  54.         T t1 = new T("小张", 2, commanderCd, countDownLatch);
  55.         t1.start();
  57.         T t2 = new T("小李", 5, commanderCd, countDownLatch);
  58.         t2.start();
  60.         T t3 = new T("路人甲", 10, commanderCd, countDownLatch);
  61.         t3.start();
  63.         //主线程休眠5秒,模拟指令员准备发枪耗时操作
  64.         TimeUnit.SECONDS.sleep(5);
  65.         System.out.println(System.currentTimeMillis() + ",枪响了，大家开始跑");
  66.         commanderCd.countDown();
  68.         countDownLatch.await();
  69.         long endTime = System.currentTimeMillis();
  70.         System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + "所有人跑完了，主线程耗时:" + (endTime - starTime));
  72.     }
  73. }

输出：

  1. 1563767691087,main线程 start!
  2. 1563767696092,枪响了，大家开始跑
  3. 1563767696092,小张,开始跑!
  4. 1563767696092,小李,开始跑!
  5. 1563767696092,路人甲,开始跑!
  6. 1563767698093,小张,跑步结束,耗时:2001
  7. 1563767701093,小李,跑步结束,耗时:5001
  8. 1563767706093,路人甲,跑步结束,耗时:10001
  9. 1563767706093,main所有人跑完了，主线程耗时:15004

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

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

  1. package com.itsoku.chat13;
  3. import org.springframework.util.CollectionUtils;
  5. import java.util.List;
  6. import java.util.concurrent.CountDownLatch;
  7. import java.util.concurrent.ExecutorService;
  8. import java.util.concurrent.Executors;
  9. import java.util.concurrent.TimeUnit;
  10. import java.util.function.Consumer;
  11. import java.util.stream.Collectors;
  12. import java.util.stream.Stream;
  14. /**
  15.  * 微信公众号：javacode2018，获取年薪50万课程
  16.  */
  17. public class TaskDisposeUtils {
  18.     //并行线程数
  19.     public static final int POOL_SIZE;
  21.     static {
  22.         POOL_SIZE = Integer.max(Runtime.getRuntime().availableProcessors(), 5);
  23.     }
  25.     /**
  26.      * 并行处理，并等待结束
  27.      *
  28.      * @param taskList 任务列表
  29.      * @param consumer 消费者
  30.      * @param <T>
  31.      * @throws InterruptedException
  32.      */
  33.     public static <T> void dispose(List<T> taskList, Consumer<T> consumer) throws InterruptedException {
  34.         dispose(true, POOL_SIZE, taskList, consumer);
  35.     }
  37.     /**
  38.      * 并行处理，并等待结束
  39.      *
  40.      * @param moreThread 是否多线程执行
  41.      * @param poolSize   线程池大小
  42.      * @param taskList   任务列表
  43.      * @param consumer   消费者
  44.      * @param <T>
  45.      * @throws InterruptedException
  46.      */
  47.     public static <T> void dispose(boolean moreThread, int poolSize, List<T> taskList, Consumer<T> consumer) throws InterruptedException {
  48.         if (CollectionUtils.isEmpty(taskList)) {
  49.             return;
  50.         }
  51.         if (moreThread && poolSize > 1) {
  52.             poolSize = Math.min(poolSize, taskList.size());
  53.             ExecutorService executorService = null;
  54.             try {
  55.                 executorService = Executors.newFixedThreadPool(poolSize);
  56.                 CountDownLatch countDownLatch = new CountDownLatch(taskList.size());
  57.                 for (T item : taskList) {
  58.                     executorService.execute(() -> {
  59.                         try {
  60.                             consumer.accept(item);
  61.                         } finally {
  62.                             countDownLatch.countDown();
  63.                         }
  64.                     });
  65.                 }
  66.                 countDownLatch.await();
  67.             } finally {
  68.                 if (executorService != null) {
  69.                     executorService.shutdown();
  70.                 }
  71.             }
  72.         } else {
  73.             for (T item : taskList) {
  74.                 consumer.accept(item);
  75.             }
  76.         }
  77.     }
  79.     public static void main(String[] args) throws InterruptedException {
  80.         //生成1-10的10个数字，放在list中，相当于10个任务
  81.         List<Integer> list = Stream.iterate(1, a -> a + 1).limit(10).collect(Collectors.toList());
  82.         //启动多线程处理list中的数据，每个任务休眠时间为list中的数值
  83.         TaskDisposeUtils.dispose(list, item -> {
  84.             try {
  85.                 long startTime = System.currentTimeMillis();
  86.                 TimeUnit.SECONDS.sleep(item);
  87.                 long endTime = System.currentTimeMillis();
  89.                 System.out.println(System.currentTimeMillis() + ",任务" + item + "执行完毕，耗时:" + (endTime - startTime));
  90.             } catch (InterruptedException e) {
  91.                 e.printStackTrace();
  92.             }
  93.         });
  94.         //上面所有任务处理完毕完毕之后，程序才能继续
  95.         System.out.println(list + "中的任务都处理完毕!");
  96.     }
  97. }

运行代码输出：

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

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

来源：http://itsoku.com/course/1/16

标签：JUC,Java,16,countDownLatch,System,耗时,线程,CountDownLatch,public
来源： https://www.cnblogs.com/konglxblog/p/16222187.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有