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

时间:2021-06-01 17:32:33

1. CountDownLatch的介绍

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

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

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

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

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)
doAcquireSharedInterruptibly(arg);
}

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

doAcquireSharedInterruptibly 主要实现线程的入列与阻塞。

countDown方法

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. 使用示例

package javalearning;

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线程结束后开始执行。

运行结果

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

3. 参考内容

[1] http://developer.51cto.com/art/201403/432095.htm