Java8 CompletableFuture使用及說(shuō)明

更新時(shí)間：2026年04月02日 17:07:26 作者：Jerry的技術(shù)博客

文章介紹了Future和CompletableFuture接口及其方法,解釋了它們的異同、使用場(chǎng)景與應(yīng)用場(chǎng)景,示例了Future和CompletableFuture的使用方法和場(chǎng)景,強(qiáng)調(diào)了CompletableFuture的函數(shù)式編程特性

一、Future接口

1.1 Runnable與Callable

Runnable接口源自JDK1.1，它只有一個(gè)run()方法，該方法沒有返回結(jié)果：

public interface Runnable {
    public abstract void run();
}

Callable接口是JDK1.5中添加，只有一個(gè)call()方法，該方法支持結(jié)果返回且可以拋出異常：

public interface Callable<V> {
    V call() throws Exception;
}

無(wú)論是Runnalble還是Callable任務(wù)實(shí)例，直接調(diào)用時(shí)不會(huì)新開啟子線程，是在主線程中運(yùn)行。

如果要在子線程中運(yùn)行這些任務(wù)，需要將任務(wù)提交到Thread實(shí)例或線程池執(zhí)行。

值得注意的是通過(guò)Thread實(shí)例執(zhí)行Callable任務(wù)，Thread沒有接受Callable入?yún)⒌臉?gòu)造函數(shù)，因此，不能直接構(gòu)造Callable任務(wù)的Thread實(shí)例。

可以通過(guò)構(gòu)建FutureTask任務(wù)將Callable任務(wù)提供到Thread實(shí)例執(zhí)行，因?yàn)镕utureTask實(shí)現(xiàn)了Runnable接口，且其構(gòu)造函數(shù)入?yún)⒅С諧allable類型。

	@Test
    public void testRunable() throws Exception {
        Runnable r = new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(1000L);
                    System.out.println(Thread.currentThread().getName());
                } catch (InterruptedException e) {
                }
            }
        };

        Callable<Integer> c = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                try {
                    Thread.sleep(1000L);
                    System.out.println(Thread.currentThread().getName());
                } catch (InterruptedException e) {
                }
                return ThreadLocalRandom.current().nextInt(100);
            }
        };

        r.run();
        System.out.println(c.call());

        Thread t = new Thread(r);
        t.start();
        t.join(); // 不阻塞時(shí)，如果主線程比子線程先完成，則子線程中任務(wù)不會(huì)完整執(zhí)行

        FutureTask<Integer> f = new FutureTask<>(c);
        Thread t2 = new Thread(f);
        t2.start();
        t2.join(); // 不阻塞時(shí)，如果主線程比子線程先完成，則子線程中任務(wù)不會(huì)完整執(zhí)行
        System.out.println(f.get());

        ExecutorService executorService = Executors.newFixedThreadPool(1);
        executorService.submit(r);
        Future<Integer> future = executorService.submit(c);
        System.out.println(future.get()); // 不阻塞時(shí)，如果主線程比子線程先完成，則子線程中任務(wù)不會(huì)完整執(zhí)行
    }
---
main
main
62
Thread-0
Thread-1
6
pool-1-thread-1
pool-1-thread-1
36

注意：

通過(guò)junit對(duì)以上代碼進(jìn)行測(cè)試時(shí)，如果主線程比子線程先執(zhí)行完成，則主線程執(zhí)行完成后會(huì)程序會(huì)退出，意味著沒有完成的子線程任務(wù)不會(huì)再繼續(xù)執(zhí)行。
因此，如果需要子線程完整執(zhí)行，則需要阻塞主線程。

1.2 Future接口

Future接口也是在JDK1.5中添加的，用于描述一個(gè)異步Callable任務(wù)的計(jì)算結(jié)果，如下所示。

Future接口一共有5個(gè)方法：

通過(guò)cancel()方法停止任務(wù)；
通過(guò)isCanceld()方法判斷任務(wù)是否被停止；
通過(guò)isDone()方法檢查任務(wù)是否完成；
提供了兩個(gè)重載版本的get方法，用于阻塞調(diào)用線程，直到任務(wù)完成或超時(shí)。

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Future只能通過(guò)阻塞或輪訓(xùn)的方式獲取任務(wù)結(jié)果，阻塞違背異步變成的初衷，輪訓(xùn)的方式又耗費(fèi)了CPU資源，通過(guò)Future獲取異步任務(wù)結(jié)果的示里如下：

    @Test
    public void testFuture() throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        List<Future<Integer>> futures = new ArrayList<>();
        List<Integer> results = new ArrayList<>();
        for(int i=0; i <= 3; i++) {
            final int j = i;
            Future<Integer> future = executorService.submit(() -> {
                try {
                    Thread.sleep(j * 100L);
                } catch (InterruptedException e) {
                }
                return ThreadLocalRandom.current().nextInt(100);
            });
            futures.add(future);
        }
        for(Future<Integer> future : futures) {
            while(true) {
                if(future.isDone() && !future.isCancelled()) {
                    results.add(future.get());
                    break;
                }
            }
        }
        System.out.println(results);
    }

