二十七、并发编程之并发工具类CountDownLatch详解

在JDK的并发包里(java.util.concurrent)提供了这样几个非常有用的并发工具类来解决并发编程的流程控制。分别是CountDownLatch、CyclicBarrier和Semaphore。

1、CountDownLatch是什么?

CountDownLatch类位于java.util.concurrent包下,利用它可以实现类似计数器的功能。CountDownLatch大多是被用在等待多线程完成,具体来说就是允许一个或多个线程等待其他线程完成操作。比如有一个任务A,它要等待其他4个任务执行完毕之后才能执行,此时就可以利用CountDownLatch来实现这种功能了。

2、CountDownLatch原理

CountDownLatch类只提供了一个构造器:

1public void CountDownLatch(int count) {...} 2 3

在 CountDownLunch启动的时候。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。这样主线程的操作就会在这个方法上阻塞,直到其他线程完成各自的任务。在每次任务执行完直接调用,计数器就会减一操作。

1//调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行 2public void await() throws InterruptedException { }; 3//和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行 4public boolean await(long timeout, TimeUnit unit) throws InterruptedException { }; 5//将count值减1 6public void countDown() { }; 7 8

await()这个方法就是用来堵塞主线程的,前者是有等待时间的,可以自定义,后者是无限等待,直到其他count 计数器为0为止。

在这里插入图片描述

3、使用场景

  • 超时机制

主线程里面设置好等待时间,如果发现在规定时间内还是没有返回结果,那就唤醒主线程,抛弃。

  • 开始执行前等待n个线程完成各自任务

例如应用程序启动类要确保在处理用户请求前,所有N个外部系统已经启动和运行了。

  • 死锁检测

一个非常方便的使用场景是,你可以使用n个线程访问共享资源,在每次测试阶段的线程数目是不同的,并尝试产生死锁。

4、深入源码

CountDownLunch 源码底层是由AbstractQueuedSynchronizer提供支持(后面就简称 AQS),所以其数据结构就是AQS的数据结构,而AQS的核心就是两个虚拟队列:同步队列syncQueue 和条件队列conditionQueue(前者数据结构是双向链表,后者是单向链表)不同的条件会有不同的条件队列。

  • Sync源码

