社区微信群开通啦,扫一扫抢先加入社区官方微信群
社区微信群
深入学习java源码之CompletableFuture.reportGet()与CompletableFuture.supplyAsync()
异步计算
所谓异步调用其实就是实现一个可无需等待被调用函数的返回值而让操作继续运行的方法。在 Java 语言中,简单的讲就是另启一个线程来完成调用中的部分计算,使调用继续运行或返回,而不需要等待计算结果。但调用者仍需要取线程的计算结果。
JDK5新增了Future接口,用于描述一个异步计算的结果。虽然 Future 以及相关使用方法提供了异步执行任务的能力,但是对于结果的获取却是很不方便,只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背,轮询的方式又会耗费无谓的 CPU 资源,而且也不能及时地得到计算结果。
Future 接口的局限性
Future接口可以构建异步应用,但依然有其局限性。它很难直接表述多个Future 结果之间的依赖性。实际开发中,我们经常需要达成以下目的:
将多个异步计算的结果合并成一个
等待Future集合中的所有任务都完成
Future完成事件(即,任务完成以后触发执行动作)
函数式编程
CompletionStage
CompletionStage代表异步计算过程中的某一个阶段,一个阶段完成以后可能会触发另外一个阶段
一个阶段的计算执行可以是一个Function,Consumer或者Runnable。比如:stage.thenApply(x -> square(x)).thenAccept(x -> System.out.print(x)).thenRun(() -> System.out.println())
一个阶段的执行可能是被单个阶段的完成触发,也可能是由多个阶段一起触发
CompletableFuture
创建CompletableFuture
thenApply
当前阶段正常完成以后执行,而且当前阶段的执行的结果会作为下一阶段的输入参数。thenApplyAsync默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。
thenApply相当于回调函数(callback)
thenAccept与thenRun
同样是执行指定的动作,同样是消耗,二者也有区别:
thenAccept接收上一阶段的输出作为本阶段的输入
例如,此阶段与其它阶段一起完成,进而触发下一阶段:
whenComplete
例子
事实上,如果每个操作都很简单的话(比如:上面的例子中按照id去查)没有必要用这种多线程异步的方式,因为创建线程还需要时间,还不如直接同步执行来得快。
事实证明,只有当每个操作很复杂需要花费相对很长的时间(比如,调用多个其它的系统的接口;比如,商品详情页面这种需要从多个系统中查数据显示的)的时候用CompletableFuture才合适,不然区别真的不大,还不如顺序同步执行。
java源码
Modifier and Type | Method and Description |
---|---|
CompletableFuture<Void> | acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,执行相应的结果作为提供的操作的参数。 |
CompletableFuture<Void> | acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,将使用此阶段的默认异步执行工具执行,其中相应的结果作为提供的操作的参数。 |
CompletableFuture<Void> | acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,将使用提供的执行器执行,其中相应的结果作为参数提供给函数。 |
static CompletableFuture<Void> | allOf(CompletableFuture<?>... cfs) 返回一个新的CompletableFuture,当所有给定的CompletableFutures完成时,完成。 |
static CompletableFuture<Object> | anyOf(CompletableFuture<?>... cfs) 返回一个新的CompletableFuture,当任何一个给定的CompletableFutures完成时,完成相同的结果。 |
<U> CompletableFuture<U> | applyToEither(CompletionStage<? extends T> other, Function<? super T,U> fn) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,执行相应的结果作为提供的函数的参数。 |
<U> CompletableFuture<U> | applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,将使用此阶段的默认异步执行工具执行,其中相应的结果作为提供函数的参数。 |
<U> CompletableFuture<U> | applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T,U> fn, Executor executor) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,将使用提供的执行器执行,其中相应的结果作为参数提供给函数。 |
boolean | cancel(boolean mayInterruptIfRunning) 如果尚未完成,请使用 |
boolean | complete(T value) 如果不是已经完成,将返回的值 |
static <U> CompletableFuture<U> | completedFuture(U value) 返回已经使用给定值完成的新的CompletableFuture。 |
boolean | completeExceptionally(Throwable ex) 如果尚未完成,则调用 |
CompletableFuture<T> | exceptionally(Function<Throwable,? extends T> fn) 返回一个新的CompletableFuture,当CompletableFuture完成时完成,结果是异常触发此CompletableFuture的完成特殊功能的给定功能; 否则,如果此CompletableFuture正常完成,则返回的CompletableFuture也会以相同的值正常完成。 |
T | get() 等待这个未来完成的必要,然后返回结果。 |
T | get(long timeout, TimeUnit unit) 如果有必要等待这个未来完成的给定时间,然后返回其结果(如果有的话)。 |
T | getNow(T valueIfAbsent) 如果已完成,则返回结果值(或抛出任何遇到的异常),否则返回给定的值IfAbsent。 |
int | getNumberOfDependents() 返回完成等待完成此CompletableFuture的CompletableFutures的估计数。 |
<U> CompletableFuture<U> | handle(BiFunction<? super T,Throwable,? extends U> fn) 返回一个新的CompletionStage,当此阶段正常或异常完成时,将使用此阶段的结果和异常作为所提供函数的参数执行。 |
<U> CompletableFuture<U> | handleAsync(BiFunction<? super T,Throwable,? extends U> fn) 返回一个新的CompletionStage,当该阶段完成正常或异常时,将使用此阶段的默认异步执行工具执行,此阶段的结果和异常作为提供函数的参数。 |
<U> CompletableFuture<U> | handleAsync(BiFunction<? super T,Throwable,? extends U> fn, Executor executor) 返回一个新的CompletionStage,当此阶段完成正常或异常时,将使用提供的执行程序执行此阶段的结果和异常作为提供的函数的参数。 |
boolean | isCancelled() 如果此CompletableFuture在正常完成之前被取消,则返回 |
boolean | isCompletedExceptionally() 如果此CompletableFuture以任何方式完成异常完成,则返回 |
boolean | isDone() 退货 |
T | join() 完成后返回结果值,如果完成异常,则返回(未检查)异常。 |
void | obtrudeException(Throwable ex) 强制导致后续调用方法 |
void | obtrudeValue(T value) 强制设置或重置随后方法返回的值 |
CompletableFuture<Void> | runAfterBoth(CompletionStage<?> other, Runnable action) 返回一个新的CompletionStage,当这个和另一个给定的阶段都正常完成时,执行给定的动作。 |
CompletableFuture<Void> | runAfterBothAsync(CompletionStage<?> other, Runnable action) 返回一个新的CompletionStage,当这个和另一个给定阶段正常完成时,使用此阶段的默认异步执行工具执行给定的操作。 |
CompletableFuture<Void> | runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) 返回一个新CompletionStage,当这和其他特定阶段正常完成,使用附带的执行见执行给定的动作 |
CompletableFuture<Void> | runAfterEither(CompletionStage<?> other, Runnable action) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,执行给定的操作。 |
CompletableFuture<Void> | runAfterEitherAsync(CompletionStage<?> other, Runnable action) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,使用此阶段的默认异步执行工具执行给定的操作。 |
CompletableFuture<Void> | runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) 返回一个新的CompletionStage,当这个或另一个给定阶段正常完成时,使用提供的执行器执行给定的操作。 |
static CompletableFuture<Void> | runAsync(Runnable runnable) 返回一个新的CompletableFuture,它在运行给定操作后由运行在 |
static CompletableFuture<Void> | runAsync(Runnable runnable, Executor executor) 返回一个新的CompletableFuture,它在运行给定操作之后由在给定执行程序中运行的任务异步完成。 |
static <U> CompletableFuture<U> | supplyAsync(Supplier<U> supplier) 返回一个新的CompletableFuture,它通过在 |
static <U> CompletableFuture<U> | supplyAsync(Supplier<U> supplier, Executor executor) 返回一个新的CompletableFuture,由给定执行器中运行的任务异步完成,并通过调用给定的供应商获得的值。 |
CompletableFuture<Void> | thenAccept(Consumer<? super T> action) 返回一个新的CompletionStage,当此阶段正常完成时,将以该阶段的结果作为提供的操作的参数执行。 |
CompletableFuture<Void> | thenAcceptAsync(Consumer<? super T> action) 返回一个新的CompletionStage,当此阶段正常完成时,将使用此阶段的默认异步执行工具执行,此阶段的结果作为提供的操作的参数。 |
CompletableFuture<Void> | thenAcceptAsync(Consumer<? super T> action, Executor executor) 返回一个新的CompletionStage,当此阶段正常完成时,将使用提供的执行程序执行此阶段的结果作为提供的操作的参数。 |
<U> CompletableFuture<Void> | thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action) 返回一个新的CompletionStage,当这个和另一个给定的阶段都正常完成时,两个结果作为提供的操作的参数被执行。 |
<U> CompletableFuture<Void> | thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action) 返回一个新的CompletionStage,当这个和另一个给定阶段正常完成时,将使用此阶段的默认异步执行工具执行,其中两个结果作为提供的操作的参数。 |
<U> CompletableFuture<Void> | thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T,? super U> action, Executor executor) 返回一个新的CompletionStage,当这个和另一个给定阶段正常完成时,使用提供的执行器执行,其中两个结果作为提供的函数的参数。 |
<U> CompletableFuture<U> | thenApply(Function<? super T,? extends U> fn) 返回一个新的CompletionStage,当此阶段正常完成时,将以该阶段的结果作为所提供函数的参数执行。 |
<U> CompletableFuture<U> | thenApplyAsync(Function<? super T,? extends U> fn) 返回一个新的CompletionStage,当该阶段正常完成时,将使用此阶段的默认异步执行工具执行此阶段的结果作为所提供函数的参数。 |
<U> CompletableFuture<U> | thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) 返回一个新的CompletionStage,当此阶段正常完成时,将使用提供的执行程序执行此阶段的结果作为提供函数的参数。 |
<U,V> CompletableFuture<V> | thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) 返回一个新的CompletionStage,当这个和另一个给定的阶段都正常完成时,两个结果作为提供函数的参数执行。 |
<U,V> CompletableFuture<V> | thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) 返回一个新的CompletionStage,当这个和另一个给定阶段正常完成时,将使用此阶段的默认异步执行工具执行,其中两个结果作为提供函数的参数。 |
<U,V> CompletableFuture<V> | thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor) 返回一个新的CompletionStage,当这个和另一个给定阶段正常完成时,使用提供的执行器执行,其中两个结果作为提供的函数的参数。 |
<U> CompletableFuture<U> | thenCompose(Function<? super T,? extends CompletionStage<U>> fn) 返回一个新的CompletionStage,当这个阶段正常完成时,这个阶段将作为提供函数的参数执行。 |
<U> CompletableFuture<U> | thenComposeAsync(Function<? super T,? extends CompletionStage<U>> fn) 返回一个新的CompletionStage,当此阶段正常完成时,将使用此阶段的默认异步执行工具执行,此阶段作为提供的函数的参数。 |
<U> CompletableFuture<U> | thenComposeAsync(Function<? super T,? extends CompletionStage<U>> fn, Executor executor) 返回一个新的CompletionStage,当此阶段正常完成时,将使用提供的执行程序执行此阶段的结果作为提供函数的参数。 |
CompletableFuture<Void> | thenRun(Runnable action) 返回一个新的CompletionStage,当此阶段正常完成时,执行给定的操作。 |
CompletableFuture<Void> | thenRunAsync(Runnable action) 返回一个新的CompletionStage,当此阶段正常完成时,使用此阶段的默认异步执行工具执行给定的操作。 |
CompletableFuture<Void> | thenRunAsync(Runnable action, Executor executor) 返回一个新的CompletionStage,当此阶段正常完成时,使用提供的执行程序执行给定的操作。 |
CompletableFuture<T> | toCompletableFuture() 返回此CompletableFuture |
String | toString() 返回一个标识此CompletableFuture的字符串及其完成状态。 |
CompletableFuture<T> | whenComplete(BiConsumer<? super T,? super Throwable> action) 返回与此阶段相同的结果或异常的新的CompletionStage,当此阶段完成时,使用结果(或 |
CompletableFuture<T> | whenCompleteAsync(BiConsumer<? super T,? super Throwable> action) 返回一个与此阶段相同结果或异常的新CompletionStage,当此阶段完成时,执行给定操作将使用此阶段的默认异步执行工具执行给定操作,结果(或 |
CompletableFuture<T> | whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor) 返回与此阶段相同的结果或异常的新的CompletionStage,当此阶段完成时,使用提供的执行者执行给定的操作,如果没有,则使用结果(或 |
package java.util.concurrent;
import java.util.function.Supplier;
import java.util.function.Consumer;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.BiFunction;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.locks.LockSupport;
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
volatile Object result; // Either the result or boxed AltResult
volatile Completion stack; // Top of Treiber stack of dependent actions
final boolean internalComplete(Object r) { // CAS from null to r
return UNSAFE.compareAndSwapObject(this, RESULT, null, r);
}
final boolean casStack(Completion cmp, Completion val) {
return UNSAFE.compareAndSwapObject(this, STACK, cmp, val);
}
/** Returns true if successfully pushed c onto stack. */
final boolean tryPushStack(Completion c) {
Completion h = stack;
lazySetNext(c, h);
return UNSAFE.compareAndSwapObject(this, STACK, h, c);
}
final void pushStack(Completion c) {
do {} while (!tryPushStack(c));
}
/* ------------- Encoding and decoding outcomes -------------- */
static final class AltResult { // See above
final Throwable ex; // null only for NIL
AltResult(Throwable x) { this.ex = x; }
}
/** The encoding of the null value. */
static final AltResult NIL = new AltResult(null);
/** Completes with the null value, unless already completed. */
final boolean completeNull() {
return UNSAFE.compareAndSwapObject(this, RESULT, null,
NIL);
}
/** Returns the encoding of the given non-exceptional value. */
final Object encodeValue(T t) {
return (t == null) ? NIL : t;
}
/** Completes with a non-exceptional result, unless already completed. */
final boolean completeValue(T t) {
return UNSAFE.compareAndSwapObject(this, RESULT, null,
(t == null) ? NIL : t);
}
static AltResult encodeThrowable(Throwable x) {
return new AltResult((x instanceof CompletionException) ? x :
new CompletionException(x));
}
/** Completes with an exceptional result, unless already completed. */
final boolean completeThrowable(Throwable x) {
return UNSAFE.compareAndSwapObject(this, RESULT, null,
encodeThrowable(x));
}
static Object encodeThrowable(Throwable x, Object r) {
if (!(x instanceof CompletionException))
x = new CompletionException(x);
else if (r instanceof AltResult && x == ((AltResult)r).ex)
return r;
return new AltResult(x);
}
final boolean completeThrowable(Throwable x, Object r) {
return UNSAFE.compareAndSwapObject(this, RESULT, null,
encodeThrowable(x, r));
}
Object encodeOutcome(T t, Throwable x) {
return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x);
}
static Object encodeRelay(Object r) {
Throwable x;
return (((r instanceof AltResult) &&
(x = ((AltResult)r).ex) != null &&
!(x instanceof CompletionException)) ?
new AltResult(new CompletionException(x)) : r);
}
final boolean completeRelay(Object r) {
return UNSAFE.compareAndSwapObject(this, RESULT, null,
encodeRelay(r));
}
private static <T> T reportGet(Object r)
throws InterruptedException, ExecutionException {
if (r == null) // by convention below, null means interrupted
throw new InterruptedException();
if (r instanceof AltResult) {
Throwable x, cause;
if ((x = ((AltResult)r).ex) == null)
return null;
if (x instanceof CancellationException)
throw (CancellationException)x;
if ((x instanceof CompletionException) &&
(cause = x.getCause()) != null)
x = cause;
throw new ExecutionException(x);
}
@SuppressWarnings("unchecked") T t = (T) r;
return t;
}
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;
}
/* ------------- Async task preliminaries -------------- */
public static interface AsynchronousCompletionTask {
}
private static final boolean useCommonPool =
(ForkJoinPool.getCommonPoolParallelism() > 1);
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
/** Fallback if ForkJoinPool.commonPool() cannot support parallelism */
static final class ThreadPerTaskExecutor implements Executor {
public void execute(Runnable r) { new Thread(r).start(); }
}
static Executor screenExecutor(Executor e) {
if (!useCommonPool && e == ForkJoinPool.commonPool())
return asyncPool;
if (e == null) throw new NullPointerException();
return e;
}
// Modes for Completion.tryFire. Signedness matters.
static final int SYNC = 0;
static final int ASYNC = 1;
static final int NESTED = -1;
/* ------------- Base Completion classes and operations -------------- */
@SuppressWarnings("serial")
abstract static class Completion extends ForkJoinTask<Void>
implements Runnable, AsynchronousCompletionTask {
volatile Completion next; // Treiber stack link
abstract CompletableFuture<?> tryFire(int mode);
/** Returns true if possibly still triggerable. Used by cleanStack. */
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) {}
}
static void lazySetNext(Completion c, Completion next) {
UNSAFE.putOrderedObject(c, NEXT, next);
}
final void postComplete() {
CompletableFuture<?> f = this; Completion h;
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) {
if (f != this) {
pushStack(h);
continue;
}
h.next = null; // detach
}
f = (d = h.tryFire(NESTED)) == null ? this : d;
}
}
}
/** Traverses stack and unlinks dead Completions. */
final void cleanStack() {
for (Completion p = null, q = stack; q != null;) {
Completion s = q.next;
if (q.isLive()) {
p = q;
q = s;
}
else if (p == null) {
casStack(q, s);
q = stack;
}
else {
p.next = s;
if (p.isLive())
q = s;
else {
p = null; // restart
q = stack;
}
}
}
}
/* ------------- One-input Completions -------------- */
/** A Completion with a source, dependent, and executor. */
@SuppressWarnings("serial")
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;
}
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; }
}
/** Pushes the given completion (if it exists) unless done. */
final void push(UniCompletion<?,?> c) {
if (c != null) {
while (result == null && !tryPushStack(c))
lazySetNext(c, null); // clear on failure
}
}
final CompletableFuture<T> postFire(CompletableFuture<?> a, int mode) {
if (a != null && a.stack != null) {
if (mode < 0 || a.result == null)
a.cleanStack();
else
a.postComplete();
}
if (result != null && stack != null) {
if (mode < 0)
return this;
else
postComplete();
}
return null;
}
@SuppressWarnings("serial")
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);
}
}
final <S> boolean uniApply(CompletableFuture<S> a,
Function<? super S,? extends T> f,
UniApply<S,T> c) {
Object r; Throwable x;
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 {
if (c != null && !c.claim())
return false;
@SuppressWarnings("unchecked") S s = (S) r;
completeValue(f.apply(s));
} catch (Throwable ex) {
completeThrowable(ex);
}
}
return true;
}
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>();
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;
}
@SuppressWarnings("serial")
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);
}
}
final <S> boolean uniAccept(CompletableFuture<S> a,
Consumer<? super S> f, UniAccept<S> c) {
Object r; Throwable x;
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 {
if (c != null && !c.claim())
return false;
@SuppressWarnings("unchecked") S s = (S) r;
f.accept(s);
completeNull();
} catch (Throwable ex) {
completeThrowable(ex);
}
}
return true;
}
private CompletableFuture<Void> uniAcceptStage(Executor e,
Consumer<? super T> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
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;
}
@SuppressWarnings("serial")
static final class UniRun<T> extends UniCompletion<T,Void> {
Runnable fn;
UniRun(Executor executor, CompletableFuture<Void> dep,
CompletableFuture<T> src, Runnable 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.uniRun(a = src, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
final boolean uniRun(CompletableFuture<?> a, Runnable f, UniRun<?> c) {
Object r; Throwable x;
if (a == null || (r = a.result) == null || f == null)
return false;
if (result == null) {
if (r instanceof AltResult && (x = ((AltResult)r).ex) != null)
completeThrowable(x, r);
else
try {
if (c != null && !c.claim())
return false;
f.run();
completeNull();
} catch (Throwable ex) {
completeThrowable(ex);
}
}
return true;
}
private CompletableFuture<Void> uniRunStage(Executor e, Runnable f) {
if (f == null) throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
if (e != null || !d.uniRun(this, f, null)) {
UniRun<T> c = new UniRun<T>(e, d, this, f);
push(c);
c.tryFire(SYNC);
}
return d;
}
@SuppressWarnings("serial")
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);
}
}
final boolean uniWhenComplete(CompletableFuture<T> a,
BiConsumer<? super T,? super Throwable> f,
UniWhenComplete<T> c) {
Object r; T t; Throwable x = null;
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;
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;
}
private CompletableFuture<T> uniWhenCompleteStage(
Executor e, BiConsumer<? super T, ? super Throwable> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<T> d = new CompletableFuture<T>();
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;
}
@SuppressWarnings("serial")
static final class UniHandle<T,V> extends UniCompletion<T,V> {
BiFunction<? super T, Throwable, ? extends V> fn;
UniHandle(Executor executor, CompletableFuture<V> dep,
CompletableFuture<T> src,
BiFunction<? super T, Throwable, ? 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.uniHandle(a = src, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
}
final <S> boolean uniHandle(CompletableFuture<S> a,
BiFunction<? super S, Throwable, ? extends T> f,
UniHandle<S,T> c) {
Object r; S s; Throwable x;
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;
}
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>();
if (e != null || !d.uniHandle(this, f, null)) {
UniHandle<T,V> c = new UniHandle<T,V>(e,
如果觉得我的文章对您有用,请随意打赏。你的支持将鼓励我继续创作!