1. 介紹

CompletableFuture 是 Java 8 引入的一個用于異步編程的類，位于 java.util.concurrent 包中。它是對 Future 的增強，提供了更強大的功能來支持異步任務的編排、組合和處理。

2. 方法

不使用自定義線程池，會使用ForkJoinPool中的共用線程池CommonPool（CommonPool的大小是CPU核數-1，如果是IO密集的應用，線程數可能成為瓶頸）

2.1 異步執行，不需要結果

使用默認線程池

CompletableFuture<Void> future1 = CompletableFuture.runAsync(new Runnable() {
    @Override
    public void run() {
        // 邏輯
    }
});

使用自定義線程池

CompletableFuture<Void> future2 = CompletableFuture.runAsync(new Runnable() {
    @Override
    public void run() {
        // 邏輯
    }
}, executor);

2.2 異步執行，需要結果

使用默認線程池

CompletableFuture<String> future3 = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public String get() {
        // 邏輯寫在這里
        return "result"; // 結果類型 可以修改T public interface Supplier<T>{}
    }
});

使用自定義線程池

CompletableFuture<String> future4 = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public String get() {
        // 邏輯寫在這里
        return "result"; // 結果類型 可以修改T public interface Supplier<T>{}
    }
}, executor);

2.3 thenRun() 同步執行 thenRunAsync() 異步執行

CompletableFuture<Void> future5 = future1.thenRun(new Runnable() {
    @Override
    public void run() {
        // 邏輯
    }
});

2.4 thenApply() 同步執行 thenApplyAsync() 異步執行

CompletableFuture<String> future6 = future3.thenApply(new Function<String, String>() {
    @Override
    public String apply(String s) {
        return "result";
    }
});

2.5 thenAccept() 同步執行 thenAcceptAsync() 異步執行， 與 thenRun 不同， thenAccept 存在入參

CompletableFuture<Void> future7 = future3.thenAccept(new Consumer<String>() {
    @Override
    public void accept(String s) {
        // 邏輯
    }
});

2.6 thenAcceptBoth、thenAcceptBothAsync() 兩個 future 全部執行完成，才會執行

CompletableFuture<Void> future8 = future3.thenAcceptBoth(future4, new BiConsumer<String, String>() {
    // 獲取倆個 future 的結果
    @Override
    public void accept(String s8, String s4) {

    }
});

2.7 applyToEither()、applyToEitherAsync() 兩個 future 任意一個執行完成，才會執行

CompletableFuture<String> future9 = future3.applyToEither(future4, new Function<String, String>() {
    @Override
    public String apply(String s) {
        return s;
    }
});

2.8 acceptEither 消費 future6 和 future4 的任意一個結果，并消費

CompletableFuture<Void> future10 = future6.acceptEither(future4, new Consumer<String>() {
    @Override
    public void accept(String s) {
        System.out.println(s);
    }
});

2.9 thenCompose()、thenComposeAsync() 新建一個 CompletableFuture，并返回

CompletableFuture<String> future11 = future3.thenCompose(new Function<String, CompletableFuture<String>>() {
    @Override
    public CompletableFuture<String> apply(String s) {
        return CompletableFuture.supplyAsync(() -> s + " World");
    }
});

2.10 thenCombine()、thenCombineAsync()

CompletableFuture<String> future12 = future3.thenCombine(future4, new BiFunction<String, String, String>() {
    @Override
    public String apply(String s, String s2) {
        return s + s2;
    }
});

2.11 whenComplete()、whenCompleteAsync() 不會吞掉異常

CompletableFuture<String> future13 = future3.whenComplete(new BiConsumer<String, Throwable>() {
    @Override
    public void accept(String s, Throwable throwable) {
        System.out.println(s);
    }
});

2.12 handle()、handleAsync() 會吞掉異常

CompletableFuture<String> future14 = future3.handle(new BiFunction<String, Throwable, String>() {
    @Override
    public String apply(String s, Throwable throwable) {
        return s;
    }
});

2.13 exceptionally()

CompletableFuture<String> future15 = future3.exceptionally(new Function<Throwable, String>() {
    @Override
    public String apply(Throwable throwable) {
        return "error";
    }
});

2.14 allOf()、anyOf()

CompletableFuture<Void> future16 = CompletableFuture.allOf(future3, future4); // allOf() 是同步所有任務結果，而不需要進行處理他們的結果
CompletableFuture<Object> future17 = CompletableFuture.anyOf(future3, future4); // anyOf() 是等待其中一個future處理完成，并返回結果

3 使用場景

3.1 例子1

如圖所示，各個流程之間有相關依賴，比如執行CF4需要CF1、CF2的結果：

        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "cf1 result"); // 異步執行有返回值
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> "cf2 result");

        CompletableFuture<String> cf3 = cf1.thenApply(s -> "cf3 result");
        CompletableFuture<String> cf4 = cf1.thenCombineAsync(cf2, (s1, s2) -> {
            System.out.println("cf1 result: " + s1);
            System.out.println("cf2 result: " + s2);
            return "cf4 result";
        });
        CompletableFuture<String> cf5 = cf2.thenApply(s -> "cf5 result");

        CompletableFuture<Void> cf6 = CompletableFuture.allOf(cf3, cf4, cf5);
        String cf6Result = cf6.thenApply(v -> {
            cf3.join();
            cf4.join();
            cf5.join();
            return "cf6 result";
        }).get();
        System.out.println(cf6Result);

3.2 例子2

