Java多线程编程中经常会碰到这样一种场景——某个线程需要等待一个或多个线程操作结束(或达到某种状态)才开始执行。比如开发一个关发测试工具时,主线程需要等到所有测试线程均执行完成再开始统计总共耗费的时间,此时可以通过CountDownLatch轻松实现。
package com.test.thread;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
int totalThread = 3;
long start = System.currentTimeMillis();
CountDownLatch countDown = new CountDownLatch(totalThread);
for(int i = 0; i < totalThread; i++) {
final String threadName = "Thread " + i;
new Thread(() -> {
System.out.println(String.format("%s/t%s %s", new Date(), threadName, "started"));
try {
Thread.sleep(1000);
} catch (Exception ex) {
ex.printStackTrace();
}
countDown.countDown();
System.out.println(String.format("%s/t%s %s", new Date(), threadName, "ended"));
}).start();;
}
countDown.await();
long stop = System.currentTimeMillis();
System.out.println(String.format("Total time : %sms", (stop - start)));
}
}
执行结果
Sun Jun 19 20:34:31 CST 2016 Thread 1 started
Sun Jun 19 20:34:31 CST 2016 Thread 0 started
Sun Jun 19 20:34:31 CST 2016 Thread 2 started
Sun Jun 19 20:34:32 CST 2016 Thread 2 ended
Sun Jun 19 20:34:32 CST 2016 Thread 1 ended
Sun Jun 19 20:34:32 CST 2016 Thread 0 ended
Total time : 1072ms
可以看到,主线程等待所有3个线程都执行结束后才开始执行。
CountDownLatch工作原理相对简单,可以简单看成一个倒计时器,在构造方法中指定初始值,每次调用 countDown() 方法时讲计数器减1,而 await() 会等待计数器变为0。CountDownLatch关键接口如下
在《 当我们说线程安全时,到底在说什么 》一文中讲过内存屏障,它能保证屏障之前的代码一定在屏障之后的代码之前被执行。CyclicBarrier可以译为循环屏障,也有类似的功能。CyclicBarrier可以在构造时指定需要在屏障前执行await的个数,所有对await的调用都会等待,只到调用await的次数达到预定指,所有等待都会立即被唤醒。
从使用场景上来说,CyclicBarrier是让多个线程互相等待某一事件的发生,然后同时被唤醒。而上文讲的CountDownLatch是让某一线程等待多个线程的状态,然后该线程被唤醒。
package com.test.thread;
import java.util.Date;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
public static void main(String[] args) {
int totalThread = 5;
CyclicBarrier barrier = new CyclicBarrier(totalThread);
for(int i = 0; i < totalThread; i++) {
String threadName = "Thread " + i;
new Thread(() -> {
System.out.println(String.format("%s/t%s %s", new Date(), threadName, " is waiting"));
try {
barrier.await();
} catch (Exception ex) {
ex.printStackTrace();
}
System.out.println(String.format("%s/t%s %s", new Date(), threadName, "ended"));
}).start();
}
}
}
执行结果如下
Sun Jun 19 21:04:49 CST 2016 Thread 1 is waiting
Sun Jun 19 21:04:49 CST 2016 Thread 0 is waiting
Sun Jun 19 21:04:49 CST 2016 Thread 3 is waiting
Sun Jun 19 21:04:49 CST 2016 Thread 2 is waiting
Sun Jun 19 21:04:49 CST 2016 Thread 4 is waiting
Sun Jun 19 21:04:49 CST 2016 Thread 4 ended
Sun Jun 19 21:04:49 CST 2016 Thread 0 ended
Sun Jun 19 21:04:49 CST 2016 Thread 2 ended
Sun Jun 19 21:04:49 CST 2016 Thread 1 ended
Sun Jun 19 21:04:49 CST 2016 Thread 3 ended
从执行结果可以看到,每个线程都不会在其它所有线程执行 await() 方法前继续执行,而等所有线程都执行 await() 方法后所有线程的等待都被唤醒从而继续执行。
CyclicBarrier提供的关键方法如下
CountDownLatch和CyclicBarrier都是JDK 1.5引入的,而Phaser是JDK 1.7引入的。Phaser的功能与rCountDownLatch和CyclicBarrier有部分重叠,同时也提供了更丰富的语义和更灵活的用法。
Phaser顾名思义,与阶段相关。Phaser比较适合这样一种场景,一种任务可以分为多个阶段,现希望多个线程去处理该批任务,对于每个阶段,多个线程可以并发进行,但是希望保证只有前面一个阶段的任务完成之后才能开始后面的任务。这种场景可以使用多个CyclicBarrier来实现,每个CyclicBarrier负责等待一个阶段的任务全部完成。但是使用CyclicBarrier的缺点在于,需要明确知道总共有多少个阶段,同时并行的任务数需要提前预定义好,且无法动态修改。而Phaser可同时解决这两个问题。
public class PhaserDemo {
public static void main(String[] args) throws IOException {
int parties = 3;
int phases = 4;
final Phaser phaser = new Phaser(parties) {
@Override
protected boolean onAdvance(int phase, int registeredParties) {
System.out.println("====== Phase : " + phase + " ======");
return registeredParties == 0;
}
};
for(int i = 0; i < parties; i++) {
int threadId = i;
Thread thread = new Thread(() -> {
for(int phase = 0; phase < phases; phase++) {
System.out.println(String.format("Thread %s, phase %s", threadId, phase));
phaser.arriveAndAwaitAdvance();
}
});
thread.start();
}
}
}
执行结果如下
Thread 0, phase 0
Thread 1, phase 0
Thread 2, phase 0
====== Phase : 0 ======
Thread 2, phase 1
Thread 0, phase 1
Thread 1, phase 1
====== Phase : 1 ======
Thread 1, phase 2
Thread 2, phase 2
Thread 0, phase 2
====== Phase : 2 ======
Thread 0, phase 3
Thread 1, phase 3
Thread 2, phase 3
====== Phase : 3 ======
从上面的结果可以看到,多个线程必须等到其它线程的同一阶段的任务全部完成才能进行到下一个阶段,并且每当完成某一阶段任务时,Phaser都会执行其 onAdvance 方法。
Phaser主要接口如下