Future 是在Java 5引入的，CompletableFuture 是在Java 8引入的，提供了更加强大的功能。两者都是用来实现异步的，避免阻塞主线程.
Future
Future 提供了get 方法来返回异步任务的执行结果，当调用get方法会阻塞直至任务结束返回结果。
public class TestFuture {
public static void main(String[] args) {
ExecutorService exec = Executors.newSingleThreadExecutor();
Future<Integer> f = exec.submit(new MyCallable());
System.out.println(f.isDone()); // false
try {
System.out.println(f.get()); // 1，等待直到Callable完成
System.out.println(f.isDone()); // true
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
exec.shutdown();
}
}
}
class MyCallable implements Callable<Integer> {
@Override
public Integer call() throws Exception {
Thread.sleep(1000);
return 1;
}
}
CompletableFuture
CompletableFuture 实现了Future 和CompletionStage 接口，实现异步任务的链式处理，支持多个任务的并发执行、顺序执行，对任务的控制更加精细。
runAsync()
异步执行一个 Runnable 实例，异步任务没有返回值，该方法返回CompletableFuture实例。
CompletableFuture future = CompletableFuture.runAsync(() -> {
try {
System.out.println("Running asynchronous task in parallel");
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException ex) {
throw new IllegalStateException(ex);
}
});
supplyAsync()
异步执行一个Supplier，异步任务具有返回值，该方法返回CompletableFuture实例。
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
return "This is the result of the asynchronous computation";
});
thenApply()
执行一个 Function，这个函数的执行还是在CompletableFuture.supplyAsync 开启的线程中执行，Function有返回值所以可以链式的连接多个thenApply方法。
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "Feng");
CompletableFuture<String> cf2 = cf.thenApply(name -> "Hello " + name).thenApply(greeting -> greeting + ", Welcome to SH!");
System.out.println(cf.get()); // Feng
System.out.println(cf2.get()); // Hello Feng, Welcome to SH!
thenAccept()
执行一个 Consumer，此时执行异步任务的线程没有返回值。通常该方法应该在操作链的最后。
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "Feng");
CompletableFuture<String> cf2 = cf.thenApply(name -> "Hello " + name).thenApply(greeting -> greeting + ", Welcome to SH!");
cf2.thenAccept(v-> System.out.println(v)); // Hello Feng, Welcome to SH!
thenRun()
执行一个Runnable，和thenAccept一样，此时执行异步任务的线程没有返回值，但不同的是此时也不接收任何参数的传入。
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "Feng");
cf.thenRun(()-> System.out.println("then run.."));
thenCompose()
thenCompose方法接受一个返回CompletableFuture的Function做为参数，和thenApply不同的是：
thenApply接受是一个同步方法，而thenCompose接受的是异步方法。
public class CF2 {
private static CompletableFuture<String> sendMailAsync(String input) {
return CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return input + " | " + "2. Send mail to Administrator.";
});
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "1. Write mail.");
CompletableFuture<String> cf2 = cf.thenCompose(CF2::sendMailAsync);
System.out.println(cf2.get()); // 1. Write mail. | 2. Send mail to Administrator.
}
}
thenCombine()
thenCombine接受一个 BiFunction，来处理两个CompletableFuture的结果。
public class CF3 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "1. A");
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "2. B");
CompletableFuture<String> cf3 = cf1.thenCombine(cf2, (a, b) -> a + " - " + b);
System.out.println(cf3.get());
}
}
thenAcceptBoth()
thenAcceptBoth 接受一个 BiConsumer，来处理两个CompletableFuture的结果。
public class CF4 {
public static void main(String[] args) {
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "1. A");
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "2. B");
cf1.thenAcceptBoth(cf2, (a, b) -> System.out.println(a + b));
}
}
| 方法 | 描述 |
|---|---|
| <U, R> CompletionStage<R> thenCombine(CompletionStage<U> other, BiFunction<T, U, R> action) | Combines the result of this and other in one, using a BiFunction |
| <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<U> other, BiConsumer<T, U> action) | Consumes the result of this and other, using a BiConsumer |
| <U> CompletionStage<Void> runAfterBoth(CompletionStage<U> other, BiConsumer<T, U> action) | Triggers the execution of a Runnable on the completion of this and other |
CompletableFuture.allOf()
返回新的CompletableFuture实例，当形参中指定的CompletableFuture都完成了，该方法返回的新CompletableFuture也就完成了，如果要等待形参中指定的CompletableFuture都完成可以使用CompletableFuture.allOf(cf1, cf2, cf3).join()。
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> 1);
CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> 2);
CompletableFuture cf3 = CompletableFuture.supplyAsync(() -> 3);
CompletableFuture.allOf(cf1, cf2, cf3).join(); // All CompletableFuture success
System.out.println(cf1.get()); // 1
System.out.println(cf2.get()); // 2
System.out.println(cf3.get()); // 3
异常处理
参考如下代码，当cf1执行出现异常，所有下游的CompletableFuture都将出错，可以通过以下两个方法验证：
- isCompletedExceptionally() 会返回true
- get() 方法会抛出ExecutionException异常，异常内容是cf1中出现的异常.
public class CF6 {
public static void main(String[] args) {
CompletableFuture.supplyAsync(() -> 4 / 1).thenAccept(v -> System.out.println(v)); // 4
System.out.println("1 -------");
// 如果异常不处理，所有下游的CompletableFuture将都会出错
CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> 4 / 0);
CompletableFuture<Integer> cf2 = cf1.thenApply(v -> v++);
cf2.thenAccept(v -> System.out.println(v)); // nothing
System.out.println(cf2.isCompletedExceptionally()); // true
try {
System.out.println(cf2.get());
} catch (InterruptedException | ExecutionException e) {
System.out.println("error"); // error
}
System.out.println("2 -------");
// 通过exceptionally处理异常，提供
CompletableFuture.supplyAsync(() -> 4 / 0).exceptionally(ex -> -1).thenAccept(v -> System.out.println(v)); // -1
}
}
执行结果：
4
1 -------
true
error
2 -------
-1
异常可以通过exceptionally(Function<Throwable, T> function)方法来处理，如果出现异常，异常会被传入该方法，如果没有异常，直接返回上游的结果。
异常还可以通过handle(BiFunction<T, Throwable, R> bifunction)来处理，如果上游处理出现异常，T值将为null，异常为上游异常。如果上游正常执行，T将是上游的返回值，异常将是null.
CompletableFuture.supplyAsync(() -> 4 / 0)
.handle((v, ex) -> v != null ? v : -1)
.thenAccept(v -> System.out.println(v)); // -1
第三种是通过whenComplete(BiConsumer<T, Throwable> biconsumer)来处理异常，同handle, 如果上游处理出现异常，T值将为null，异常为上游异常。如果上游正常执行，T将是上游的返回值，异常将是null。不同的是如果出现异常，whenComplete后续将没有返回值，如果没有异常将向下返回上游的返回值。
CompletableFuture.supplyAsync(() -> 4 / 1)
.whenComplete((v, ex) -> {
if (ex == null) {
System.out.println(v); // 4
} else {
System.out.println(ex.getMessage());
}
}).thenAccept(v -> System.out.println(v)); // 4, 只有当whenComplete之前没有异常才会取到上游的返回值
CompletableFuture.supplyAsync(() -> 4 / 0)
.whenComplete((v, ex) -> {
if (ex == null) {
System.out.println(v);
} else {
System.out.println(ex.getMessage()); // java.lang.ArithmeticException: / by zero
}
}).thenAccept(v -> System.out.println(v)); // 无
参考：
https://stackoverflow.com/questions/35329845/difference-between-completablefuture-future-and-rxjavas-observable
https://blog.knoldus.com/2018/01/20/future-vs-completablefuture-1/
https://blog.knoldus.com/2018/03/30/future-vs-completablefuture-in-java-2/
http://www.deadcoderising.com/java8-writing-asynchronous-code-with-completablefuture/
https://community.oracle.com/docs/DOC-995305
