public static void main(String[] args) {
    // 任務(wù) A，耗時(shí) 2 秒
    int resultA = compute(1);
    // 任務(wù) B，耗時(shí) 2 秒
    int resultB = compute(2);
    // 后續(xù)業(yè)務(wù)邏輯處理
    System.out.println(resultA + resultB);
}

public static void main(String[] args) {
    // 僅簡(jiǎn)單舉例，在生產(chǎn)代碼中可別這么寫(xiě)！
    // 統(tǒng)計(jì)耗時(shí)的函數(shù)
    time(() -> {
        CompletableFuture<Integer> result = Stream.of(1, 2)
                                                  // 創(chuàng)建異步任務(wù)
                                                  .map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor))
                                                  // 聚合
                                                  .reduce(CompletableFuture.completedFuture(0), (x, y) -> x.thenCombineAsync(y, Integer::sum, executor));
        // 等待結(jié)果
        try {
            System.out.println("結(jié)果：" + result.get());
        } catch (ExecutionException | InterruptedException e) {
            System.err.println("任務(wù)執(zhí)行異常");
        }
    });
}
輸出：
[async-1]: 任務(wù)執(zhí)行開(kāi)始：1
[async-2]: 任務(wù)執(zhí)行開(kāi)始：2
[async-1]: 任務(wù)執(zhí)行完成：1
[async-2]: 任務(wù)執(zhí)行完成：2
結(jié)果：3
耗時(shí)：2 秒

public static void main(String[] args) {
    // 僅簡(jiǎn)單舉例，在生產(chǎn)代碼中可別這么寫(xiě)！
    // 統(tǒng)計(jì)耗時(shí)的函數(shù)
    time(() -> {
        List<CompletableFuture<Integer>> result = Stream.of(1, 2)
                                                        // 創(chuàng)建異步任務(wù)，compute(x) 超時(shí)拋出異常
                                                        .map(x -> CompletableFuture.supplyAsync(() -> compute(x), executor))
                                                        .toList();
        // 等待結(jié)果
        int res = 0;
        for (CompletableFuture<Integer> future : result) {
            try {
                res += future.get(2, SECONDS);
            } catch (ExecutionException | InterruptedException | TimeoutException e) {
                System.err.println("任務(wù)執(zhí)行異常或超時(shí)");
            }
        }
        System.out.println("結(jié)果：" + res);
    });
}
輸出：
[async-2]: 任務(wù)執(zhí)行開(kāi)始：2
[async-1]: 任務(wù)執(zhí)行開(kāi)始：1
[async-1]: 任務(wù)執(zhí)行完成：1
任務(wù)執(zhí)行異?；虺瑫r(shí)
結(jié)果：1
耗時(shí)：2 秒

public V get(long timeout, TimeUnit unit) throws InterruptedException {
    // 配置的超時(shí)時(shí)間
    timeout = unit.toMillis(timeout);
    // 剩余等待時(shí)間
    long remaintime = timeout - (this.sentTime - this.genTime);
    if (remaintime <= 0L) {
        if (this.isDone()) {
            // 反序列化獲取結(jié)果
            return this.getNow();
        }
    } else if (this.await(remaintime, TimeUnit.MILLISECONDS)) {
        // 等待時(shí)間內(nèi)任務(wù)完成，反序列化獲取結(jié)果
        return this.getNow();
    }
    this.setDoneTime();
    // 超時(shí)拋出異常
    throw this.clientTimeoutException(false);
}

public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) {
    if (unit == null)
        throw new NullPointerException();
    if (result == null)
        whenComplete(new Canceller(Delayer.delay(new Timeout(this), timeout, unit)));
    return this;
}

static final class Timeout implements Runnable {
    final CompletableFuture<?> f;
    Timeout(CompletableFuture<?> f) { this.f = f; }
    public void run() {
        if (f != null && !f.isDone())
            // 拋出超時(shí)異常
            f.completeExceptionally(new TimeoutException());
    }
}

