如何安全发布 CompletableFuture ？Java9新增方法分析
本文未经允许禁止转载。
JDK9 中对于CompletableFuture做了新的增强，除了（orTimeout），还有面向继承、安全发布等相关方法。本文中，我们将详细分析各个新增方法，同时说明其安全发布的重要性，最后提出相关的实践原则。
1. newIncompleteFuture
public <U> CompletableFuture<U> newIncompleteFuture() {
return new CompletableFuture<U>();
}
这是一个面向继承的方法，Java支持方法返回值协变，子类可以返回自己的实现。如果你不需要继承 CompletableFuture，对于使用来说，这个方法没有新增新的功能。
2. 默认执行器
public Executor defaultExecutor() {
return ASYNC_POOL;
}
子类可以指定默认执行器，不过需要注意调用不显式指定执行器的async相关方法时，其执行回调所在的线程有两种可能：当前调用线程或者子类指定的执行器线程。如果使用 CompletableFuture，推荐显式指定执行器；如果使用 CFFU，无需每次都显式指定执行器。
3. 复制方法
public CompletableFuture<T> copy() {
return uniCopyStage(this);
}
此方法等同于调用 cf.thenApply(x -> x), 目的是实现保护性复制，避免后续操作对于原实例的修改。
4. 只读形式与安全发布
// 返回结果仅支持 CompletionStage 操作
public CompletionStage<T> minimalCompletionStage() {
return uniAsMinimalStage();
}
private MinimalStage<T> uniAsMinimalStage() {
Object r;
// 性能优化1：若已有结果，当前线程下直接返回stage
if ((r = result) != null)
return new MinimalStage<T>(encodeRelay(r));
// 性能优化2：若还未有结果，注册回调
MinimalStage<T> d = new MinimalStage<T>();
unipush(new UniRelay<T,T>(d, this));
return d;
}
// 实际实现，CompletableFuture 独有方法均不支持
static final class MinimalStage<T> extends CompletableFuture<T> {
MinimalStage() { }
MinimalStage(Object r) { super(r); }
@Override public <U> CompletableFuture<U> newIncompleteFuture() {
return new MinimalStage<U>(); }
@Override public T get() {
throw new UnsupportedOperationException(); }
@Override public T get(long timeout, TimeUnit unit) {
throw new UnsupportedOperationException(); }
@Override public T getNow(T valueIfAbsent) {
throw new UnsupportedOperationException(); }
@Override public T join() {
throw new UnsupportedOperationException(); }
@Override public T resultNow() {
throw new UnsupportedOperationException(); }
@Override public Throwable exceptionNow() {
throw new UnsupportedOperationException(); }
@Override public boolean complete(T value) {
throw new UnsupportedOperationException(); }
@Override public boolean completeExceptionally(Throwable ex) {
throw new UnsupportedOperationException(); }
@Override public boolean cancel(boolean mayInterruptIfRunning) {
throw new UnsupportedOperationException(); }
@Override public void obtrudeValue(T value) {
throw new UnsupportedOperationException(); }
@Override public void obtrudeException(Throwable ex) {
throw new UnsupportedOperationException(); }
@Override public boolean isDone() {
throw new UnsupportedOperationException(); }
@Override public boolean isCancelled() {
throw new UnsupportedOperationException(); }
@Override public boolean isCompletedExceptionally() {
throw new UnsupportedOperationException(); }
@Override public State state() {
throw new UnsupportedOperationException(); }
@Override public int getNumberOfDependents() {
throw new UnsupportedOperationException(); }
@Override public CompletableFuture<T> completeAsync
(Supplier<? extends T> supplier, Executor executor) {
throw new UnsupportedOperationException(); }
@Override public CompletableFuture<T> completeAsync
(Supplier<? extends T> supplier) {
throw new UnsupportedOperationException(); }
@Override public CompletableFuture<T> orTimeout
(long timeout, TimeUnit unit) {
throw new UnsupportedOperationException(); }
@Override public CompletableFuture<T> completeOnTimeout
(T value, long timeout, TimeUnit unit) {
throw new UnsupportedOperationException(); }
// 返回新的CompletableFuture, 等效于thenApply(x -> x)
@Override public CompletableFuture<T> toCompletableFuture() {
Object r;
if ((r = result) != null)
return new CompletableFuture<T>(encodeRelay(r));
else {
CompletableFuture<T> d = new CompletableFuture<>();
unipush(new UniRelay<T,T>(d, this));
return d;
}
}
}
此方法返回只读的 CompletionStage，可以实现安全发布。可以类比于一个方法返回了ImmutableList，优点是可以防止后续的错误操作，比如其他线程强制写结果（类比List#add）。
有这样一个安全发布的保证至关重要，特别是面临复杂场景时，可以减轻编程负担，放心地使用 CompletableFuture 返回的结果。
我们知道CompletionStage接口定义了链式异步回调相关方法， CompletableFuture 相关读写操作（join, complete, obtrudeValue 等），不在CompletionStage的定义下，可以通过方法toCompletableFuture进行中转。
class InventoryServiceDemo {
final Executor executor = Executors.newCachedThreadPool();
// 获取库存信息
// 方法可以返回 CompletionStage，相比于CompletableFuture更安全
CompletionStage<Integer> getInventoryAsync(String productId) {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(Duration.ofSeconds(2));
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return 42;
}, executor)
.minimalCompletionStage();
}
}
5. 异步写操作
public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier) {
return completeAsync(supplier, defaultExecutor());
}
public CompletableFuture<T> completeAsync(Supplier<? extends T> supplier,
Executor executor) {
if (supplier == null || executor == null)
throw new NullPointerException();
executor.execute(new AsyncSupply<T>(this, supplier));
return this;
}
这两个方法很容易理解，适用于子类返回自己的实现。CompletableFuture#supplyAsync 工厂方法返回的实例是 ComplteableFuture 无法更改。子类可以自己定义自己的工厂方法；用户也可以使用这里的异步写方法。
6. CFFU的优化拓展
开源项目CFFU（功夫未来）对于CompletableFuture 进行了拓展，以下仅列举和Java9新增功能相关的拓展点：
- 可以在对象创建时指定执行器，后续操作无需重复指定
- Java8 版本下可以使用后续版本的功能
- 提供了相关安全发布功能，比如
- 内部的很多便利方法应用了保护性复制方法
- CffuFactoryBuilder 支持以参数形式配置是否允许强制重写结果
7. 实践原则
- 异步方法可以返回 CompletionStage，其相比于CompletableFuture更为安全，同时可以提示用户不要使用阻塞相关方法
- 就像推荐使用不可变对象一样，推荐默认使用 CompletableFuture#minimalCompletionStage 方法
- 无论是对于方法入参还是返回结果，使用接口一般都优于实体类
- 理想的 Future 设计应该是读写分离的，不可变的。虽然 CompletableFuture 并没有完全遵循以上原则，但是我们在使用时应当注意遵循。