CompletableFuture使用說明

前言
創建線程的方式只有兩種：繼承Thread或者實現Runnable接口。但是這兩種方法都存在一個缺陷，沒有返回值

Java 1.5 以后，可以通過向線程池提交一個Callable來獲取一個包含返回值的Future對象

Future接口的局限性
當Future的線程進行了一個非常耗時的操作，那我們的主線程也就阻塞了。

當我們在簡單業務上，可以使用Future的另一個重載方法get(long,TimeUnit)來設置超時時間，避免我們的主線程被無窮盡地阻塞。

單純使用Future接口或者FutureTask類并不能很好地完成以下我們所需的業務

將兩個異步計算合并為一個，這兩個異步計算之間相互獨立，同時第二個又依賴于第一個的結果
等待Future集合中的所有任務都完成。
僅等待Future集合種最快結束的任務完成，并返回它的結果。
通過編程方式完成一個Future任務的執行
當Future的完成時間完成時會收到通知，并能使用Future的計算結果進行下一步的的操作，不只是簡單地阻塞等待操作的結果
什么是CompletableFuture
在Java 8中, 新增類: CompletableFuture，結合了Future的優點，提供了非常強大的Future的擴展功能，可以幫助我們簡化異步編程的復雜性，提供了函數式編程的能力，可以通過回調的方式處理計算結果

CompletableFuture被設計在Java中進行異步編程。主線程不用為了任務的完成而阻塞/等待，你可以用主線程去并行執行其他的任務。使用這種并行方式，極大地提升了程序的表現。

CompletableFuture實現了Future接口，因此有異步執行返回結果的能力。
CompletableFuture實現了CompletionStage接口，該接口是Java8新增得一個接口，用于異步執行中的階段處理，其大量用在Lambda表達式計算過程中，目前只有CompletableFuture一個實現類。
public class CompletableFuture implements Future, CompletionStage {
方法命名規則
帶有Async后綴方法都是異步另外線程執行，沒有就是復用之前任務的線程

帶有Apply標識方法都是可以獲取返回值+有返回值的

帶有Accept標識方法都是可以獲取返回值

帶有run標識的方法不可以獲取返回值和無返回值，只是運行

get方法和join方法
join：阻塞獲取結果或拋出非受檢異常。
get: 阻塞獲取結果或拋出受檢測異常，需要顯示進行try...catch處理
不同線程池使用
默認線程池執行

/**
 * 默認線程池
 * 運行結果：
 * main.................start.....
 * main.................end......
 * 當前線程：ForkJoinPool.commonPool-worker-9
 * 運行結果：5
 */
@Test
public void defaultThread() {
    System.out.println("main.................start.....");
    CompletableFuture.runAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getName());
        int i = 10 / 2;
        System.out.println("運行結果：" + i);
    });
    System.out.println("main.................end......");
}

默認使用 ForkJoinPool.commonPool()，commonPool是一個會被很多任務共享的線程池，commonPool 設計時的目標場景是運行非阻塞的 CPU 密集型任務，為最大化利用 CPU，其線程數默認為 CPU 數量- 1

哪些地方使用了commonPool
CompletableFuture
Parallel Streams
為什么要引入commonPool
為了避免任何并行操作都引入一個線程池，最壞情況會導致在單個JVM上創建了太多的池線程，降低效率。
commonPool線程池是怎么創建和使用的
ForkJoinTask一定會運行在一個ForkJoinPool中，如果沒有顯式地交它提交到ForkJoinPool，會使用一個common池（全進程共享）來執行任務。
自定義線程池執行
自定義一個線程池

private ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<Runnable>());

使用定義的線程池

/**
 * 自定義線程池
 * 運行結果：
 * main.................start.....
 * main.................end......
 * 當前線程：pool-1-thread-1
 * 運行結果：5
 */
@Test
public void myThread() {
    System.out.println("main.................start.....");
    CompletableFuture.runAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getName());
        int i = 10 / 2;
        System.out.println("運行結果：" + i);
    },executor);
    System.out.println("main.................end......");
}

開啟一個異步
runAsync-無返回值
使用runAsync開啟一個異步任務線程，該方法無結果返回，適合一些不需要結果的異步任務

/***
 * 無返回值
 *  runAsync
 *  結果：
 * main.................start.....
 * main.................end......
 * 當前線程：33
 * 運行結果：5
 */
