快捷導(dǎo)航

詳解Java CompletableFuture使用方法以及與FutureTask的區(qū)別

更新時間：2021年10月26日 15:34:07 作者：一月三千五

CompletableFuture實現(xiàn)了CompletionStage接口和Future接口，前者是對后者的一個擴(kuò)展，增加了異步回調(diào)、流式處理、多個Future組合處理的能力，使Java在處理多任務(wù)的協(xié)同工作時更加順暢便利

futureTask 創(chuàng)建異步任務(wù)

        FutureTask<String> stringFutureTask = new FutureTask<>(() -> {
            return "aa";
        });
        executorService.execute(stringFutureTask);
        System.out.println(stringFutureTask.get());
 
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            return "aa";
        }, executorService); // 不用手動提交了
        System.out.println(future1.get());

還有很多異步回調(diào), 組合處理

創(chuàng)建任務(wù)

1. .supplyAsync

創(chuàng)建一個帶返回值的任務(wù)

2. .runAsync

創(chuàng)建一個不帶返回值的任務(wù)

        ExecutorService executorService = Executors.newFixedThreadPool(1);
 
        // 帶返回值
        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("future　" + new Date());
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "aaa";
        }, executorService); // 推薦使用

以上兩個方法都有兩個構(gòu)造方法, 默認(rèn)不指定自定義線程池, 他會指定默認(rèn)的提交任務(wù)的方法

    // 查看cpu的核數(shù)是否大于1核
    private static final boolean useCommonPool =
        (ForkJoinPool.getCommonPoolParallelism() > 1);
 
    // 如果大于1核 則調(diào)用execute方法, 每次創(chuàng)建一個線程
    private static final Executor asyncPool = useCommonPool ?
        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
 
    static final class ThreadPerTaskExecutor implements Executor {
        public void execute(Runnable r) { new Thread(r).start(); }
    }

所以推薦自定義線程池的方式

異步回調(diào)

指的是異步任務(wù)結(jié)束后調(diào)用的任務(wù)

1. .thenApply

帶返回值的異步調(diào)用函數(shù), 有入?yún)? 有出參

2. .thenAccept

不帶返回值的異步回調(diào)函數(shù), 有入?yún)?/p>

 
        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("future　" + new Date());
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "aaa";
        }, executorService);
 
        // future執(zhí)行完之后執(zhí)行的異步任務(wù)
        CompletableFuture<String> thenApply = future.thenApply((result) -> {
            System.out.println("future2　" +new Date());
            System.out.println(result);
            return "bbb" + result;
        });

3. .exceptionally

異步任務(wù)出現(xiàn)異常調(diào)用的回調(diào)方法

        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("future　" + new Date());
                Thread.sleep(2000);
                int a = 1 / 0;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "aaa";
        }, executorService);
 
        CompletableFuture<String> exceptionally = future.exceptionally((result) -> {
            System.out.println("future3 " + result);
            return "bbb" + result;
        });
        
        // 出現(xiàn)異常則返回異常, 沒異常則返回future的返回值
        System.out.println(exceptionally.get());

去掉異常

4. .whenComplete

當(dāng)主任務(wù)出現(xiàn)異常時, 會終止任務(wù),get的時候會拋出主任務(wù)的異常, 入?yún)⒅禐閚ull, 否則正常運行

        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("future　" + new Date());
                Thread.sleep(2000);
                int a = 1/0;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "aaa";
        }, executorService);
 
        CompletableFuture<String> exceptionally = future.whenComplete((result, error) -> {
            System.out.println("future3 " + result);
            System.out.println("future3 " + error);
        });
        System.out.println(exceptionally.get());