1.CountDownLatch
1.适用场景
- 协调子线程结束动作：等待所有子线程运行结束
主线程创建了5个子线程，各子任务执行确认动作，期间主线程进入等待状态，直到各子线程的任务均已经完成，主线程恢复继续执行。
- 协调子线程开始动作：统一各线程动作开始的时机
从多线程的角度看，这恰似你创建了一些多线程，但是你需要统一管理它们的任务开始时间。
2.设计思想
CountDownLatch基于一个同步器实现，并且只有CountDownLatch(int count)一个构造器，指定数量count不得在中途修改它。
核心函数
- await()：等待latch降为0；
- boolean await(long timeout, TimeUnit unit)：等待latch降为0，但是可以设置超时时间。
- countDown()：latch数量减1；
- getCount()：获取当前的latch数量。
3.场景实例
场景1. 对各子线程的等待
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(4);
Thread t1 = new Thread(countDownLatch::countDown);
Thread t2 = new Thread(countDownLatch::countDown);
Thread t3 = new Thread(countDownLatch::countDown);
Thread t4 = new Thread(() -> {
try {
// 稍等...
Thread.sleep(1500);
countDownLatch.countDown();
} catch (InterruptedException ignored) {}
});
t1.start();
t2.start();
t3.start();
t4.start();
//直到所有线程都对计数器进行减一后，这里才放行
countDownLatch.await();
System.out.println("所有子线程就位，可以继续执行其他任务");
}
场景2. 对多线程的统一管理
我们仍然用4个线程调用了start()，但是它们在运行时都在等待countDownLatch的信号，在信号未收到前，它们不会往下执行。
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(1);
Thread t1 = new Thread(() -> waitForCountDown(countDownLatch));
Thread t2 = new Thread(() -> waitForCountDown(countDownLatch));
Thread t3 = new Thread(() -> waitForCountDown(countDownLatch));
Thread t4 = new Thread(() -> waitForCountDown(countDownLatch));
t1.start();
t2.start();
t3.start();
t4.start();
Thread.sleep(1000);
countDownLatch.countDown();
System.out.println("所有线程准备完成");
}
private static void waitForCountDown(CountDownLatch countDownLatch) {
try {
// 等待信号
countDownLatch.await();
System.out.println("本线程等待完毕");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
输出：
所有线程准备完成
本线程等待完毕
本线程等待完毕
本线程等待完毕
本线程等待完毕
Process finished with exit code 0
场景3. SOFAJRaft的实践
// 定义一个CountDownLatch计数器
private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
public void start() {
switch (workerStateUpdater.get(this)) {
case WORKER_STATE_INIT:
if (workerStateUpdater.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
//此处调用工作线程执行CountDownLatch的countDown()方法
//即startTimeInitialized.countDown();
workerThread.start();
}
break;
case WORKER_STATE_STARTED:
break;
case WORKER_STATE_SHUTDOWN:
throw new IllegalStateException("cannot be started once stopped");
default:
throw new Error("Invalid WorkerState");
}
// 等待startTime被工作线程初始化完成
while (startTime == 0) {
try {
startTimeInitialized.await();
} catch (InterruptedException ignore) {
// Ignore - it will be ready very soon.
}
}
}
2.CyclicBarrier
1.适用场景
栅栏类似于闭锁，它能阻塞一组线程直到某个事件的发生。栅栏与闭锁的关键区别在于，所有的线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。
CyclicBarrier与CountDownLatch的区别
2.设计思想
1.构造器
// 指定参与方的数量；
public CyclicBarrier(int parties);
// 指定参与方的数量，并指定在本代次结束时运行的代码
public CyclicBarrier(int parties, Runnable barrierAction)：
2.核心方法
//如果当前线程不是第一个到达屏障的话，它将会进入等待，直到其他线程都到达
//除非发生被中断、屏障被拆除、屏障被重设等情况
public int await();
//和await()类似，但是加上了时间限制；
public int await(long timeout, TimeUnit unit);
//当前屏障是否被拆除；
public boolean isBroken();
//重设当前屏障。会先拆除屏障再设置新的屏障
public void reset();
//正在等待的线程数量
public int getNumberWaiting();
3.场景实例
下面以一个简单的日常对话来讲解CyclicBarrier的使用实例
private static String appointmentPlace = "书房";
public static void main(String[] args) {
CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> System.out.println("yuan所在的地点:" + appointmentPlace));
// 线程Akai
Thread Akai = newThread("Akai", () -> {
System.out.println("yuan,饭好了快来吃饭...");
try {
// 此时Akai在屏障前等待
cyclicBarrier.await();
System.out.println("yuan，你来了...");
// 开始吃饭...
Thread.sleep(2600);
System.out.println("好的，你去洗你的碗吧！");
// 第二次调用await
cyclicBarrier.await();
Thread.sleep(100);
System.out.println("好吧，你这个懒猪！");
} catch (Exception e) {
e.printStackTrace();
}
});
// 线程yuan
Thread yuan = newThread("yuan", () -> {
try {
// yuan在敲代码
Thread.sleep(500);
System.out.println("我在敲代码，我马上就来!");
// yuan到达饭桌前
cyclicBarrier.await();
Thread.sleep(500);
System.out.println("Akai，不好意思，刚刚沉迷于敲代码了！");
// 开始吃饭...
Thread.sleep(1500);
// yuan想先吃完赶快洗碗然后溜出去敲代码
System.out.println("我吃完了，我要去洗碗了");
// yuan把地点改成了厨房
appointmentPlace = "厨房";
// 洗碗中...
Thread.sleep(1500);
System.out.println("︎yuan终于洗完自己的碗了");
// 第二次调用await
cyclicBarrier.await();
System.out.println("Akai你吃完了，你的碗自己去洗吧，我已经在敲代码了");
} catch (Exception ignored) {}
});
Akai.start();
yuan.start();
}
输出结果：
yuan,饭好了快来吃饭...
我在敲代码，我马上就来!
yuan所在的地点:书房
yuan，你来了...
Akai，不好意思，刚刚沉迷于敲代码了！
我吃完了，我要去洗碗了
好的，你去洗你的碗吧！
yuan终于洗完自己的碗了
yuan所在的地点:厨房
Akai你吃完了，你的碗自己去洗吧，我已经在敲代码了
好吧，你这个懒猪！
3.CompletableFuture
1.设计思想
1.Future的局限性
- 并发执行多任务：Future只提供了get()方法来获取结果，并且是阻塞的。所以，除了等待你别无他法；
- 无法对多个任务进行链式调用：如果你希望在计算任务完成后执行特定动作，比如发邮件，但Future却没有提供这样的能力；
- 无法组合多个任务：如果你运行了10个任务，并期望在它们全部执行结束后执行特定动作，那么在Future中这是无能为力的；
- 没有异常处理：Future接口中没有关于异常处理的方法；
2.Completable有哪些优势
CompletableFuture是Future接口的扩展和增强。
CompletableFuture完整地继承了Future接口，并在此基础上进行了丰富地扩展，完美地弥补了Future上述的种种问题。更为重要的是，CompletableFuture实现了对任务的编排能力。借助这项能力，我们可以轻松地组织不同任务的运行顺序、规则以及方式。
从某种程度上说，这项能力是它的核心能力。而在以往，虽然通过CountDownLatch等工具类也可以实现任务的编排，但需要复杂的逻辑处理，不仅耗费精力且难以维护。
2.核心设计
我们首先来讨论CompletableFuture的核心：CompletionStage
顾名思义，根据CompletionStage名字中的"Stage"，你可以把它理解为任务编排中的步骤。步骤，即任务编排的基本单元，它可以是一次纯粹的计算或者是一个特定的动作。在一次编排中，会包含多个步骤，这些步骤之间会存在依赖、链式和组合等不同的关系，也存在并行和串行的关系。这种关系，类似于Pipeline或者流式计算。
既然是编排，就需要维护任务的创建、建立计算关系。为此，CompletableFuture提供了多达50多个方法，但没有必要全部完全理解，但我们可以通过分类的方式简化对方法的理解，理解了类型和变种，基本上我们也就掌握了CompletableFuture的核心能力。
这些方法可以总结为以下四类，其他大部分方法都是基于这四种类型的变种:
3.核心用法
1.runAsync
- runAsync()是CompletableFuture最常用的方法之一，它可以接收一个待运行的任务并返回一个CompletableFuture
- 当我们想异步运行某个任务时，在以往需要手动实现Thread或者借助Executor实现。而通过runAsync()`就简单多了。比如，我们可以直接传入Runnable类型的任务：
CompletableFuture.runAsync(new Runnable() {
@Override
public void run() {
System.out.println("something");
}
});
2.supply与supplyAsync
- 所谓supply表示提供结果，换句话说当我们使用supply()时，就表明我们会返回一个结果，并且这个结果可以被后续的任务所使用。
// 创建nameFuture，返回姓名
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
return "Akai-yuan";
});
// 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
return "love you，" + name;
});
//阻塞获得表白的结果
System.out.println(sayLoveFuture.get()); // love you，Akai-yuan
一旦理解了supply()的含义，它也就如此简单。如果你希望用新的线程运行任务，可以使用supplyAsync().
3.thenApply与thenApplyAsync
- 我们已经知道supply()是用于提供结果的，并且顺带提了thenApply()。很明显，thenApply()是supply()的搭档，用于接收supply()的执行结果，并执行特定的代码逻辑，最后返回CompletableFuture结果。
// 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
return "爱你，" + name;
});
public <U> CompletableFuture <U> thenApplyAsync(
Function <? super T, ? extends U> fn) {
return uniApplyStage(null, fn);
}
4.thenAccept与thenAcceptAsync
作为supply()的档案，thenApply()并不是唯一的存在，thenAccept()也是。但与thenApply()不同，thenAccept()只接收数据，但不会返回，它的返回类型是Void.
CompletableFuture<Void> sayLoveFuture = nameFuture.thenAccept(name -> {
System.out.println("爱你，" + name);
});
public CompletableFuture < Void > thenAccept(Consumer < ? super T > action) {
return uniAcceptStage(null, action);
}
5.thenRun
thenRun()就比较简单了，不接收任务的结果，只运行特定的任务，并且也不返回结果。
public CompletableFuture < Void > thenRun(Runnable action) {
return uniRunStage(null, action);
}
所以，如果你在回调中不想返回任何的结果，只运行特定的逻辑，那么你可以考虑使用thenAccept和thenRun一般来说，这两个方法会在调用链的最后面使用。
6.thenCompose与 thenCombine
以上几种方法都是各玩各的，但thenCompose()与thenCombine()就不同了，它们可以实现对依赖和非依赖两种类型的任务的编排。
编排两个存在依赖关系的任务
在前面的例子中，在接收前面任务的结果时，我们使用的是thenApply(). 也就是说，sayLoveFuture在执行时必须依赖nameFuture的完成，否则执行个锤子。
// 创建Future
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
return "Akai-yuan";
});
// 使用thenApply()接收nameFuture的结果，并执行回调动作
CompletableFuture <String> sayLoveFuture = nameFuture.thenApply(name -> {
return "爱你，" + name;
});
但其实，除了thenApply()之外，我们还可以使用thenCompose()来编排两个存在依赖关系的任务。比如，上面的示例代码可以写成：
// 创建Future
CompletableFuture <String> nameFuture = CompletableFuture.supplyAsync(() -> {
return "Akai-yuan";
});
CompletableFuture<String> sayLoveFuture2 = nameFuture.thenCompose(name -> {
return CompletableFuture.supplyAsync(() -> "爱你，" + name);
});
可以看到，thenCompose()和thenApply()的核心不同之处在于它们的返回值类型：
- thenApply()：返回计算结果的原始类型，比如返回String;
- thenCompose()：返回CompletableFuture类型，比如返回CompletableFuture.
组合两个相互独立的任务
考虑一个场景，当我们在执行某个任务时，需要其他任务就绪才可以，应该怎么做？这样的场景并不少见，我们可以使用前面学过的并发工具类实现，也可以使用thenCombine()实现。
举个例子，当我们计算某个胜率时，我们需要获取她参与的总场次（rounds），以及获胜的场次（winRounds），然后再通过winRounds / rounds来计算。对于这个计算，我们可以这么做：
CompletableFuture < Integer > roundsFuture = CompletableFuture.supplyAsync(() -> 500);
CompletableFuture < Integer > winRoundsFuture = CompletableFuture.supplyAsync(() -> 365);
CompletableFuture < Object > winRateFuture = roundsFuture
.thenCombine(winRoundsFuture, (rounds, winRounds) -> {
if (rounds == 0) {
return 0.0;
}
DecimalFormat df = new DecimalFormat("0.00");
return df.format((float) winRounds / rounds);
});
System.out.println(winRateFuture.get());
thenCombine()将另外两个任务的结果同时作为参数，参与到自己的计算逻辑中。在另外两个参数未就绪时，它将会处于等待状态。
7.allOf与anyOf
allOf()与anyOf()也是一对孪生兄弟，当我们需要对多个Future的运行进行组织时，就可以考虑使用它们：
- allOf()：给定一组任务，等待所有任务执行结束；
- anyOf()：给定一组任务，等待其中任一任务执行结束。
allOf()与anyOf()的方法签名如下：
static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)
需要注意的是，anyOf()将返回完任务的执行结果，但是allOf()不会返回任何结果，它的返回值是Void.
allOf()与anyOf()的示例代码如下所示。我们创建了roundsFuture和winRoundsFuture，并通过sleep模拟它们的执行时间。在执行时，winRoundsFuture将会先返回结果，所以当我们调用 CompletableFuture.anyOf时也会发现输出的是365.
CompletableFuture < Integer > roundsFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(200);
return 500;
} catch (InterruptedException e) {
return null;
}
});
CompletableFuture < Integer > winRoundsFuture = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(100);
return 365;
} catch (InterruptedException e) {
return null;
}
});
CompletableFuture < Object > completedFuture = CompletableFuture.anyOf(winRoundsFuture, roundsFuture);
System.out.println(completedFuture.get()); // 返回365
CompletableFuture < Void > completedFutures = CompletableFuture.allOf(winRoundsFuture, roundsFuture);
在CompletableFuture之前，如果要实现所有任务结束后执行特定的动作，我们可以考虑CountDownLatch等工具类。现在，则多了一选项，我们也可以考虑使用CompletableFuture.allOf.
8.异常处理
在CompletableFuture链式调用中，如果某个任务发生了异常，那么后续的任务将都不会再执行。对于异常，我们有两种处理方式：exceptionally()和handle().
1.使用exceptionally()回调处理异常
在链式调用的尾部使用exceptionally()，捕获异常并返回错误情况下的默认值。需要注意的是，exceptionally()仅在发生异常时才会调用。
CompletableFuture < ? extends Serializable > winRateFuture = roundsFuture
.thenCombine(winRoundsFuture, (rounds, winRounds) -> {
if (rounds == 0) {
throw new RuntimeException("总场次错误");
}
DecimalFormat df = new DecimalFormat("0.00");
return df.format((float) winRounds / rounds);
}).exceptionally(ex -> {
System.out.println("出错：" + ex.getMessage());
return "";
});
System.out.println(winRateFuture.get());
2. 使用handle()处理异常
除了exceptionally()，CompletableFuture也提供了handle()来处理异常。不过，与exceptionally()不同的是，当我们在调用链中使用了handle()，那么无论是否发生异常，都会调用它。所以，在handle()方法的内部，我们需要通过 if (ex != null) 来判断是否发生了异常。
CompletableFuture < ? extends Serializable > winRateFuture = roundsFuture
.thenCombine(winRoundsFuture, (rounds, winRounds) -> {
if (rounds == 0) {
throw new RuntimeException("总场次错误");
}
DecimalFormat df = new DecimalFormat("0.00");
return df.format((float) winRounds / rounds);
}).handle((res, ex) -> {
if (ex != null) {
System.out.println("出错：" + ex.getMessage());
return "";
}
return res;
});
System.out.println(winRateFuture.get());
当然，如果我们允许某个任务发生异常而不中断整个调用链路，那么可以在其内部通过try-catch消化掉。