@Test
public void runAsync() {
    System.out.println("main.................start.....");
    CompletableFuture.runAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("運行結果：" + i);
    }, executor);
    System.out.println("main.................end......");
}

supplyAsync-有返回值
使用completableFuture.get()方法獲取結果，這時程序會阻塞到這里直到結果返回。

/**
 * 有返回值
 * supplyAsync
 * 結果：
 * main.................start.....
 * 當前線程：33
 * 運行結果：5
 * main.................end.....5
 */
@Test
public void supplyAsync() throws ExecutionException, InterruptedException {
    System.out.println("main.................start.....");
    CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("運行結果：" + i);
        return i;
    }, executor);
    System.out.println("main.................end....." + completableFuture.get());
}

如果要超時就得往下執行，請使用completableFuture.get(long timeout, TimeUnit unit)方法。

線程串行化方法
帶有Async后綴方法都是異步另外線程執行，沒有就是復用之前任務的線程

thenApply-上面任務執行完執行+獲取返回值+有返回值

/**
 * 上面任務執行完執行+可以拿到結果+可以返回值
 * 結果：
 * thenApplyAsync當前線程：33
 * thenApplyAsync運行結果：5
 * thenApplyAsync任務2啟動了。。。。。上步結果：5
 * main.................end.....hello10
 * 
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Test
public void thenApplyAsync() throws ExecutionException, InterruptedException {
 
    CompletableFuture<String> thenApplyAsync = CompletableFuture.supplyAsync(() -> {
        System.out.println("thenApplyAsync當前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("thenApplyAsync運行結果：" + i);
        return i;
    }, executor).thenApplyAsync(result -> {
        System.out.println("thenApplyAsync任務2啟動了。。。。。上步結果：" + result);
        return "hello" + result * 2;
    }, executor);
    System.out.println("main.................end....." + thenApplyAsync.get());
}

thenAccept-上面任務執行完執行+獲取返回值

/**
 * 上面任務執行完執行+可以拿到結果
 * 結果:
 * thenAcceptAsync當前線程：33
 * thenAcceptAsync運行結果：5
 * thenAcceptAsync任務2啟動了。。。。。上步結果：5
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Test
public void thenAcceptAsync() throws ExecutionException, InterruptedException {
    CompletableFuture<Void> thenAcceptAsync = CompletableFuture.supplyAsync(() -> {
        System.out.println("thenAcceptAsync當前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("thenAcceptAsync運行結果：" + i);
        return i;
    }, executor).thenAcceptAsync(result -> {
        System.out.println("thenAcceptAsync任務2啟動了。。。。。上步結果：" + result);
    }, executor);
}

thenRun-上面任務執行完執行

/**
 * 上面任務執行完執行
 * 結果
 * main.................start.....
 * 當前線程：33
 * 運行結果：5
 * 任務2啟動了。。。。。
 */
@Test
public void thenRunAsync() throws ExecutionException, InterruptedException {
    System.out.println("main.................start.....");
    CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("運行結果：" + i);
        return i;
    }, executor).thenRunAsync(() -> {
        System.out.println("任務2啟動了。。。。。");
    }, executor);
}

thenCompose-接收返回值并生成新的任務
當原任務完成后接收返回值，返回一個新的任務

thenApply（）轉換的是泛型中的類型，相當于將CompletableFuture 轉換生成新的CompletableFuture
thenCompose（）用來連接兩個CompletableFuture，是生成一個新的CompletableFuture。

/**
 * 當原任務完成后接收返回值，返回一個新的任務
 * 結果：
 * hello: thenCompose
 */
@Test
public void thenCompose() {
    CompletableFuture cf = CompletableFuture.completedFuture("hello")
            .thenCompose(str -> CompletableFuture.supplyAsync(() -> {
                return str + ": thenCompose";
            },executor));
    System.out.println(cf.join());
}
```java
任務組合
thenCombine-消費兩個結果+返回結果
```java
/**
 * 兩任務組合 都要完成
 * completableFuture.thenCombine()獲取兩個future返回結果，有返回值
 * 結果：
 * 任務1線程：33
 * 任務1運行結果：5
 * 任務2線程：34
 * 任務2運行結果：
 * 任務5啟動。。。結果1：5。。。結果2：hello
 * 任務5結果hello-->5
 */