1 private static final class Sync extends AbstractQueuedSynchronizer { 2 // 版本号 3 private static final long serialVersionUID = 4982264981922014374L; 4 5 // 构造器 6 Sync(int count) { 7 setState(count); 8 } 9 10 // 返回当前计数 11 int getCount() { 12 return getState(); 13 } 14 15 // 试图在共享模式下获取对象状态 16 protected int tryAcquireShared(int acquires) { 17 return (getState() == 0) ? 1 : -1; 18 } 19 20 // 试图设置状态来反映共享模式下的一个释放 21 protected boolean tryReleaseShared(int releases) { 22 // Decrement count; signal when transition to zero 23 // 无限循环 24 for (;;) { 25 // 获取状态 26 int c = getState(); 27 if (c == 0) // 没有被线程占有 28 return false; 29 // 下一个状态 30 int nextc = c-1; 31 if (compareAndSetState(c, nextc)) // 比较并且设置成功 32 return nextc == 0; 33 } 34 } 35 } 36 37
  • await函数

此函数将会使当前线程在锁存器倒计数至零之前一直等待,除非线程被中断。其源码如下:

1public void await() throws InterruptedException{ 2 // 转发到sync对象上 3 sync.acquireSharedInterruptibly(1); 4} 5 6

源码可知,对CountDownLatch对象的await的调用会转发为对Sync的acquireSharedInterruptibly(从AQS继承的方法)方法的调用。

1 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { 2 if (Thread.interrupted()) 3 throw new InterruptedException(); 4 if (tryAcquireShared(arg) < 0) 5 doAcquireSharedInterruptibly(arg); 6 } 7 8

这里先检测了线程中断状态,中断了则抛出异常,接下来调用tryAcquireShared,tryAcquireShared是Syn的实现的。

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

其实就是简单的获取了同步器的state,判断是否为0。
接下来是

1 private void doAcquireSharedInterruptibly(int arg)throws InterruptedException { 2 final Node node = addWaiter(Node.SHARED); 3 boolean failed = true; 4 try { 5 for (;;) { 6 final Node p = node.predecessor(); 7 if (p == head) { 8 int r = tryAcquireShared(arg); 9 if (r >= 0) { 10 setHeadAndPropagate(node, r); 11 p.next = null; // help GC 12 failed = false; 13 return; 14 } 15 } 16 if (shouldParkAfterFailedAcquire(p, node) && 17 parkAndCheckInterrupt()) 18 throw new InterruptedException(); 19 } 20 } finally { 21 if (failed) 22 cancelAcquire(node); 23 } 24 } 25 26

关键点是parkAndCheckInterrupt()

1private final boolean parkAndCheckInterrupt() { 2 LockSupport.park(this); 3 return Thread.interrupted(); 4 } 5 6

执行到此处时,线程会阻塞,知道有其他线程唤醒此线程,执行await之后,上文中的主线程阻塞在这。
在这里插入图片描述

  • countDown函数

此函数将递减锁存器的计数,如果计数到达零,则释放所有等待的线程

1void countDown() { 2 sync.releaseShared(1); 3} 4 5

可以看出 对countDown的调用转换为对Sync对象的releaseShared(从AQS继承而来)方法的调用。

5、使用

1/** 2 * 对一个文本中的所有数字先行求和,再把所有行汇总 3 * */ 4public class Demo2 { 5 private int[] nums; //存储每一行求和数 6 7 public Demo2(int line) { 8 nums = new int[line]; 9 } 10 11 //行求和 12 public void calc(String line, int index,CountDownLatch latch) { 13 String[] nus = line.split(","); // 切分出每个值 14 int total = 0; 15 for (String num : nus) { 16 total += Integer.parseInt(num); 17 } 18 nums[index] = total; // 把计算的结果放到数组中指定的位置 19 System.out.println(Thread.currentThread().getName() + " 执行计算任务... " + line + " 结果为:" + total); 20 latch.countDown();//CountDownLatch计数器减1 21 } 22 23 //行汇总 24 public void sum() { 25 System.out.println(Thread.currentThread().getName()+"汇总线程开始执行... "); 26 int total = 0; 27 for (int i = 0; i < nums.length; i++) { 28 total += nums[i]; 29 } 30 System.out.println("最终的结果为:" + total); 31 } 32 //读取文本数据 33 private static List<String> readFile() { 34 List<String> contents = new ArrayList<>(); 35 String line = null; 36 BufferedReader br = null; 37 try { 38 br = new BufferedReader(new FileReader("/Users/Amy/Downloads/nums.txt")); 39 while ((line = br.readLine()) != null) { 40 contents.add(line); 41 } 42 } catch (Exception e) { 43 e.printStackTrace(); 44 } finally { 45 if (br != null) { 46 try { 47 br.close(); 48 } catch (IOException e) { 49 e.printStackTrace(); 50 } 51 } 52 } 53 return contents; 54 } 55 public static void main(String[] args) { 56 List<String> contents = readFile();//读取文本数据 57 int lineCount = contents.size();//行数 58 CountDownLatch latch = new CountDownLatch(lineCount); 59 Demo2 d = new Demo2(lineCount); 60 for (int i = 0; i < lineCount; i++) { 61 final int j = i; 62 new Thread(new Runnable() { 63 @Override 64 public void run() { 65 d.calc(contents.get(j), j,latch);//行求和 66 } 67 }).start(); 68 } 69 try { 70 latch.await();//等待所有线程将行求和全部进行完毕 71 } catch (InterruptedException e) { 72 e.printStackTrace(); 73 } 74 d.sum();//汇总求和 75 } 76} 77 78

代码交流 2021