简约而不简单的CountDownLatch

发布时间:2022-03-01 11:06:27 作者:yexindonglai@163.com 阅读(433)

CountDownLatch是什么?

    CountDownLatch是JDK1.5之后提供的一个同步工具,在并发包下面,它可以让一个或多个线程等待,一直等到其他线程中执行完成一组操作。实现功能和java多线程中的join() 方法很像;想要详细了解join方法可以看我另一篇文章: java多线程join()方法的作用和实现原理

 CountDownLatch有哪些常用方法

CountDownLatch在调用构造方法初始化时,需要指定用给定一个整数作为计数器;这个计数器用来阻塞await方法;

  • countDown方法 : 计数器会被减1,调用此方法不会阻塞
  • await方法           : 如果计数器大于0时,线程会被阻塞,一直到计数器被countDown方法减到0时,线程才会继续执行。计数器是无法重置的,当计数器被减到0时,调用await方法都会直接返回。从而解除阻塞状态执行await方法后面的代码;

应用场景

    说明白了用法,那么这个同步工具能用在哪些场景呢?假设我们有以下的场景:

比如有三个人小红、小李、小王, 三个人相约一起去酒店吃饭,菜已经点好了, 三个人从不同的地方出发,只有三个人都到了酒店之后才会开始上菜;那么这三个人就分别代表三个线程,这三个线程执行完之后才会执行 “上菜” 的代码逻辑;这种场景下就可以使用CountDownLatch实现

上代码  ConutDownLatchDemo.java

  1. package com.Lock;
  2. import java.util.concurrent.CountDownLatch;
  3. /**
  4. * 使用CountDownLatch同步器实现多线程等待
  5. * 比如有三个人小红、小李、小王, 三个人相约一起去酒店吃饭,菜已经点好了, 三个人从不同的地方出发,只有三个人都到了酒店之后才会开始上菜;那么这三个人就分别代表三个线程,这三个线程执行完之后才会执行 “上菜” 的代码逻辑,
  6. *
  7. */
  8. public class ConutDownLatchDemo implements Runnable{
  9. CountDownLatch latch ;
  10. public ConutDownLatchDemo (CountDownLatch latch){
  11. this.latch = latch;
  12. }
  13. @Override
  14. public void run() {
  15. System.out.println(Thread.currentThread().getName()+"开始出发了");
  16. try {
  17. Thread.sleep(1000); // 到酒店的过程中用延时1秒来代替
  18. } catch (InterruptedException e) {
  19. e.printStackTrace();
  20. }
  21. System.out.println(Thread.currentThread().getName()+"到酒店了");
  22. latch.countDown();// 计数器 - 1
  23. }
  24. }
  25. class Hotel implements Runnable{
  26. // 同步工具
  27. CountDownLatch latch ;
  28. public Hotel (CountDownLatch latch){
  29. this.latch = latch;
  30. }
  31. @Override
  32. public void run() {
  33. System.out.println(Thread.currentThread().getName() +"正在等待大家的到来.....");
  34. try {
  35. // 如果计数器的值 > 0 ,则执行当前方法时进入阻塞,如果小于0,则放行
  36. latch.await();
  37. } catch (InterruptedException e) {
  38. e.printStackTrace();
  39. }
  40. System.out.println("人齐了,"+Thread.currentThread().getName() +"服务员开始上菜");
  41. }
  42. }
  43. class Main1{
  44. public static void main(String[] args) throws InterruptedException {
  45. CountDownLatch latch = new CountDownLatch(3);
  46. new Thread(new ConutDownLatchDemo(latch), "小红").start();
  47. new Thread(new ConutDownLatchDemo(latch), "小王").start();
  48. new Thread(new ConutDownLatchDemo(latch), "小李").start();
  49. // 酒店线程
  50. new Thread(new Hotel(latch), "酒店").start();
  51. }
  52. }

通过打印结果可以看到,服务员上菜之前,酒店的线程一直阻塞着,只有当三个人都到了酒店之后才会开始上菜;

其中一个线程阻塞了或者挂了怎么办?

三个人一起到饭店吃饭,如果有一个人在路上出了点事情,迟迟不能到饭店,或者说临时被老板叫去加班, 不来了,其他的人也不能一直等啊,这时候怎么办?

解决办法很简单,await() 有一个重载的方法 

await(long timeout, TimeUnit unit)

比如我们设置为5秒钟,

latch.await(5, TimeUnit.SECONDS);

那么超过5秒后,不管人来没来齐,服务员都会照样上菜;

CountDownLatch的实现原理

CountDownLatch有一个内部类叫做Sync,它继承了AbstractQueuedSynchronizer类,其中维护了一个整数state,并且保证了修改state的可见性和原子性。

创建CountDownLatch实例时,也会创建一个Sync的实例,同时把计数器的值传给Sync实例,具体是这样的:

  1. public CountDownLatch(int count) {
  2. if (count < 0) throw new IllegalArgumentException("count < 0");
  3. this.sync = new Sync(count);
  4. }

在 countDown方法中,只调用了Sync实例的releaseShared方法,具体是这样的:

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

其中的releaseShared方法,先对计数器进行减1操作,如果减1后的计数器为0,唤醒被await方法阻塞的所有线程,具体是这样的:

  1. public final boolean releaseShared(int arg) {
  2. if (tryReleaseShared(arg)) { //对计数器进行减一操作
  3. doReleaseShared();//如果计数器为0,唤醒被await方法阻塞的所有线程
  4. return true;
  5. }
  6. return false;
  7. }

其中的tryReleaseShared方法,先获取当前计数器的值,如果计数器为0时,就直接返回;如果不为0时,使用CAS方法对计数器进行减1操作,具体是这样的:

  1. protected boolean tryReleaseShared(int releases) {
  2. for (;;) {//死循环,如果CAS操作失败就会不断继续尝试。
  3. int c = getState();//获取当前计数器的值。
  4. if (c == 0)// 计数器为0时,就直接返回。
  5. return false;
  6. int nextc = c-1;
  7. if (compareAndSetState(c, nextc))// 使用CAS方法对计数器进行减1操作
  8. return nextc == 0;//如果操作成功,返回计数器是否为0
  9. }
  10. }

await方法中,只调用了Sync实例的acquireSharedInterruptibly方法,具体是这样的:

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

其中acquireSharedInterruptibly方法,判断计数器是否为0,如果不为0则阻塞当前线程,具体是这样的:

  1. public final void acquireSharedInterruptibly(int arg)
  2. throws InterruptedException {
  3. if (Thread.interrupted())
  4. throw new InterruptedException();
  5. if (tryAcquireShared(arg) < 0)//判断计数器是否为0
  6. doAcquireSharedInterruptibly(arg);//如果不为0则阻塞当前线程
  7. }

 其中tryAcquireShared方法,是AbstractQueuedSynchronizer中的一个模板方法,其具体实现在Sync类中,其主要是判断计数器是否为零,如果为零则返回1,如果不为零则返回-1,具体是这样的:

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

 

关键字Java