二、CompletableFuture

JDK8中添加了ComletableFuture類，它實(shí)現(xiàn)了Future接口和CompletionStage接口，其中CompletionStage可以看做是一個(gè)異步任務(wù)執(zhí)行過(guò)程的抽象，如下。

它簡(jiǎn)化了異步變成的復(fù)雜性，并提供了函數(shù)式變成的能力。

ComletableFuture提供了同步和異步的執(zhí)行方式，其異步方式又支持外部聲明的線程池，如果不提供外部線程池，則異步時(shí)默認(rèn)使用ForkJoinPool.commonPool()。

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
...
}

CompletableFuture實(shí)現(xiàn)了Future接口，因此，也可以通過(guò)以前的方式阻塞或輪訓(xùn)獲取結(jié)果：

// 同F(xiàn)uture
public T get() throws InterruptedException, ExecutionException {
...
}

// 同F(xiàn)uture
public T get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException {
...
}

// 返回執(zhí)行結(jié)果或拋出一個(gè)unchecked異常
public T join() {
...
}

// 如果執(zhí)行結(jié)束則返回執(zhí)行結(jié)果，否則返回valueIfAbsent
public T getNow(T valueIfAbsent) {
...
}

CompetableFuture.completedFuture是一個(gè)靜態(tài)輔助類，用于返回一個(gè)計(jì)算好的CompletableFuture，示例：

    @Test
    public void testComletedFuture() throws InterruptedException, ExecutionException {
        CompletableFuture<String> future = CompletableFuture.completedFuture("result");
        System.out.println(future.get());
    }
---
result

通過(guò)以下四個(gè)靜態(tài)方法為異步執(zhí)行代碼創(chuàng)建CompletableFuture對(duì)象：

// 接受Runnable任務(wù)，返回結(jié)果為空
public static CompletableFuture<Void> runAsync(Runnable runnable) {
...
}

// 接受Runnable任務(wù)，返回結(jié)果為空，指定線程池
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) {
...
}

// 接受Supplier對(duì)象，可指定返回類型
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
...
}

// 接受Supplier對(duì)象，可指定返回類型，指定線程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) {
...
}

CompletableFuture任務(wù)執(zhí)行示例如下，可指定執(zhí)行線程池，如不指定則使用ForkJoinPool.commonPool()。

    @Test
    public void testCreateSyncTask() throws InterruptedException, ExecutionException {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "supplyAsync result";
        });
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        CompletableFuture<Void> f2 = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
        }, executorService);
        System.out.println(f1.get());
    }
---
ForkJoinPool.commonPool-worker-1
supplyAsync result
pool-1-thread-1

Future和CompletableFuture提供的get()方法是阻塞的，為了獲取任務(wù)結(jié)果同時(shí)不阻塞當(dāng)前線程，可使用CompletionStage提供的方法實(shí)現(xiàn)任務(wù)異步處理，有以下4中處理方式：

// 上游任務(wù)完成后在當(dāng)前主線程中同步執(zhí)行處理，向下傳遞上游處理結(jié)果或異常
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
...
}

// 上游任務(wù)完成后在當(dāng)前子線程中異步執(zhí)行處理，向下傳遞上游處理結(jié)果或異常
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
...
}

// 上游任務(wù)完成后在當(dāng)前主線程中同步執(zhí)行處理，指定線程池，向下傳遞上游處理結(jié)果或異常
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
...
}

// 上游任務(wù)異常時(shí)處理，在主線程中完成
public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) {
...
}

示例代碼如下：

    @Test
    public void testWhenComplete() throws InterruptedException, ExecutionException {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            return "test";
        }).whenCompleteAsync((result, exception) -> {
            if (result != null) {
                System.out.println(result);
            } else {
                exception.printStackTrace();
            }
        });
        System.out.println(f1.get());

        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
            int i = 1 / 0;
            return "test";
        }).exceptionally((e) -> {
            e.printStackTrace();
            return "error";
        });
        System.out.println(f2.get());
    }
---
test
error
java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
...

whenComplete方法入?yún)⑹荁iConsumer，它是一個(gè)純消費(fèi)者函數(shù)，不會(huì)修改返回值及其類型。

下面一組接口除了whenComplete的功能外，同時(shí)具備轉(zhuǎn)換結(jié)果的功能，通過(guò)參數(shù)BiFunction實(shí)現(xiàn)：

public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
...
}

public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
...
}

public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
...
}

示例代碼如下：

    @Test
    public void testHandle() throws InterruptedException, ExecutionException {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            return "100";
        });
        CompletableFuture<Integer> f2 = f1.handleAsync((result, exception) -> {
            if(exception != null) {
                exception.printStackTrace();
            }
            return Integer.parseInt(result) * 10;
        });
        System.out.println(f2.get());
    }