ExecutorService executor = Executors.newFixedThreadPool(5);
//1、使用runAsync或supplyAsync發起異步調用
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
  return "result1";
}, executor);

//2、CompletableFuture.completedFuture()直接創建一個已完成狀態的CompletableFuture
//使用場景為緩存，命中緩存直接返回
CompletableFuture<String> cf2 = CompletableFuture.completedFuture("result2");

//3、先初始化一個未完成的CompletableFuture，然后通過complete()、completeExceptionally()，完成該CompletableFuture
//使用場景為回調機制、超時處理、任務取消等場景。
CompletableFuture<String> cf2 = new CompletableFuture<>();
cf.complete("success");

4. 問題

合理使用線程池，避免死鎖

        ExecutorService threadPool1 = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100));
        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            return CompletableFuture.supplyAsync(() -> {
                System.out.println("child");
                return "child";
            }, threadPool1).join();// 子任務
        }, threadPool1);
        future.join();
        System.out.println("main");

上面的例子中，當10個線程同事請求時，父類future將10個線程獲取，子類不能獲取線程導致死鎖出現。

5. 總結

附錄

美團工具類 : CompletableFuture原理與實踐-外賣商家端API的異步化

import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.BinaryOperator;
import java.util.function.Predicate;
import java.util.stream.Collectors;

/**
 * CompletableFuture封裝工具類
 */
@Slf4j
public class FutureUtils {

    /**
     * 設置CF狀態為失敗
     */
    public static <T> CompletableFuture<T> failed(Throwable ex) {
        CompletableFuture<T> completableFuture = new CompletableFuture<>();
        completableFuture.completeExceptionally(ex);
        return completableFuture;
    }

    /**
     * 設置CF狀態為成功
     */
    public static <T> CompletableFuture<T> success(T result) {
        CompletableFuture<T> completableFuture = new CompletableFuture<>();
        completableFuture.complete(result);
        return completableFuture;
    }

    /**
     * 將List<CompletableFuture<T>> 轉為 CompletableFuture<List<T>>
     */
    public static <T> CompletableFuture<List<T>> sequence(Collection<CompletableFuture<T>> completableFutures) {
        return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList())
                );
    }

    /**
     * 將List<CompletableFuture<List<T>>> 轉為 CompletableFuture<List<T>>
     * 多用于分頁查詢的場景
     */
    public static <T> CompletableFuture<List<T>> sequenceList(Collection<CompletableFuture<List<T>>> completableFutures) {
        return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream()
                        .flatMap(listFuture -> listFuture.join().stream())
                        .collect(Collectors.toList())
                );
    }

    /*
     * 將List<CompletableFuture<Map<K, V>>> 轉為 CompletableFuture<Map<K, V>>
     * @Param mergeFunction 自定義key沖突時的merge策略
     */
    public static <K, V> CompletableFuture<Map<K, V>> sequenceMap(
            Collection<CompletableFuture<Map<K, V>>> completableFutures, BinaryOperator<V> mergeFunction) {
        return CompletableFuture
                .allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream().map(CompletableFuture::join)
                        .flatMap(map -> map.entrySet().stream())
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, mergeFunction)));
    }

    /**
     * 將List<CompletableFuture<T>> 轉為 CompletableFuture<List<T>>，并過濾調null值
     */
    public static <T> CompletableFuture<List<T>> sequenceNonNull(Collection<CompletableFuture<T>> completableFutures) {
        return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream()
                        .map(CompletableFuture::join)
                        .filter(e -> e != null)
                        .collect(Collectors.toList())
                );
    }

    /**
     * 將List<CompletableFuture<List<T>>> 轉為 CompletableFuture<List<T>>，并過濾調null值
     * 多用于分頁查詢的場景
     */
    public static <T> CompletableFuture<List<T>> sequenceListNonNull(Collection<CompletableFuture<List<T>>> completableFutures) {
        return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream()
                        .flatMap(listFuture -> listFuture.join().stream().filter(e -> e != null))
                        .collect(Collectors.toList())
                );
    }

    /**
     * 將List<CompletableFuture<Map<K, V>>> 轉為 CompletableFuture<Map<K, V>>
     *
     * @Param filterFunction 自定義過濾策略
     */
    public static <T> CompletableFuture<List<T>> sequence(Collection<CompletableFuture<T>> completableFutures,
                                                          Predicate<? super T> filterFunction) {
        return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream()
                        .map(CompletableFuture::join)
                        .filter(filterFunction)
                        .collect(Collectors.toList())
                );
    }

    /**
     * 將List<CompletableFuture<List<T>>> 轉為 CompletableFuture<List<T>>
     *
     * @Param filterFunction 自定義過濾策略
     */
    public static <T> CompletableFuture<List<T>> sequenceList(Collection<CompletableFuture<List<T>>> completableFutures,
                                                              Predicate<? super T> filterFunction) {
        return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream()
                        .flatMap(listFuture -> listFuture.join().stream().filter(filterFunction))
                        .collect(Collectors.toList())
                );
    }

    /**
     * 將CompletableFuture<Map<K,V>>的list轉為 CompletableFuture<Map<K,V>>。 多個map合并為一個map。 如果key沖突，采用新的value覆蓋。
     */
    public static <K, V> CompletableFuture<Map<K, V>> sequenceMap(
            Collection<CompletableFuture<Map<K, V>>> completableFutures) {
        return CompletableFuture
                .allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
                .thenApply(v -> completableFutures.stream().map(CompletableFuture::join)
                        .flatMap(map -> map.entrySet().stream())
                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (a, b) -> b)));
    }
}