@Test
public void thenCombine() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("任務1運行結果：" + i);
        return i;
    }, executor);
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務2線程：" + Thread.currentThread().getId());
        System.out.println("任務2運行結果：");
        return "hello";
    }, executor);
    CompletableFuture<String> thenCombineAsync = future1.thenCombineAsync(future2, (result1, result2) -> {
        System.out.println("任務5啟動。。。結果1：" + result1 + "。。。結果2：" + result2);
        return result2 + "-->" + result1;
    }, executor);
    System.out.println("任務5結果" + thenCombineAsync.get());
}

thenAcceptBoth-消費兩個結果+無返回

/**
 * 兩任務組合 都要完成
 * completableFuture.thenAcceptBoth() 獲取兩個future返回結果，無返回值
 * 結果：
 * 任務1線程：33
 * 任務1運行結果：5
 * 任務2線程：34
 * 任務2運行結果：
 * 任務4啟動。。。結果1：5。。。結果2：hello
 */
@Test
public void thenAcceptBothAsync() throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("任務1運行結果：" + i);
        return i;
    }, executor);
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務2線程：" + Thread.currentThread().getId());
        System.out.println("任務2運行結果：");
        return "hello";
    }, executor);
    CompletableFuture<Void> thenAcceptBothAsync = future1.thenAcceptBothAsync(future2, (result1, result2) -> {
        System.out.println("任務4啟動。。。結果1：" + result1 + "。。。結果2：" + result2);
    }, executor);
}

runAfterBoth-兩個任務完成接著運行

/**
 * 兩任務組合 都要完成
 * completableFuture.runAfterBoth() 組合兩個future
 * 結果：
 * 任務1線程：33
 * 任務1運行結果：5
 * 任務2線程：34
 * 任務2運行結果：
 * 任務3啟動。。。
 */
@Test
public void runAfterBothAsync() {
    CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("任務1運行結果：" + i);
        return i;
    }, executor);
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務2線程：" + Thread.currentThread().getId());
        System.out.println("任務2運行結果：");
        return "hello";
    }, executor);
    CompletableFuture<Void> runAfterBothAsync = future1.runAfterBothAsync(future2, () -> {
        System.out.println("任務3啟動。。。");
    }, executor);
}

兩任務完成一個就執行
applyToEither-其中一個執行完執行+獲取返回值+有返回值

/**
 * 兩任務組合，一個任務完成就執行
 * objectCompletableFuture.applyToEither() 其中一個執行完執行+獲取返回值+有返回值
 * 結果：
 * 任務1線程：33
 * 任務2線程：34
 * 任務2運行結果：
 * 任務5開始執行。。。結果：hello
 * 任務5結果：hello world
 * Process finished with exit code 0
 */
@Test
public void applyToEither() throws ExecutionException, InterruptedException {
    CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        try {
            Thread.sleep(3000);
            System.out.println("任務1運行結果：" + i);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return i;
    }, executor);
    CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務2線程：" + Thread.currentThread().getId());
        System.out.println("任務2運行結果：");
        return "hello";
    }, executor);
    CompletableFuture<String> applyToEitherAsync = future1.applyToEitherAsync(future2, result -> {
        System.out.println("任務5開始執行。。。結果：" + result);
        return result.toString() + " world";
    }, executor);
    System.out.println("任務5結果：" + applyToEitherAsync.get());
}

acceptEither-其中一個執行完執行+獲取返回值

/**
 * 兩任務組合，一個任務完成就執行
 * objectCompletableFuture.acceptEither() 其中一個執行完執行+獲取返回值
 * 結果：
 * 任務1線程：33
 * 任務2線程：34
 * 任務2運行結果：
 * 任務4開始執行。。。結果：hello
 */
@Test
public void acceptEither() {
    CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        try {
            Thread.sleep(3000);
            System.out.println("任務1運行結果：" + i);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return i;
    }, executor);
    CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務2線程：" + Thread.currentThread().getId());
        System.out.println("任務2運行結果：");
        return "hello";
    }, executor);
    CompletableFuture<Void> acceptEitherAsync = future1.acceptEitherAsync(future2, result -> {
        System.out.println("任務4開始執行。。。結果：" + result);
    }, executor);
}

runAfterEither-有一任務完成就執行

/**
 * 兩任務組合，一個任務完成就執行
 * objectCompletableFuture.runAfterEither() 其中一個執行完執行
 * 結果：
 * 任務1線程：33
 * 任務2線程：34
 * 任務2運行結果：
 * 任務3開始執行。。。
 */