---
1000

handle方法依然保留了對(duì)異常請(qǐng)款的處理，在BiFunction中指定其第二個(gè)參數(shù)類型是Throwble。

接下來(lái)一組方法thenApply與handle類似，也可以對(duì)上游結(jié)果進(jìn)行轉(zhuǎn)換，同時(shí)忽略對(duì)異常情況的處理。

public <U> CompletableFuture<U> thenApplyFunction<? super T,? extends U> fn){
...
}

public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) {
...
}

public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) {
...
}

示例代碼：

	@Test
    public void testThenApply() throws InterruptedException, ExecutionException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            //int i= 1/0;
            return "100";
        });
        CompletableFuture<Integer> f = future.thenApply(t -> Integer.parseInt(t) * 10).thenApply(t -> t * 10);
        System.out.println(f.get());
    }
---
10000

上述方法處理完成后，都會(huì)返回計(jì)算結(jié)果，CompletableFuture還提供了一組處理方法，只對(duì)上游處理結(jié)果進(jìn)行消費(fèi)，且沒有返回，如果上游發(fā)生異常，則不執(zhí)行該方法；同時(shí)，CompletableFuture還提供了一組加強(qiáng)版方法，提供兩個(gè)CompletableFuture任務(wù)都完成或完成一個(gè)時(shí)執(zhí)行的方法。

public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
...
}

public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
...
}

public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor) {
...
}

public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action) {
...
}

public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action) {
...
}

public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor) {
...
}
public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
...
}

public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
...
}

public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action,Executor executor) {
...
}

示例代碼：

    @Test
    public void testThenAccept() throws InterruptedException, ExecutionException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            int i = 1/0;
            return "1";
        });
        CompletableFuture<Void> f = future.thenAccept(t -> {
            System.out.println(t);
        });
    }
---

上述方法都依賴上游處理的結(jié)果，CompletableFuture還提供了一組方法，不依賴上游的結(jié)果，如下

public CompletableFuture<Void> thenRun(Runnable action) {
...
}

public CompletableFuture<Void> thenRunAsync(Runnable action) {
...
}

public CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor) {
...
}

示例代碼：

    @Test
    public void testThenRun() throws InterruptedException, ExecutionException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "result";
        });
        CompletableFuture<Void> f = future.thenRunAsync(() -> {
            System.out.println("future then run");
        });
    }
---
future then run

thenCompose這組方法與thenApply類似，都可以將上游執(zhí)行結(jié)果作為本stage入?yún)⒗^續(xù)計(jì)算，并轉(zhuǎn)換返回類型，不同的是thenCompose在一個(gè)新CompletableFuture對(duì)象執(zhí)行計(jì)算。

public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
...
}

public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
...
}

public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,Executor executor) {
...
}

示例代碼：

    @Test
    public void testThenCompose() throws InterruptedException, ExecutionException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            return "10";
        });
        CompletableFuture<Integer> f = future.thenCompose(v -> {
            return CompletableFuture.supplyAsync(() -> {
               return Integer.parseInt(v) * 100;
            });
        });
        System.out.println(f.get());
    }
---
1000

thenCombine這組方法與thenAccepBoth類似，不同的是thenCombine有返回值：

public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) {
...
}

public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) {
...
}

public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor) {
...
}

示例代碼：

    @Test
    public void testThenCombine() throws InterruptedException, ExecutionException {
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            return "abc";
        });
        CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> {
            return 123;
        });
        CompletableFuture<String> f3 = f1.thenCombine(f2, (m,n) -> {
            return m + " is " + n;
        });
        System.out.println(f3.get());
    }
---
abc is 123

最后兩個(gè)方法用于組合多個(gè)CompletableFuture，allOf方法是當(dāng)所有CompletableFuture執(zhí)行完成后執(zhí)行計(jì)算，anyOf方法是任意一個(gè)CompletableFuture執(zhí)行完成后執(zhí)行計(jì)算。

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) {
...
}

public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {
...
}

示例代碼：

    @Test
    public void testAllof() throws InterruptedException, ExecutionException {
        List<CompletableFuture<Integer>> list = new ArrayList<>();
        Random random = new Random();
        for (int i=1; i < 10; i++) {
            list.add(CompletableFuture.supplyAsync(() -> {
                return random.nextInt(100);
            }));
        }
        CompletableFuture<Void> fn = CompletableFuture.allOf(list.toArray(new CompletableFuture[list.size()]));
        CompletableFuture<List<Integer>> fr = fn.thenApply(v -> {
            return list.stream().map(CompletableFuture::join).collect(Collectors.toList()); 
        });
        System.out.println(fr.get());
    }
---
[3, 97, 11, 93, 22, 77, 68, 26, 89]