编程语言
首页 > 编程语言> > Java高并发专题之16、JUC中的CountDownLatch

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

作者:互联网

本篇内容

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

思考一个需求

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

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

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

输出:

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

输出:

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

输出:

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

输出:

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

运行代码输出:

  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