@Test
public void runAfterEither() {
    CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        try {
            Thread.sleep(3000);
            System.out.println("任務1運行結果：" + i);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return i;
    }, executor);
    CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務2線程：" + Thread.currentThread().getId());
        System.out.println("任務2運行結果：");
        return "hello";
    }, executor);
    CompletableFuture<Void> runAfterEitherAsync = future1.runAfterEitherAsync(future2, () -> {
        System.out.println("任務3開始執行。。。");
    }, executor);
}

多任務組合
allOf-等待全部完成后才執行

/**
 * 多任務組合
 * allOf 等待所有任務完成
 * 結果：
 * 任務1
 * 任務3
 * 任務2
 * allOf任務1-------任務2-------任務3
 *
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Test
public void allOf() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1");
        return "任務1";
    }, executor);
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
            System.out.println("任務2");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "任務2";
    }, executor);
    CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務3");
        return "任務3";
    }, executor);
    CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3);
    //等待所有任務完成
    //allOf.get();
    allOf.join();
    System.out.println("allOf" + future1.get() + "-------" + future2.get() + "-------" + future3.get());
}

anyOf-等待其中之一完成后就執行

/**
 * 多任務組合
 * anyOf 只要一個任務完成
 * 結果：
 * 任務1
 * anyOf--最先完成的是任務1
 * 任務3
 * 等等任務2
 * 任務2
 *
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Test
public void anyOf() throws ExecutionException, InterruptedException {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務1");
        return "任務1";
    }, executor);
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            Thread.sleep(2000);
            System.out.println("任務2");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "任務2";
    }, executor);
    CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
        System.out.println("任務3");
        return "任務3";
    }, executor);
    CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2, future3);
    System.out.println("anyOf--最先完成的是" + anyOf.get());
    //等待future2打印
    System.out.println("等等任務2");
    Thread.sleep(3000);
}

感知異常
handle-捕獲結果或異常并返回新結果
入參為結果或者異常，返回新結果

/**
 * 入參為結果或者異常，返回新結果
 * 結果：
 * main.................start.....
 * 當前線程：33
 * main.................end.....報錯返回
 */
@Test
public void handle() throws ExecutionException, InterruptedException {
    System.out.println("main.................start.....");
    final CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getId());
        int i = 10 / 0;
        System.out.println("運行結果：" + i);
        return i;
    }, executor).handleAsync((in, throwable) -> {
        if (throwable != null) {
            return "報錯返回";
        }
        return "正確了";
    });
    System.out.println("main.................end....." + completableFuture.get());
}

whenComplete-感知結果或異常并返回相應信息
whenComplete雖然得到異常信息，但是不能修改返回信息

/**
 * 有返回值并且有后續操作 whenComplete
 * 結果：
 * main.................start.....
 * 當前線程：33
 * 異步完成。。。。結果是：null...異常是：java.util.concurrent.CompletionException: java.lang.ArithmeticException: 除以零
 * 報錯了2
 *
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Test
public void whenComplete() {
    System.out.println("main.................start.....");
    final CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getId());
        int i = 10 / 0;
        System.out.println("運行結果：" + i);
        return i;
    }, executor).whenComplete((result, throwable) -> {
        //whenComplete雖然得到異常信息，但是不能修改返回信息
        System.out.println("異步完成。。。。結果是：" + result + "...異常是：" + throwable);
    });
    try {
        System.out.println("main.................end..T..." + completableFuture.get());
    } catch (InterruptedException e) {
        System.out.println("報錯了1");
    } catch (ExecutionException e) {
        System.out.println("報錯了2");
    }
}

exceptionally-捕獲異常并返回指定值

/**
 * 方法完成后的感知
 * 感知錯誤并返回指定值 exceptionally
 * 結果：
 * main.................start.....
 * 當前線程：33
 * 執行了exceptionally
 * main.................end.....0
 * @throws ExecutionException
 * @throws InterruptedException
 */
@Test
public void exceptionally() throws ExecutionException, InterruptedException {
    System.out.println("main.................start.....");
    CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("當前線程：" + Thread.currentThread().getId());
        int i = 10 / 0;
        System.out.println("運行結果：" + i);
        return i;
    }, executor).exceptionally(throwable -> {
        //R apply(T t);
        //exceptionally可以感知錯誤并返回指定值
        System.out.println("執行了exceptionally");
        return 0;
    });
    System.out.println("main.................end....." + completableFuture.get());
}

posted @ 2024-03-03 14:07 太陽終將升起閱讀(106) 評論(0) 收藏舉報

刷新頁面返回頂部