Java中CompletableFuture四种调用模式的实现

var cf = CompletableFuture.completed(1);
cf.thenApply(x -> x + 1)
  .thenRun(x -> System.out.println(x))
  .join();

var cf = new CompletableFuture<Integer>();
cf.thenApply(x -> x + 1)
  .thenRun(x -> System.out.println(x));
new Thread(() -> cf.complete(1)).start();
Uninterruptible.sleep(1, TimeUnit.SECONDS);

var cf = CompletableFuture.completed(1);
cf.thenApplyAsync(x -> x + 1)
  .thenRunAsync(x -> System.out.println(x))
  .join();
Uninterruptible.sleep(1, TimeUnit.SECONDS);

public static <T, F extends CompletionStage<?>> F relayAsync0(
        CompletionStage<? extends T> cfThis,
        Function<CompletableFuture<T>, F> relayComputations, Executor executor)

cf -> cf.thenApply(...)
cf -> cf.thenCompose(...)
cf -> cf.thenRun(...)

public class RelayAsyncDescriptionAndExample {
    static void executeComputationsOfNewStage(CompletableFuture<String> cf) {

        // ================================================================================
        // Default execution
        // ================================================================================

        cf.thenApply(s -> {
            // a simulating long-running computation...
            sleep(1000);
            // if input cf is COMPLETED when computations execute,
            // executes the long time computation SYNCHRONOUSLY (aka. in the caller thread);
            // this SYNCHRONIZED execution leads to BLOCKing sequential codes of caller... ⚠️

            return s + s;
        });

        // ================================================================================
        // Asynchronous execution of CompletableFuture(default executor or custom executor)
        // ================================================================================

        cf.thenApplyAsync(s -> {
            // a simulating long-running computation...
            sleep(1000);
            // always executes via an executor(guarantees not to block sequential code of caller).
            // if input cf is INCOMPLETE when computations execute,
            // the execution via an executor leads to ONE MORE thread switching. ⚠️

            return s + s;
        });

        // ================================================================================
        // How about the fourth way to arrange execution of a new stage's computations?
        // ================================================================================
        //
        // - if input cf is COMPLETED when computations execute, use "asynchronous execution" (via supplied Executor),
        //   won't block sequential code of caller ✅
        // - otherwise, use "default execution", save one thread switching ✅
        //
        // Let's call this way as "relay async".

        LLCF.relayAsync0(cf, f -> f.thenApply(s -> {
            // a simulating long-running computation...
            sleep(1000);
            // if input cf is COMPLETED, executes via supplied executor
            // if input cf is INCOMPLETE, use "default execution"

            return s + s;
        }), ForkJoinPool.commonPool());
    }
}

public static <T, F extends CompletionStage<?>> F relayAsync0(
        CompletionStage<? extends T> cfThis,
        Function<CompletableFuture<T>, F> relayComputations, Executor executor) {
    final CompletableFuture<T> promise = new CompletableFuture<>();
    final F ret = relayComputations.apply(promise);

    final Thread callerThread = currentThread();
    final boolean[] returnedFromPeek0 = {false};

    LLCF.peek0(cfThis, (v, ex) -> {
        if (currentThread().equals(callerThread) && !returnedFromPeek0[0]) {
            // If the action is running in the caller thread(single same thread) and `peek0` invocation does not
            // return to caller(flag returnedFromPeek0 is false), the action is being executed synchronously.
            // To prevent blocking the caller's sequential code, use the supplied executor to complete the promise.
            executor.execute(() -> completeCf0(promise, v, ex));
        } else {
            // Otherwise, complete the promise directly, avoiding one thread switching.
            completeCf0(promise, v, ex);
        }
    }, "relayAsync0");

    returnedFromPeek0[0] = true;
    return ret;
}