图解 Java并发包之CountDownLatch计数器 原理

Java (42) 2024-02-09 17:12

Hi，大家好，我是编程小6，很荣幸遇见你，我把这些年在开发过程中遇到的问题或想法写出来，今天说一说图解 Java并发包之CountDownLatch计数器 原理,希望能够帮助你!!!。

分享自己在Java方面的所思所想，希望你看完之后能有更多更深入的了解

本人微信公众号（jwfy的学习分享），欢迎关注～

关键字：AQS，CountDownLatch，countDown，await，线程无序

CountDownLatch 是基于AQS共享模式特定场景开发的一种同步器，也是在JUC并发包里面的内容，通过设置一个计数器来实现的，计时器的初始化值为线程个数，当线程完成自身的任务后，计数器会进行减一操作，直到计数器个数为0，意味着当前的所有任务已经完成，可以进行后续的任务操作。先来了解下CountDownLatch的使用demo，再深入学习一下其代码实现原理。

目录

Java 并发之CountDownLatch 计数器 操作图解细节

1、CountDownLatch Demo

1.1、主线程等待计数器完成任务

1.2、任务线程等待计数器完成任务

2、源码解析

3、问题

4、总结

1、CountDownLatch Demo

1.1、主线程等待计数器完成任务

CountDownLatch countDownLatch = new CountDownLatch(3);
Runnable r1 = () -> {
 try {
 Thread.sleep(1000 * 2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 countDownLatch.countDown();
 // 运行完成之后，计数器进行减一操作，以下同理
 System.out.println("R1 run end");
};
Runnable r2 = () -> {
 try {
 Thread.sleep(2000 * 2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 countDownLatch.countDown();
 System.out.println("R2 run end");
};
Runnable r3 = () -> {
 try {
 Thread.sleep(3000 * 2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 countDownLatch.countDown();
 System.out.println("R3 run end");
};
new Thread(r1).start();
new Thread(r2).start();
new Thread(r3).start();
System.out.println("start await");
try {
 countDownLatch.await();
 // 主线程会阻塞在这里等待计数器变成0之后，方可继续运行
} catch (InterruptedException e) {
 e.printStackTrace();
}
System.out.println("run end");

1.2、任务线程等待计数器完成任务

CountDownLatch mainLatch = new CountDownLatch(1);
CountDownLatch workLatch = new CountDownLatch(3);
Runnable r1 = () -> {
 try {
 Thread.sleep(2000 * 2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 workLatch.countDown();
 System.out.println("R1 run end");
};
Runnable r2 = () -> {
 try {
 Thread.sleep(3000 * 2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 workLatch.countDown();
 System.out.println("R2 run end");
};
Runnable r3 = () -> {
 try {
 Thread.sleep(4000 * 2);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 workLatch.countDown();
 System.out.println("R3 run end");
};
Runnable m = () -> {
 try {
 workLatch.await();
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("work run end");
 mainLatch.countDown();
};
new Thread(r1).start();
new Thread(r2).start();
new Thread(r3).start();
new Thread(m).start();
System.out.println("main run");

其实这种方法是把某一个countdownlatch运行完成之后消息通知给非主线程，主线程不必阻塞在这里，在例子中的main run 会优先被输出

需要注意上面这只是两种比较简单的使用方法，在线程执行await完成之后，相当于所有的线程都已经被唤醒了，可以操作，类似于栅栏，实际情况中存在着n个线程在countdown操作，而m个线程在await操作，为了确保线程安全，所有各种地方都存在CAS操作

2、源码解析

CountDownLatch中最主要的两个函数是countDown()和await()，现在就假设存在A、B、C、D 四个线程是进行countdown操作（也就是执行任务），线程E、F 是await操作（等待ABCD四个任务完成后方可通过栅栏）

在使用过程中，计数器格式是线程执行任务的个数，本文中也就是4，在起初设置的时候应该有 new CountDownLatch(4)

先来看看await方法，现在假设E、F线程是await操作的

如下代码不是在一个文件内的，这样拼接是为了便于分析源码

public void await() throws InterruptedException {
 // 支持中断的共享模式锁
 sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
 throws InterruptedException {
 if (Thread.interrupted())
 // 当前的线程如果出现中断了，则抛出一个新的中断异常事件
 throw new InterruptedException();
 
 if (tryAcquireShared(arg) < 0)
 // 返回true，也就表示还包含了没有完成任务的线程，当前线程需要尝试获取锁
 doAcquireSharedInterruptibly(arg);
}
protected int tryAcquireShared(int acquires) {
 return (getState() == 0) ? 1 : -1;
 // 尝试获取当前状态是否是0
 // 如果当前计时器数据为0，则认为可以获取了，返回1，否则返回-1
}

到现在EF两个线程在执行await时，肯定都会进入到doAcquireSharedInterruptibly中

private void doAcquireSharedInterruptibly(int arg)
 throws InterruptedException {
 final Node node = addWaiter(Node.SHARED);
 // 情况1 创建一个共享模式的节点插入到CHL队列中，会确保线程安全的
 boolean failed = true;
 try {
 for (;;) {
 final Node p = node.predecessor();
 if (p == head) {
 // 如果当前节点的前节点是head，头节点
 int r = tryAcquireShared(arg);
 if (r >= 0) {
 // 大于0 则意味着任务线程计数器减小到0了，可以唤醒下一个节点操作了
 setHeadAndPropagate(node, r);
 p.next = null; // help GC
 failed = false;
 return;
 }
 }
 if (shouldParkAfterFailedAcquire(p, node) &&
 parkAndCheckInterrupt())
 // 把前节点状态设置为-1，然后返回false
 // 然后再次运行到这里时，进入到park中，线程挂起
 // 注意！被唤醒的线程也是在这里继续执行的，等下还会回过头来到这里的
 throw new InterruptedException();
 }
 } finally {
 if (failed)
 cancelAcquire(node);
 }
}

在线程E通过addWaiter方法操作后，CHL队列变成

图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第1张

  • 在for循环的第一遍把头结点的waitStatus值设置为-1，返回false
  • 在for循环的第二部发现头结点状态位-1，返回true，进入到park中线程挂起
图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第2张

然后在线程F进来的时候，同理也会进入到该CHL队列中，最后的情况如下图

图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第3张

现在EF就是执行了await函数后进入到CHL队列中被挂起了，一直在等待直到ABCD四个线程完成了各自的任务使得status=0

现在再看看countDown函数的操作细节

public void countDown() {
 sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
 if (tryReleaseShared(arg)) {
 // 只有最后一个通过CAS操作后把status变成0的线程会返回true
 // 否则都是返回false
 doReleaseShared();
 return true;
 }
 return false;
}

ABCD四个线程调用countDown 方法时，基本上只会进行CAS减一操作，而只有最后一个完成任务的线程会返回true，然后调用doReleaseShared方法，也就是说只有一个线程会去开启唤醒CHL队列的线程

现在假设D线程执行了最后一个CAS减一操作，然后触发了唤醒操作

private void doReleaseShared() {
 for (;;) {
 Node h = head;
 if (h != null && h != tail) {
 // 如果队列存在有效节点数据
 int ws = h.waitStatus;
 if (ws == Node.SIGNAL) {
 // 头结点的状态位-1，可以唤醒头部后置节点
 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
 continue; 
 // 此时头部节点状态位0，修改头部节点而且进行唤醒操作 
 // 这个地方就是开始唤醒头结点的下个节点的操作，也就是样例中的E节点
 unparkSuccessor(h);
 }
 else if (ws == 0 &&
 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
 continue; // loop on failed CAS
 }
 if (h == head) // loop if head changed
 break;
 }
}

现在由当前执行中的任务线程D通过unpark操作，唤醒了CHL队列的E节点所在线程，再次来到了doAcquireSharedInterruptibly方法中，E线程继续运行

final Node p = node.predecessor();
if (p == head) {
 // E节点的前节点就是head，node节点就是E本身
 int r = tryAcquireShared(arg);
 if (r >= 0) {
 // 现在所有的任务现场都进行了减一操作，r一定大于0的
 setHeadAndPropagate(node, r);
 // node节点设置为head节点
 p.next = null; // help GC
 failed = false;
 return;
 }
}
private void setHead(Node node) {
 head = node;
 node.thread = null;
 node.prev = null;
}
private void setHeadAndPropagate(Node node, int propagate) {
 Node h = head; // Record old head for check below
 setHead(node);
 // 把node节点设置为head节点
 
 if (propagate > 0 || h == null || h.waitStatus < 0 ||
 (h = head) == null || h.waitStatus < 0) {
 Node s = node.next;
 if (s == null || s.isShared())
 // 然后又开始了唤醒CHL队列的其他节点线程（此处是F线程节点）
 doReleaseShared();
 }
}

E线程经过了setHeadAndPropagate方法之后，CHL队列变成了如下图所示，图中并未画出线程E节点，是因为现在E节点正处于运行中，不在CHL队列中

图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第4张

doAcquireSharedInterruptibly 函数中每个步骤看上去并不具备线程安全的特性，那么其是如何保证线程安全的呢？因为每次添加节点都是在tail处添加节点，而唤醒则是都是从head节点开始进行唤醒操作的，能够很好的从业务上避免线程安全的问题

后续又经过了把原头结点进行GC回收操作，其await操作完成，也进行了现在F的唤醒操作

现在CHL队列如下图所示

图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第5张

现在进行唤醒F线程的操作，又来到了doReleaseShared方法，注意是E线程调用的此方法

private void doReleaseShared() {
 for (;;) {
 Node h = head;
 if (h != null && h != tail) {
 // 如果队列存在有效节点数据
 int ws = h.waitStatus;
 if (ws == Node.SIGNAL) {
 // 头结点的状态位-1，可以唤醒头部后置节点
 // 现在的头结点状态位确实为-1，可以唤醒
 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
 // CAS设置头结点失败，继续轮询操作
 // 此处为问题1，后面会重点解释为什么会出现该类情况
 continue; 
 // 此时头部节点状态位0，修改头部节点而且进行唤醒操作 
 // 这里就是唤醒F节点的操作入口
 unparkSuccessor(h);
 }
 else if (ws == 0 &&
 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
 // 头结点为0，然后进行CAS设置为-3失败了，继续轮询
 // 问题2，为什么会出现CAS失败的情况
 continue; // loop on failed CAS
 }
 if (h == head) // loop if head changed
 // 经过上面的操作，E线程唤醒任务完成，本身退出该for循环
 break;
 }
}

图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第6张

然后又进入到unpark中唤醒F线程的操作，而E线程本身进入到break中，完成任务。

E线程经历了加入到CHL队列中、主动挂起、被前置节点（头节点）唤醒、CAS操作设置自身状态为0、唤醒下一个节点等五个主要步骤后完成await操作，进入到下一步。

3、问题

那么在doReleaseShared中有提出两个问题，现在就来分析下其具体原因，在分析前我们需要清晰的认识到doReleaseShared方法调用方来源自两个地方，一个是最后一个任务线程的首次调用，另一个是CHL上一个节点的唤醒操作，而且一旦首次唤醒之后，后面执行的顺序就可能因为各种问题导致各自执行的顺序不完全一致

问题1：compareAndSetWaitStatus(h, Node.SIGNAL, 0) 操作失败原因

进入到这一步之前已经通过了ws == Node.SIGNAL校验，但是CAS失败，那只能说明h结点和head节点不是同一个，查看代码执行的整个链路，setHeadAndPropagate操作可能会进行重置head节点操作

上面已经强调了CHL队列中的节点是由其他线程节点唤醒的，那么就存在这样的一种场景

  • 线程E通过unparkSuccessor(h) 操作唤醒线程F节点（此时线程F是CHL队列的第一个节点，也是头部的下一个节点）
  • 线程F节点被唤醒后，通过setHeadAndPropagate方法重置head
  • 线程E判断h == head 出现问题，因为head已经被线程F重置了，进入到第二个循环
  • 线程F继续通过setHeadAndPropagate方法也进入到了doReleaseShared方法中尝试唤醒下一个节点，那么就导致了存在线程E和F都进入到这个方法的场景，此操作极大导致CAS失败
  • 线程E因为h!=head 进入第二个循环，然后经过了wa==Node.SIGNAL 操作后，CPU时间片被线程F获取了，线程F也进入到该if判断，优先通过了CAS操作，设置了当前的新节点状态位0
  • 线程F后续继续唤醒下一个节点操作，而此时线程E重新获取CPU时间片，进行CAS操作，肯定会失败，因为已经被线程F修改为了0

总结：正是因为线程之间的无序性，导致了每一个代码块都可能出现线程不安全的问题，所以加上CAS和for严格确保线程的安全特性

问题2：compareAndSetWaitStatus(h, 0, Node.PROPAGATE) 操作失败原因

先仔细想想为什么ws==0会成立，也只有一种可能，头结点在问题1中的CAS操作成功之后被重置为0，如下图这种场景

图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第7张

  • 线程E设置头节点状态为0后，使用unparkSuccessor唤醒线程F操作
  • 线程F在setHeadAndPropagate操作中重置head为自身
  • 此时新插入一个节点线程G到CHL队列中（不是很理解这种新增节点的场景），只进行了addWaiter方法调用，如下图为最新的CHL队列
图解 Java并发包之CountDownLatch计数器 原理_https://bianchenghao6.com/blog_Java_第8张

  • 线程E因为h!=head，进入到第二个for循环中，而且ws=0
  • 线程G获取CPU时间片，通过shouldParkAfterFailedAcquire 方法把head节点的状态更改为-1
  • 线程E在第二个for循环中的CAS必然失败，因为此时head线程的状态位-1

4、总结

相比AQS而言，CountDownLatch还是相对简单的，使用了AQS的共享模式，当任务完成后，由最后一个线程开启唤醒的操作，然后由头结点开始一个接着一个进行唤醒，因为线程的无序性，加上CAS乐观的保证线程安全，如需好好理解还是要细细琢磨其中的细节

原创推荐

「面试」new String("abc")和"abc"有什么区别？反编译看看原理吧

面试：你是否了解Spring BeanPostProcessor的原理和具体使用场景

「Spring源码」XML解析BeanDefinition原理学习

从Tomcat入口了解Spring MVC的请求处理流程

「Spring源码」深入学习事件通知机制

Spring原理：10个Spring&SpringBoot高阶用法，你是否清楚？

今天的分享到此就结束了，感谢您的阅读，如果确实帮到您，您可以动动手指转发给其他人。

上一篇

java爬虫 网页请求HttpClient

下一篇

Java工作五年还在写业务代码做着CRUD的工作？该好好反思自己了

发表回复

相关推荐

Java-System.currentTimeMillis() 存在性能问题

（附源码课件）10款Java小游戏满足你各种需求

Javaweb学生信息管理系统(源码)

Java之路：this关键字的用法

最新文章

更多>

Java-System.currentTimeMillis() 存在性能问题

8小时前 3

（附源码课件）10款Java小游戏满足你各种需求

1天前 3

Javaweb学生信息管理系统(源码)

2天前 3

Java之路：this关键字的用法

2天前 2

JAVA 新手入门

2天前 2

java实现5毛钱特效

3天前 2

java生成二维码图片

3天前 2

快速搭建一个自己的服务器详解（java环境）

3天前 2

Java案例：实现九九乘法表

3天前 3

java冒泡排序详解

3天前 3