java并发编程(7)：CompletableFuture异步框
2019-11-02 本文已影响0人
桥头放牛娃
CompletableFuture为异步编程框架，当我们在使用线程池处理任务时，我们只能通过阻塞的Future#get()获取异步的结果，当任务处理需要的时间比较长时，效率和性能就会比较差。而CompletableFuture弥补了Future，其主要是在任务处理完成后，调用应用的回调函数，这样应用就无需通过Future#get()的方式获取处理结果，而是通过任务的回调来通知应用结果，这样极大的提高了应用的效率。同时CompletableFuture还提供了任务串行、并行等处理，方便了任务的异步逻辑组合。
1、CompletableFuture继承关系
CompletableFuture继承关系.png
CompletableFuture主要继承于Future接口及CompletionStage接口，Future为异步结果接口，CompletionStage定义了CompletableFuture异步处理及依赖接口。
2、Completion继承关系
Completion为CompletableFuture的任务依赖堆，保存了当前CompletableFuture依赖的任务。其继承于ForkJoinTask，主要继承结构如下：
Completion继承关系.png
UniCompletion为基础抽象类，其包含了任务的线程池信息、依赖任务及任务执行体。
2.1、Completion解析
abstract static class Completion extends ForkJoinTask<Void>
implements Runnable, AsynchronousCompletionTask {
//堆中的下个任务
volatile Completion next; // Treiber stack link
//执行被触发的任务，返回需要传播的依赖任务
abstract CompletableFuture<?> tryFire(int mode);
//任务是否可触发
abstract boolean isLive();
public final void run() { tryFire(ASYNC); }
public final boolean exec() { tryFire(ASYNC); return true; }
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
}
2.2、UniCompletion解析
abstract static class UniCompletion<T,V> extends Completion {
//执行当前任务的线程池
Executor executor; // executor to use (null if none)
//当然依赖的任务
CompletableFuture<V> dep; // the dependent to complete
//当前任务的执行实体
CompletableFuture<T> src; // source for action
UniCompletion(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src) {
this.executor = executor; this.dep = dep; this.src = src;
}
//若当前任务可执行，则返回true。若异步执行，则提交当前任务
final boolean claim() {
Executor e = executor;
if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {
if (e == null)
return true;
executor = null; // disable
e.execute(this);
}
return false;
}
final boolean isLive() { return dep != null; }
}
2.3、BiCompletion解析
BiCompletion主要增加了一个任务。
abstract static class BiCompletion<T,U,V> extends UniCompletion<T,V> {
CompletableFuture<U> snd; // second source for action
BiCompletion(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src, CompletableFuture<U> snd) {
super(executor, dep, src); this.snd = snd;
}
}
3、主要方法详解
3.1、工厂方法创建CompletableFuture
CompletableFuture的工厂方法方便用户创建及使用CompletableFuture。主要分为两类，执行有返回值的任务（Callable）和无返回值的任务（Runnable）
//在线程池中异步执行一个有返回值的任务，返回结果封装在CompletableFuture中，
//任务体为supplier中
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
return asyncSupplyStage(asyncPool, supplier);
}
//在线程池中异步执行一个有返回值的任务，返回结果封装在CompletableFuture中，
//显式提供线程池executor
//任务体为supplier中
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
Executor executor) {
return asyncSupplyStage(screenExecutor(executor), supplier);
}
//在线程池中异步执行一个无返回值的任务，返回结果封装在CompletableFuture中;
//任务体为supplier中
public static CompletableFuture<Void> runAsync(Runnable runnable) {
return asyncRunStage(asyncPool, runnable);
}
//在线程池中异步执行一个无返回值的任务，返回结果封装在CompletableFuture中;
//显式提供线程池executor
//任务体为supplier中
public static CompletableFuture<Void> runAsync(Runnable runnable,
Executor executor) {
return asyncRunStage(screenExecutor(executor), runnable);
}
//获取一个已完成的CompletableFuture，并用value作为结果。
//任务体为supplier中
public static <U> CompletableFuture<U> completedFuture(U value) {
return new CompletableFuture<U>((value == null) ? NIL : value);
}
//执行有返回值的任务，主要是将任务封装为一个AsyncSupply并交由线程池执行
static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
Supplier<U> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<U> d = new CompletableFuture<U>();
e.execute(new AsyncSupply<U>(d, f));
return d;
}
//执行无返回值的任务，主要是将任务封装为一个AsyncRun并交由线程池执行
static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
if (f == null) throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
e.execute(new AsyncRun(d, f));
return d;
}
AsyncSupply及AsyncRun实现：
//封装的task，用于执行无返回值的任务
static final class AsyncRun extends ForkJoinTask<Void>
implements Runnable, AsynchronousCompletionTask {
//dep：当前任务的异步执行结果的Future；fn：当前任务的执行体，函数式编程
CompletableFuture<Void> dep; Runnable fn;
AsyncRun(CompletableFuture<Void> dep, Runnable fn) {
this.dep = dep; this.fn = fn;
}
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
public final boolean exec() { run(); return true; }
public void run() {
CompletableFuture<Void> d; Runnable f;
if ((d = dep) != null && (f = fn) != null) {
dep = null; fn = null;
//任务未执行结束？
if (d.result == null) {
try {
//执行任务
f.run();
//设置执行结果问null的AltResult
d.completeNull();
} catch (Throwable ex) {
//若异常则设置异常结果
d.completeThrowable(ex);
}
}
//传播任务完成的消息，执行所有依赖此任务的其他任务，依赖任务存储在栈中
d.postComplete();
}
}
}
//封装的task，用于执行有返回值的任务
static final class AsyncSupply<T> extends ForkJoinTask<Void>
implements Runnable, AsynchronousCompletionTask {
//dep：当前任务的异步执行结果的Future；fn：当前任务的执行体，函数式编程
CompletableFuture<T> dep; Supplier<T> fn;
AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
this.dep = dep; this.fn = fn;
}
public final Void getRawResult() { return null; }
public final void setRawResult(Void v) {}
public final boolean exec() { run(); return true; }
public void run() {
CompletableFuture<T> d; Supplier<T> f;
if ((d = dep) != null && (f = fn) != null) {
dep = null; fn = null;
//任务未执行？
if (d.result == null) {
try {
//执行任务，并设置任务的执行结果
d.completeValue(f.get());
} catch (Throwable ex) {
//执行异常则设置异常结果
d.completeThrowable(ex);
}
}
//传播任务完成的消息，执行所有依赖此任务的其他任务，依赖任务存储在栈中
d.postComplete();
}
}
}
final void postComplete() {
//f:当前CompletableFuture
CompletableFuture<?> f = this; Completion h;
//当前CompletableFuture的依赖栈不为空；
//或当f的stack为空时，使f重新指向当前的CompletableFuture，继续后面的结点
//一次执行一个依赖任务的处理
while ((h = f.stack) != null ||
(f != this && (h = (f = this).stack) != null)) {
CompletableFuture<?> d; Completion t;
//更新堆的头节点下个节点
if (f.casStack(h, t = h.next)) {
//头结点的下个节点不为空？
if (t != null) {
// 如果f不是当前CompletableFuture，则将它的头结点压入到当前CompletableFuture的stack中，
// 使树形结构变成链表结构，避免递归层次过深
if (f != this) {
// 继续下一个结点，批量压入到当前栈中
pushStack(h);
continue;
}
// 如果是当前CompletableFuture, 解除头节点与栈的联系
h.next = null; // detach
}
// 调用头节点的tryFire()方法，该方法可看作Completion的钩子方法，
// 执行完逻辑后，会向后传播的
f = (d = h.tryFire(NESTED)) == null ? this : d;
}
}
}
3.2、获取CompletableFuture的异步结果
CompletableFuture继承于Future，实现了获取异步执行结果的一些方法。
//异步任务是否已经完成
public boolean isDone() {
return result != null;
}
//获取异步的执行结果，若任务未执行完成，则阻塞等待；
//若执行结果中有异常，则直接抛出异常
public T get() throws InterruptedException, ExecutionException {
Object r;
return reportGet((r = result) == null ? waitingGet(true) : r);
}
//在给定的超时时间内获取异步结果
public T get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
Object r;
long nanos = unit.toNanos(timeout);
return reportGet((r = result) == null ? timedGet(nanos) : r);
}
//阻塞等待任务执行完成并获取任务结果
public T join() {
Object r;
return reportJoin((r = result) == null ? waitingGet(false) : r);
}
//立即获取执行结果，若任务还未执行完成则直接使用给定的默认值，否则返回结果；
//若执行结果中有异常，则直接抛出异常
public T getNow(T valueIfAbsent) {
Object r;
return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
}
//获取异步执行结果，若结果有异常，则直接抛出异常
private static <T> T reportJoin(Object r) {
if (r instanceof AltResult) {
Throwable x;
if ((x = ((AltResult)r).ex) == null)
return null;
if (x instanceof CancellationException)
throw (CancellationException)x;
if (x instanceof CompletionException)
throw (CompletionException)x;
throw new CompletionException(x);
}
@SuppressWarnings("unchecked") T t = (T) r;
return t;
}
3.3、计算结果完成后的相关处理（UniWhenComplete）
当CompletableFuture计算结果完成时,我们需要对结果进行处理，或者当CompletableFuture产生异常的时候需要对异常进行处理。方法中以Async结尾的会在新的线程池中执行，没有Async结尾的会在之前的CompletableFuture执行的线程中执行。
//当完成后同步执行action
public CompletableFuture<T> whenComplete(
BiConsumer<? super T, ? super Throwable> action) {
return uniWhenCompleteStage(null, action);
}
//完成后异步执行action
public CompletableFuture<T> whenCompleteAsync(
BiConsumer<? super T, ? super Throwable> action) {
return uniWhenCompleteStage(asyncPool, action);
}
//完成后异步执行action，带线程池
public CompletableFuture<T> whenCompleteAsync(
BiConsumer<? super T, ? super Throwable> action, Executor executor) {
return uniWhenCompleteStage(screenExecutor(executor), action);
}
//异常后执行
public CompletableFuture<T> exceptionally(
Function<Throwable, ? extends T> fn) {
return uniExceptionallyStage(fn);
}
uniWhenCompleteStage相关处理：
private CompletableFuture<T> uniWhenCompleteStage(
Executor e, BiConsumer<? super T, ? super Throwable> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<T> d = new CompletableFuture<T>();
//若线程池为空，则调用uniWhenComplete方法进行任务状态判断及处理
//若线程池非空，则构建UniWhenComplete任务并将任务入队，同时调用tryFire()进行同步处理
if (e != null || !d.uniWhenComplete(this, f, null)) {
UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);
push(c);
//调用钩子方法，对任务进行处理并处理相关依赖
c.tryFire(SYNC);
}
return d;
}
final boolean uniWhenComplete(CompletableFuture<T> a,
BiConsumer<? super T,? super Throwable> f,
UniWhenComplete<T> c) {
Object r; T t; Throwable x = null;
//检查依赖的任务是否完成，未完成则直接返回false
if (a == null || (r = a.result) == null || f == null)
return false;
//当前任务未完成？
if (result == null) {
try {
//uniWhenComplete中所有c都为空，无需考虑
if (c != null && !c.claim())
return false;
//判断执行结果是否异常
if (r instanceof AltResult) {
x = ((AltResult)r).ex;
t = null;
} else {
@SuppressWarnings("unchecked") T tr = (T) r;
t = tr;
}
//执行任务
f.accept(t, x);
if (x == null) {
internalComplete(r);
return true;
}
} catch (Throwable ex) {
if (x == null)
x = ex;
}
//设置异常结果
completeThrowable(x, r);
}
return true;
}
//whenComplete任务的封装
static final class UniWhenComplete<T> extends UniCompletion<T,T> {
BiConsumer<? super T, ? super Throwable> fn;
UniWhenComplete(Executor executor, CompletableFuture<T> dep,
CompletableFuture<T> src,
BiConsumer<? super T, ? super Throwable> fn) {
super(executor, dep, src); this.fn = fn;
}
final CompletableFuture<T> tryFire(int mode) {
CompletableFuture<T> d; CompletableFuture<T> a;
if ((d = dep) == null ||
!d.uniWhenComplete(a = src, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
3.4、计算结果完成时的转换处理（thenApply）
计算结果完成时的转换的处理会将上个计算结果转换为当前任务的输入参数。Async结尾的方法由原来的线程计算，以Async结尾的方法由默认的线程池ForkJoinPool.commonPool()或者指定的线程池executor运行。
//
public <U> CompletableFuture<U> thenApply(
Function<? super T,? extends U> fn) {
return uniApplyStage(null, fn);
}
public <U> CompletableFuture<U> thenApplyAsync(
Function<? super T,? extends U> fn) {
return uniApplyStage(asyncPool, fn);
}
public <U> CompletableFuture<U> thenApplyAsync(
Function<? super T,? extends U> fn, Executor executor) {
return uniApplyStage(screenExecutor(executor), fn);
}
uniApplyStage()处理解析：
private <V> CompletableFuture<V> uniApplyStage(
Executor e, Function<? super T,? extends V> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
//当线程池为空时，直接调用uniApply对任务进行处理
//当线程池非空时，将任务加入堆栈，并调用tryFire对任务进行处理
if (e != null || !d.uniApply(this, f, null)) {
UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
push(c);
c.tryFire(SYNC);
}
return d;
}
final <S> boolean uniApply(CompletableFuture<S> a,
Function<? super S,? extends T> f,
UniApply<S,T> c) {
Object r; Throwable x;
//依赖任务未完成？直接返回false
if (a == null || (r = a.result) == null || f == null)
return false;
//当前任务未完成？
tryComplete: if (result == null) {
//依赖的任务处理异常？则设置当前任务异常结果
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
completeThrowable(x, r);
break tryComplete;
}
r = null;
}
try {
//thenApply中所有c都为空，无需考虑
if (c != null && !c.claim())
return false;
@SuppressWarnings("unchecked") S s = (S) r;
//执行任务并设置任务结果
completeValue(f.apply(s));
} catch (Throwable ex) {
//执行任务异常则设置异常结果
completeThrowable(ex);
}
}
return true;
}
//thenApply任务的封装
static final class UniApply<T,V> extends UniCompletion<T,V> {
Function<? super T,? extends V> fn;
UniApply(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src,
Function<? super T,? extends V> fn) {
super(executor, dep, src); this.fn = fn;
}
final CompletableFuture<V> tryFire(int mode) {
CompletableFuture<V> d; CompletableFuture<T> a;
if ((d = dep) == null ||
!d.uniApply(a = src, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
3.5、计算结果完成时的消费处理（thenAccept）
计算结果完成时的消费的处理是将上一步任务处理的结果作为本次任务处理的输入参数，并且thenAccept的处理只会对上一步的结果进行处理，而不会返回任何处理结果。
public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
return uniAcceptStage(null, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
return uniAcceptStage(asyncPool, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
Executor executor) {
return uniAcceptStage(screenExecutor(executor), action);
}
uniAcceptStage()的处理流程：
private CompletableFuture<Void> uniAcceptStage(Executor e,
Consumer<? super T> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
//线程池为空则调用uniAccept同步处理任务；
//线程池非空则将任务封装为UniAccept并推入堆栈，同时调用tryFire()进行任务处理
if (e != null || !d.uniAccept(this, f, null)) {
UniAccept<T> c = new UniAccept<T>(e, d, this, f);
push(c);
c.tryFire(SYNC);
}
return d;
}
final <S> boolean uniAccept(CompletableFuture<S> a,
Consumer<? super S> f, UniAccept<S> c) {
Object r; Throwable x;
//依赖任务未完成？直接返回false
if (a == null || (r = a.result) == null || f == null)
return false;
//当前任务未完成？
tryComplete: if (result == null) {
//依赖任务结果异常，则设置当前的异常结果
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
completeThrowable(x, r);
break tryComplete;
}
r = null;
}
try {
//uniAccept中c全部为null，无需考虑
if (c != null && !c.claim())
return false;
@SuppressWarnings("unchecked") S s = (S) r;
//执行当前任务
f.accept(s);
//设置空结果
completeNull();
} catch (Throwable ex) {
//执行异常则设置异常结果
completeThrowable(ex);
}
}
return true;
}
//uniAccept的任务封装
static final class UniAccept<T> extends UniCompletion<T,Void> {
Consumer<? super T> fn;
UniAccept(Executor executor, CompletableFuture<Void> dep,
CompletableFuture<T> src, Consumer<? super T> fn) {
super(executor, dep, src); this.fn = fn;
}
final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d; CompletableFuture<T> a;
if ((d = dep) == null ||
!d.uniAccept(a = src, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
3.6、多个结果完成时消费（thenAcceptBoth、runAfterBoth）
多个结果完成时处理会等待当前结果CompletableFuture及依赖的other完成时执行action，thenAcceptBoth会将依赖的当前CompletableFuture及other的执行结果作为action的输入参数。
runAfterBoth则只等待两个依赖的任务执行完成再执行。
public <U> CompletableFuture<Void> thenAcceptBoth(
CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action) {
return biAcceptStage(null, other, action);
}
public <U> CompletableFuture<Void> thenAcceptBothAsync(
CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action) {
return biAcceptStage(asyncPool, other, action);
}
public <U> CompletableFuture<Void> thenAcceptBothAsync(
CompletionStage<? extends U> other,
BiConsumer<? super T, ? super U> action, Executor executor) {
return biAcceptStage(screenExecutor(executor), other, action);
}
public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
Runnable action) {
return biRunStage(null, other, action);
}
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
Runnable action) {
return biRunStage(asyncPool, other, action);
}
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,
Runnable action,
Executor executor) {
return biRunStage(screenExecutor(executor), other, action);
}
biAcceptStage()及biRunStage()的处理流程基本相同，不同点为biAcceptStage()会将依赖的两个任务作为执行处理的入参，而biRunStage()不会。
以下是biAcceptStage()的处理流程：
//处理thenAcceptBoth类型的任务
//e:线程池；
//o：依赖的一个任务
//f：具体执行逻辑，会将当前CompletableFuture及o的执行结果作为输入
private <U> CompletableFuture<Void> biAcceptStage(
Executor e, CompletionStage<U> o,
BiConsumer<? super T,? super U> f) {
CompletableFuture<U> b;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
//线程池为空时，调用biAccept()同步执行处理
//线程池非空，则将任务封装为BiAccept并推入堆栈，调用tryFire()进行任务处理
if (e != null || !d.biAccept(this, b, f, null)) {
BiAccept<T,U> c = new BiAccept<T,U>(e, d, this, b, f);
bipush(b, c);
c.tryFire(SYNC);
}
return d;
}
//任务的同步处理
//a,b：依赖的任务
//f：具体执行逻辑
//c：为空，无需考虑
final <R,S> boolean biAccept(CompletableFuture<R> a,
CompletableFuture<S> b,
BiConsumer<? super R,? super S> f,
BiAccept<R,S> c) {
Object r, s; Throwable x;
//判断依赖的任务a，b是否执行完毕
if (a == null || (r = a.result) == null ||
b == null || (s = b.result) == null || f == null)
return false;
//当前任务未执行完成?
tryComplete: if (result == null) {
//若a或b任务执行有异常，则设置当前任务的异常结果
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
completeThrowable(x, r);
break tryComplete;
}
r = null;
}
if (s instanceof AltResult) {
if ((x = ((AltResult)s).ex) != null) {
completeThrowable(x, s);
break tryComplete;
}
s = null;
}
try {
if (c != null && !c.claim())
return false;
@SuppressWarnings("unchecked") R rr = (R) r;
@SuppressWarnings("unchecked") S ss = (S) s;
//执行任务，并将a，b任务的执行结果作为参数输入
f.accept(rr, ss);
//设置返回结果为null
completeNull();
} catch (Throwable ex) {
//任务执行异常则设置异常结果
completeThrowable(ex);
}
}
return true;
}
//thenAcceptBoth类型任务的封装
static final class BiAccept<T,U> extends BiCompletion<T,U,Void> {
BiConsumer<? super T,? super U> fn;
BiAccept(Executor executor, CompletableFuture<Void> dep,
CompletableFuture<T> src, CompletableFuture<U> snd,
BiConsumer<? super T,? super U> fn) {
super(executor, dep, src, snd); this.fn = fn;
}
final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
if ((d = dep) == null ||
!d.biAccept(a = src, b = snd, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
3.7、某个结果完成时消费（applyToEither、acceptEither，runAfterEither）
applyToEither及acceptEither会将两个结果中任意一个的执行结果作为当前执行的输入参数，而applyToEither会返回执行结果，acceptEither则返回空的执行结果。runAfterEither则不会将依赖的执行结果作为参数，其只是当依赖的任意一个任务完成时进行处理，并返回空的执行结果。
public <U> CompletableFuture<U> applyToEither(
CompletionStage<? extends T> other, Function<? super T, U> fn) {
return orApplyStage(null, other, fn);
}
public <U> CompletableFuture<U> applyToEitherAsync(
CompletionStage<? extends T> other, Function<? super T, U> fn) {
return orApplyStage(asyncPool, other, fn);
}
public <U> CompletableFuture<U> applyToEitherAsync(
CompletionStage<? extends T> other, Function<? super T, U> fn,
Executor executor) {
return orApplyStage(screenExecutor(executor), other, fn);
}
public CompletableFuture<Void> acceptEither(
CompletionStage<? extends T> other, Consumer<? super T> action) {
return orAcceptStage(null, other, action);
}
public CompletableFuture<Void> acceptEitherAsync(
CompletionStage<? extends T> other, Consumer<? super T> action) {
return orAcceptStage(asyncPool, other, action);
}
public CompletableFuture<Void> acceptEitherAsync(
CompletionStage<? extends T> other, Consumer<? super T> action,
Executor executor) {
return orAcceptStage(screenExecutor(executor), other, action);
}
public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
Runnable action) {
return orRunStage(null, other, action);
}
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
Runnable action) {
return orRunStage(asyncPool, other, action);
}
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,
Runnable action,
Executor executor) {
return orRunStage(screenExecutor(executor), other, action);
}
orApplyStage()、orAcceptStage()、orRunStage()的处理基本相同。以下以orApplyStage()为例来分析其处理流程：
private <U extends T,V> CompletableFuture<V> orApplyStage(
Executor e, CompletionStage<U> o,
Function<? super T, ? extends V> f) {
CompletableFuture<U> b;
if (f == null || (b = o.toCompletableFuture()) == null)
throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
//若线程池为空，则调用orApply()进行任务的同步处理
//若线程池非空，则将依赖及处理封装为OrApply并推入堆栈，然后调用tryFire()进行任务处理
if (e != null || !d.orApply(this, b, f, null)) {
OrApply<T,U,V> c = new OrApply<T,U,V>(e, d, this, b, f);
orpush(b, c);
c.tryFire(SYNC);
}
return d;
}
final <R,S extends R> boolean orApply(CompletableFuture<R> a,
CompletableFuture<S> b,
Function<? super R, ? extends T> f,
OrApply<R,S,T> c) {
Object r; Throwable x;
//依赖的任务a，b都未执行完成？
if (a == null || b == null ||
((r = a.result) == null && (r = b.result) == null) || f == null)
return false;
//当前任务未完成？
tryComplete: if (result == null) {
try {
if (c != null && !c.claim())
return false;
//依赖任务处理异常，设置当前异常结果
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
completeThrowable(x, r);
break tryComplete;
}
r = null;
}
@SuppressWarnings("unchecked") R rr = (R) r;
//进行任务处理，并设置处理结果
completeValue(f.apply(rr));
} catch (Throwable ex) {
completeThrowable(ex);
}
}
return true;
}
//orApplyStage的任务封装。
static final class OrApply<T,U extends T,V> extends BiCompletion<T,U,V> {
Function<? super T,? extends V> fn;
OrApply(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src,
CompletableFuture<U> snd,
Function<? super T,? extends V> fn) {
super(executor, dep, src, snd); this.fn = fn;
}
final CompletableFuture<V> tryFire(int mode) {
CompletableFuture<V> d;
CompletableFuture<T> a;
CompletableFuture<U> b;
if ((d = dep) == null ||
!d.orApply(a = src, b = snd, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; snd = null; fn = null;
return d.postFire(a, b, mode);
}
}
3.8、异步结果的组合处理（thenCompose）
thenCompose会连接两个CompletableFuture，其处理是当前CompletableFuture完成时将结果作为fn处理的入参进行处理。
public <U> CompletableFuture<U> thenCompose(
Function<? super T, ? extends CompletionStage<U>> fn) {
return uniComposeStage(null, fn);
}
public <U> CompletableFuture<U> thenComposeAsync(
Function<? super T, ? extends CompletionStage<U>> fn) {
return uniComposeStage(asyncPool, fn);
}
public <U> CompletableFuture<U> thenComposeAsync(
Function<? super T, ? extends CompletionStage<U>> fn,
Executor executor) {
return uniComposeStage(screenExecutor(executor), fn);
}
uniComposeStage()处理流程：
private <V> CompletableFuture<V> uniComposeStage(
Executor e, Function<? super T, ? extends CompletionStage<V>> f) {
if (f == null) throw new NullPointerException();
Object r; Throwable x;
//无线程池，且当前任务处理完成
if (e == null && (r = result) != null) {
//若当前处理结果异常，则直接返回异常结果
if (r instanceof AltResult) {
if ((x = ((AltResult)r).ex) != null) {
return new CompletableFuture<V>(encodeThrowable(x, r));
}
r = null;
}
try {
//将当前处理结果作为f的输入，并执行f处理
@SuppressWarnings("unchecked") T t = (T) r;
CompletableFuture<V> g = f.apply(t).toCompletableFuture();
Object s = g.result;
//f处理完成？则直接返回处理结果
//未完成则封装处理并将任务入栈
if (s != null)
return new CompletableFuture<V>(encodeRelay(s));
CompletableFuture<V> d = new CompletableFuture<V>();
UniRelay<V> copy = new UniRelay<V>(d, g);
g.push(copy);
copy.tryFire(SYNC);
return d;
} catch (Throwable ex) {
return new CompletableFuture<V>(encodeThrowable(ex));
}
}
//当前任务未处理完成，则封装当前任务及依赖并入栈
CompletableFuture<V> d = new CompletableFuture<V>();
UniCompose<T,V> c = new UniCompose<T,V>(e, d, this, f);
push(c);
c.tryFire(SYNC);
return d;
}
3.9、等待多个执行结果完成
//所有任务都执行完毕
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
return andTree(cfs, 0, cfs.length - 1);
}
//某个任务执行完毕
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
return orTree(cfs, 0, cfs.length - 1);
}
static CompletableFuture<Object> orTree(CompletableFuture<?>[] cfs,
int lo, int hi) {
CompletableFuture<Object> d = new CompletableFuture<Object>();
//递归将任务进行遍历，若某个任务已经完成，则直接设置结果为已完成任务的结果
if (lo <= hi) {
CompletableFuture<?> a, b;
int mid = (lo + hi) >>> 1;
if ((a = (lo == mid ? cfs[lo] :
orTree(cfs, lo, mid))) == null ||
(b = (lo == hi ? a : (hi == mid+1) ? cfs[hi] :
orTree(cfs, mid+1, hi))) == null)
throw new NullPointerException();
if (!d.orRelay(a, b)) {
OrRelay<?,?> c = new OrRelay<>(d, a, b);
a.orpush(b, c);
c.tryFire(SYNC);
}
}
return d;
}
3.10、对异步结果进行处理（handle）
handle()主要获取当前任务的执行结果，并将其作为fn函数的输入参数，并接执行结果设置为返回的CompletableFuture的结果。
public <U> CompletableFuture<U> handle(
BiFunction<? super T, Throwable, ? extends U> fn) {
return uniHandleStage(null, fn);
}
public <U> CompletableFuture<U> handleAsync(
BiFunction<? super T, Throwable, ? extends U> fn) {
return uniHandleStage(asyncPool, fn);
}
public <U> CompletableFuture<U> handleAsync(
BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
return uniHandleStage(screenExecutor(executor), fn);
}
uniHandleStage()的处理流程：
private <V> CompletableFuture<V> uniHandleStage(
Executor e, BiFunction<? super T, Throwable, ? extends V> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<V> d = new CompletableFuture<V>();
//若线程池为空，则直接调用uniHandle同步执行任务，
//否则将任务及依赖信息封装为UniHandle入栈，然后调用tryFire()进行任务处理
if (e != null || !d.uniHandle(this, f, null)) {
UniHandle<T,V> c = new UniHandle<T,V>(e, d, this, f);
push(c);
c.tryFire(SYNC);
}
return d;
}
final <S> boolean uniHandle(CompletableFuture<S> a,
BiFunction<? super S, Throwable, ? extends T> f,
UniHandle<S,T> c) {
Object r; S s; Throwable x;
//依赖任务未完成？返回false
if (a == null || (r = a.result) == null || f == null)
return false;
//当前结果为完成？
if (result == null) {
try {
if (c != null && !c.claim())
return false;
//依赖执行结果异常？则设置当前结果为异常结果
if (r instanceof AltResult) {
x = ((AltResult)r).ex;
s = null;
} else {
x = null;
@SuppressWarnings("unchecked") S ss = (S) r;
s = ss;
}
//将依赖的结果作为当前函数的输入参数，并执行函数，设置当前执行结果
completeValue(f.apply(s, x));
} catch (Throwable ex) {
completeThrowable(ex);
}
}
return true;
}
