java并发学习-CountDownLatch实现原理全面讲解

淡淡的烟草味﹌ 2022-10-30 07:24 143阅读 0赞

CountDownLatch在多线程并发编程中充当一个计时器的功能，并且维护一个count的变量，并且其操作都是原子操作。

如下图，内部有下static final的Sync类继承自AQS.

该类主要通过countDown()和await()两个方法实现功能的，首先通过建立CountDownLatch对象，并且传入参数即为count初始值。

如果一个线程调用了await()方法，那么这个线程便进入阻塞状态，并进入阻塞队列。

如果一个线程调用了countDown()方法，则会使count-1；当count的值为0时，这时候阻塞队列中调用await()方法的线程便会逐个被唤醒，从而进入后续的操作。

补充：Java并发包中CountDownLatch的工作原理、使用示例

1. CountDownLatch的介绍  
CountDownLatch是一个同步工具，它主要用线程执行之间的协作。CountDownLatch 的作用和 Thread.join() 方法类似，让一些线程阻塞直到另一些线程完成一系列操作后才被唤醒。在直接创建线程的年代（Java 5.0 之前），我们可以使用 Thread.join()。在线程池出现后，因为线程池中的线程不能直接被引用，所以就必须使用 CountDownLatch 了。

CountDownLatch主要有两个方法，当一个或多个线程调用await方法时，这些线程会阻塞。其它线程调用countDown方法会将计数器减1(调用countDown方法的线程不会阻塞)，当计数器的值变为0时，因await方法阻塞的线程会被唤醒，继续执行。

实现原理：计数器的值由构造函数传入，并用它初始化AQS的state值。当线程调用await方法时会检查state的值是否为0，如果是就直接返回（即不会阻塞）；如果不是，将表示该节点的线程入列，然后将自身阻塞。当其它线程调用countDown方法会将计数器减1，然后判断计数器的值是否为0，当它为0时，会唤醒队列中的第一个节点，由于CountDownLatch使用了AQS的共享模式，所以第一个节点被唤醒后又会唤醒第二个节点，以此类推，使得所有因await方法阻塞的线程都能被唤醒而继续执行。

从源代码和实现原理中可以看出一个CountDownLatch对象，只能使用一次，不能重复使用。

await方法源码  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
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)  
doAcquireSharedInterruptibly(arg);  
\}  
  
protected int tryAcquireShared(int acquires) \{  
return (getState() == 0) ? 1 : -1;  
\}  
doAcquireSharedInterruptibly 主要实现线程的入列与阻塞。

countDown方法  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
public void countDown() \{  
sync.releaseShared(1);  
\}  
  
public final boolean releaseShared(int arg) \{  
if (tryReleaseShared(arg)) \{  
doReleaseShared();  
return true;  
\}  
return false;  
\}  
  
protected boolean tryReleaseShared(int releases) \{  
// Decrement count; signal when transition to zero  
for (;;) \{  
int c = getState();  
if (c == 0)  
return false;  
int nextc = c-1;  
if (compareAndSetState(c, nextc))  
return nextc == 0;  
\}  
\}  
doReleaseShared主要实现唤醒第一个节点，第一个节点有会唤醒第二个节点，……。

2. 使用示例  
1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  
17  
18  
19  
20  
21  
22  
23  
24  
25  
26  
27  
28  
29  
30  
31  
32  
33  
34  
35  
36  
37  
38  
39  
40  
41  
42  
43  
44  
45  
46  
47  
48  
49  
50  
51  
52  
53  
54  
55  
56  
57  
58  
59  
60  
61  
62  
63  
package demo;  
import java.util.Random;  
import java.util.concurrent.CountDownLatch;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;  
public class CountDownLatchDemo \{  
private CountDownLatch cdl = new CountDownLatch(2);  
private Random rnd = new Random();  
  
class FirstTask implements Runnable\{  
private String id;  
  
public FirstTask(String id)\{  
this.id = id;  
\}  
  
@Override  
public void run()\{  
System.out.println("Thread "+ id + " is start");  
try \{  
Thread.sleep(rnd.nextInt(1000));  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
System.out.println("Thread "+ id + " is over");  
cdl.countDown();  
\}  
\}  
  
class SecondTask implements Runnable\{  
private String id;  
  
public SecondTask(String id)\{  
this.id = id;  
\}  
  
@Override  
public void run()\{  
try \{  
cdl.await();  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
System.out.println("----------Thread "+ id + " is start");  
try \{  
Thread.sleep(rnd.nextInt(1000));  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
System.out.println("----------Thread "+ id + " is over");  
\}  
\}  
  
public static void main(String\[\] args)\{  
ExecutorService es = Executors.newCachedThreadPool();  
CountDownLatchDemo cdld = new CountDownLatchDemo();  
es.submit(cdld.new SecondTask("c"));  
es.submit(cdld.new SecondTask("d"));  
es.submit(cdld.new FirstTask("a"));  
es.submit(cdld.new FirstTask("b"));  
es.shutdown();  
\}  
\}  
在这个示例中，我们创建了四个线程a、b、c、d，这四个线程几乎同时提交给了线程池。c线程和d线程会在a线程和b线程结束后开始执行。

运行结果  
1  
2  
3  
4  
5  
6  
7  
8  
Thread a is start  
Thread b is start  
Thread b is over  
Thread a is over  
\----------Thread c is start  
\----------Thread d is start  
\----------Thread d is over  
\----------Thread c is over  
以上为个人经验，希望能给大家一个参考