static final class Delayer {
    static ScheduledFuture<?> delay(Runnable command, long delay,
                                    TimeUnit unit) {
        // 到時(shí)間觸發(fā) command 任務(wù)
        return delayer.schedule(command, delay, unit);
    }
    static final class DaemonThreadFactory implements ThreadFactory {
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setDaemon(true);
            t.setName("CompletableFutureDelayScheduler");
            return t;
        }
    }
    static final ScheduledThreadPoolExecutor delayer;
    static {
        (delayer = new ScheduledThreadPoolExecutor(
            1, new DaemonThreadFactory())).
            setRemoveOnCancelPolicy(true);
    }
}

static final class Canceller implements BiConsumer<Object, Throwable> {
    final Future<?> f;
    Canceller(Future<?> f) { this.f = f; }
    public void accept(Object ignore, Throwable ex) {
        if (ex == null && f != null && !f.isDone())
        // 3 未觸發(fā)拋異常任務(wù)則取消
            f.cancel(false);
    }
}

CompletableFutureExpandUtils.orTimeout(異步任務(wù), 超時(shí)時(shí)間, 時(shí)間單位);

package com.jd.jr.market.reduction.util;
import com.jdpay.market.common.exception.UncheckedException;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
/**
 * CompletableFuture 擴(kuò)展工具
 *
 * @author zhangtianci7
 */
public class CompletableFutureExpandUtils {
    /**
     * 如果在給定超時(shí)之前未完成，則異常完成此 CompletableFuture 并拋出 {@link TimeoutException} 。
     *
     * @param timeout 在出現(xiàn) TimeoutException 異常完成之前等待多長(zhǎng)時(shí)間，以 {@code unit} 為單位
     * @param unit    一個(gè) {@link TimeUnit}，結(jié)合 {@code timeout} 參數(shù)，表示給定粒度單位的持續(xù)時(shí)間
     * @return 入?yún)⒌?CompletableFuture
     */
    public static <T> CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit unit) {
        if (null == unit) {
            throw new UncheckedException("時(shí)間的給定粒度不能為空");
        }
        if (null == future) {
            throw new UncheckedException("異步任務(wù)不能為空");
        }
        if (future.isDone()) {
            return future;
        }
        return future.whenComplete(new Canceller(Delayer.delay(new Timeout(future), timeout, unit)));
    }
    /**
     * 超時(shí)時(shí)異常完成的操作
     */
    static final class Timeout implements Runnable {
        final CompletableFuture<?> future;
        Timeout(CompletableFuture<?> future) {
            this.future = future;
        }
        public void run() {
            if (null != future && !future.isDone()) {
                future.completeExceptionally(new TimeoutException());
            }
        }
    }
    /**
     * 取消不需要的超時(shí)的操作
     */
    static final class Canceller implements BiConsumer<Object, Throwable> {
        final Future<?> future;
        Canceller(Future<?> future) {
            this.future = future;
        }
        public void accept(Object ignore, Throwable ex) {
            if (null == ex && null != future && !future.isDone()) {
                future.cancel(false);
            }
        }
    }
    /**
     * 單例延遲調(diào)度器，僅用于啟動(dòng)和取消任務(wù)，一個(gè)線程就足夠
     */
    static final class Delayer {
        static ScheduledFuture<?> delay(Runnable command, long delay, TimeUnit unit) {
            return delayer.schedule(command, delay, unit);
        }
        static final class DaemonThreadFactory implements ThreadFactory {
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setDaemon(true);
                t.setName("CompletableFutureExpandUtilsDelayScheduler");
                return t;
            }
        }
        static final ScheduledThreadPoolExecutor delayer;
        static {
            delayer = new ScheduledThreadPoolExecutor(1, new DaemonThreadFactory());
            delayer.setRemoveOnCancelPolicy(true);
        }
    }
}