跳至主要內容

掌握 Future & CompletableFuture

God约 3738 字大约 12 分钟

1、Future

1.1 Future的使用场景

Future 类是异步思想的典型运用，主要用在一些耗时任务的场景！

具体来说：当我们执行某一耗时任务时，可以将这个耗时任务交给一个子线程去异步处理，与此同时我们可以干点其他事情，不用阻塞在这里傻傻的等待该耗时任务的完成；一段时间后，当我们已完成其他事情后，再通过 Future 类提供的方法获取该耗时任务的执行结果；这样一来，程序的整体执行效率就明显提高了！

1.2 Future提供的方法

Future 类只是一个接口，它的基础实现是FutureTask！从下面的类图可以看出，FutureTask实现了RunnableFuture接口，则RunnableFuture接口继承了Runnable接口和Future接口，同时FutureTask中还有一个成员变量Callable：

FutureTask 有两个构造函数，可传入 Callable 或者 Runnable 对象。实际上，传入 Runnable 对象也会在方法内部转换为Callable 对象：

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;
}
public FutureTask(Runnable runnable, V result) {
    // 通过适配器RunnableAdapter来将Runnable对象runnable转换成Callable对象
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;
}

Future 类只是一个泛型接口，其中定义了 5 个方法，主要包括下面这 4 个功能：

  • 取消任务
  • 判断任务是否被取消
  • 判断任务是否已经执行完成
  • 获取任务执行结果
// V 代表了Future执行的任务返回值的类型
public interface Future<V> {
    // 取消任务执行; 取消成功返回 true，否则返回 false
    boolean cancel(boolean mayInterruptIfRunning);
    // 判断任务是否被取消
    boolean isCancelled();
    // 判断任务是否已经执行完成
    boolean isDone();
    // 获取任务执行结果
    V get() throws InterruptedException, ExecutionException;
    // 指定时间内没有返回计算结果就抛出 TimeOutException 异常
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
}

FutureTask 常用来封装 Callable 和 Runnable，所以FutureTask既能当做一个Runnable or Callable直接被 Thread 执行，也可以作为一个任务提交到线程池中执行！ExecutorService.submit() 方法的返回值类型其实就是 Future 的实现类 FutureTask ：

<T> Future<T> submit(Callable<T> task);
Future<?> submit(Runnable task);

1.3 Future使用示例

public class MyTest {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();

        // 异步执行耗时任务
        Future future = executorService.submit(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                Long start = System.currentTimeMillis();
                while (true) {
                    Long current = System.currentTimeMillis();
                    if ((current - start) > 1000) {
                        return 1;
                    }
                }
            }
        });

        // 其他事情start
        // ...
        // 其他事情end

        // 获取异步耗时任务的结果
        try {
            Integer result = (Integer)future.get();
            System.out.println(result);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

1.4 Future有什么局限性？

案例1：get方法容易引起程序阻塞

如果子任务没有计算完成，而此时在主线程中调用了get()方法，则主线程会在调用get()的地方阻塞住，并且无法手动设置任务结果或完成，只能等待子任务计算完成或者达到超时时间程序才能继续往下执行。如下案例：

public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
    ExecutorService executor = Executors.newSingleThreadExecutor();

    FutureTask<String> futureTask = new FutureTask<>(() -> {
        sleep(5); //5秒
        return "hello";
    });
    Future<?> future = executor.submit(futureTask);

    // do something
    // ...

    // 调用即阻塞，直到子任务futureTask完成，才能继续执行后续任务
    // System.out.println("Result: " + future.get());
    // 调用即阻塞，直到子任务futureTask完成或达到超时时间，才能继续执行后续任务
    System.out.println("Result: " + future.get(3, TimeUnit.SECONDS));

    System.out.println("--后续任务start--");
    // do something
    // ...
    System.out.println("--后续任务end--");
}

案例2：当多个Future之间存在依赖关系时，Future无法很好的处理

举例，当第1个 Future 的返回值是第2个 Future 的输入，需要开发人员严格控制好子任务的先后执行顺序：

ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> firstApiCallResult = executor.submit(
    () -> firstApiCall(someValue)
);

String firstResult = firstApiCallResult.get(); // 主线程阻塞

Future<String> secondApiCallResult = executor.submit(
    () -> secondApiCall(firstResult)
);

如上代码，可以看到，第2个 Future 需要等待第1个 Future的返回值，而且第1个 Future 的返回值是在主线程中阻塞获取的，即无法很好的支持异步任务编排！


除了以上说的2个重点局限性，其实Future还个问题，即缺乏异常处理：Future.get()方法会抛出ExecutionException异常，其中包装了异步计算中发生的异常，但这种方式不直观，可能需要额外的异常处理机制。


为了解决Future的局限性，Java 8引入了CompletableFuture类，它提供了更丰富的功能，包括更灵活的异步计算控制、异常处理和组合多个异步任务的结果等功能。因此，如果需要更强大和灵活的异步编程功能，推荐使用CompletableFuture类！

2、CompletableFuture

2.1 为什么要有CompletableFuture

Future 存在一些局限性，比如获取计算结果的 get() 方法为阻塞调用、不支持异步任务的编排组合等。

为了解决这些问题，Java 8 引入了CompletableFuture类，该类同时实现了Future 和CompletionStage接口，用于支持异步编程和处理异步计算的结果。

CompletableFuture提供了一种更加灵活和强大的方式来处理异步任务，支持异步任务的串行和并行执行、异常处理、结果转换、组合多个异步任务的结果等功能！


下面我们来简单看看 CompletableFuture 类的定义，发现CompletableFuture 同时实现了 Future 和 CompletionStage 接口：

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

CompletionStage是Java 8中引入的接口，用于表示一个异步计算的阶段或阶段性结果；

它提供了一种更加灵活和强大的方式来处理异步计算的结果，支持异步任务的串行和并行执行、异常处理、结果转换等功能；

可以认为它是Future接口的扩展，提供了更多的操作方法和组合方法，使得异步编程更加方便和高效！

2.2 CompletableFuture中常用方法

CompletableFuture 的函数式能力就是CompletionStage 接口赋予的，从这个接口的方法参数你就可以发现其大量使用了 Java8 引入的函数式编程：

thenApply(Function<? super T,? extends U> fn)：对异步计算的结果进行转换，并返回一个新的CompletionStage对象。

thenAccept(Consumer<? super T> action)：对异步计算的结果执行一个消费操作，不返回结果。

exceptionally(Function<Throwable, ? extends T> fn)：处理异步计算中发生的异常，返回一个新的CompletionStage对象。

thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)：组合两个异步计算的结果，并返回一个新的CompletionStage对象。

通过实现CompletionStage接口，可以更加灵活地处理异步计算的结果、控制异步任务的执行流程，提高代码的可读性和可维护性。


CompletableFuture类实现了CompletionStage接口，不仅实现了CompletionStage接口中的所有方法，还扩展出一系列好用的方法来支持异步编程：

complete()：用于手动完成一个CompletableFuture对象，即设置异步任务的结果。通过调用complete方法，可以将指定的结果传递给CompletableFuture对象，使其进入已完成状态。

supplyAsync()：静态工厂方法，用于提交一个带返回值的异步任务。可以传入一个Supplier函数式接口，表示异步计算的结果。

thenSupplyAsync()：用于对异步计算的结果进行转换，并返回一个新的CompletableFuture对象。可以传入一个Function函数式接口，对结果进行处理。

thenApply()：对异步计算的结果进行转换，并返回一个新的CompletableFuture对象。可以传入一个Function函数式接口，对结果进行处理。

thenAccept()：对异步计算的结果执行一个消费操作，不返回结果。可以传入一个Consumer函数式接口，对结果进行消费。

thenCompose()：组合两个异步任务的结果，并返回一个新的CompletableFuture对象。可以传入一个Function函数式接口，将两个结果组合成一个新的结果。

thenCombine()：组合两个CompletableFuture对象的结果，并返回一个新的CompletableFuture对象。可以传入一个BiFunction函数式接口，将两个结果组合成一个新的结果。

allOf()：静态方法，等待所有CompletableFuture对象都完成后返回一个新的CompletableFuture对象。

anyOf()：静态方法，等待任意一个CompletableFuture对象完成后返回一个新的CompletableFuture对象。

handle()：处理异步计算的结果或异常情况，并返回一个新的CompletableFuture对象。可以传入一个BiFunction函数式接口，处理结果或异常并返回新的结果。

exceptionally()：处理异步计算中发生的异常情况。可以传入一个Function函数式接口，对异常进行处理并返回一个默认值。

CompletableFuture中的这些方法可以帮助开发者更好地处理异步任务的结果、控制异步任务的执行流程，以及组合多个异步任务的结果。通过合理地使用CompletableFuture中的方法，可以实现更加灵活和高效的异步编程！

2.3 CompletableFuture的使用示例

针对案例1的问题，CompletableFuture 如何解决的呢？CompletableFuture 有个complete(T value)方法，它可以手动结束执行中的任务。回顾下案例1的代码及 CompletableFuture 的方法：

案例1：

public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
    ExecutorService executor = Executors.newSingleThreadExecutor();

    FutureTask<String> futureTask = new FutureTask<>(() -> {
        sleep(5); //5秒
        return "hello";
    });
    Future<?> future = executor.submit(futureTask);

    // do something
    // ...

    // 调用即阻塞，直到子任务futureTask完成，才能继续执行后续任务
    // System.out.println("Result: " + future.get());
    // 调用即阻塞，直到子任务futureTask完成或达到超时时间，才能继续执行后续任务
    System.out.println("Result: " + future.get(3, TimeUnit.SECONDS));

    System.out.println("--后续任务start--");
    // do something
    // ...
    System.out.println("--后续任务end--");
}

案例1-问题解决代码：

public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        sleep(5); //5秒
        return "hello";
    });

    // do something
    // ...

    if(future.isDone()) {
        // 调用即阻塞，直到子任务futureTask完成，才能继续执行后续任务
        System.out.println("Result: " + future.get());
        // 调用即阻塞，直到子任务futureTask完成或达到超时时间，才能继续执行后续任务
        // System.out.println("Result: " + future.get(3, TimeUnit.SECONDS));
    } else {
        future.complete("completed");
    }

    System.out.println("--后续任务start--");
    // do something
    // ...
    System.out.println("--后续任务end--");
}

案例2的问题是两个有关联的 Future如何能真正做到异步呢？CompletableFuture的链式(chain)方法就是答案。回顾下案例2的代码及 CompletableFuture 的方法：

案例2：

ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> firstApiCallResult = executor.submit(
    () -> firstApiCall(someValue)
);

String firstResult = firstApiCallResult.get(); // 主线程阻塞

Future<String> secondApiCallResult = executor.submit(
    () -> secondApiCall(firstResult)
);

案例2-问题解决代码：

var finalResult = CompletableFuture.supplyAsync(
     () -> firstApiCall(someValue)
).thenApply(firstApiResult -> secondApiCall(firstApiResult));

可以看到，使用CompletableFuture链式(chain)方法期间，没有和主线程有任何交互。更进一步，你可以在每个链式(chain)方法中打印下线程名，你会发现都不是主线程名。也就是说，CompletableFuture链式(chain)方法完全做到了全程无阻塞！

3、总结

3.1 Future的优点

Future 类是异步思想的典型运用，主要用在一些耗时任务的场景！

具体来说：当我们执行某一耗时任务时，可以将这个耗时任务交给一个子线程去异步处理，与此同时我们可以干点其他事情，不用阻塞在这里傻傻的等待该耗时任务的完成；一段时间后，当我们已完成其他事情后，再通过 Future 类提供的方法获取该耗时任务的执行结果；这样一来，程序的整体执行效率就明显提高了！


表示异步计算结果：Future接口表示一个异步计算的结果，可以通过Future对象获取异步任务的结果或取消任务的执行。

支持阻塞获取结果：通过Future对象的get方法可以阻塞等待异步任务的完成，并获取任务的结果，或者在超时时间内获取结果。

支持取消任务：Future对象提供了cancel方法，可以取消异步任务的执行，但需要根据具体情况判断任务是否可以被取消。

简单易用：Future接口相对简单易用，可以方便地表示异步任务的结果，适用于简单的异步编程场景。

3.2 Future的局限性

容易引起程序阻塞：get()方法为阻塞方法，在主线程中调用会将主线程阻塞在调用get()的地方。

无法链式调用：Future接口本身不支持链式调用，如果需要对多个异步任务进行组合和处理，可能需要额外的代码来管理异步任务的执行顺序。

无法处理异常：Future接口在获取异步任务结果时无法处理任务中可能发生的异常，需要开发者自行处理异常情况。


总的来说，Future接口是Java中表示异步计算结果的基础接口，适用于简单的异步编程场景。然而，由于其局限性，开发者在复杂的异步编程场景中可能需要更强大和灵活的工具，如CompletableFuture等，来更好地处理异步任务的结果和处理逻辑！

3.3 强大的CompletableFuture

为了解决Future的问题，Java 8 引入了CompletableFuture类，该类提供了一种更加灵活和强大的方式来处理异步任务，支持异步任务的串行和并行执行、异常处理、结果转换、组合多个异步任务的结果等功能！


支持异步编程：CompletableFuture可以用于处理异步任务，允许开发者在一个线程中执行异步操作，并在任务完成时获取结果或执行后续操作。

支持链式调用：CompletableFuture提供了一系列方法，如supplyAsync、thenSupplyAsync等，可以方便地组合和处理异步任务的结果，实现链式调用。

支持异常处理：CompletableFuture提供了exceptionally、handle等方法，用于处理异步任务中可能发生的异常情况，帮助开发者更好地处理异常。

支持组合操作：CompletableFuture提供了thenCombine、allOf、anyOf等方法，可以组合多个CompletableFuture对象的结果，实现并行处理和组合操作。

支持自定义完成结果：通过complete方法，可以手动设置CompletableFuture对象的结果，触发后续处理逻辑。


总的来说，CompletableFuture是Java中强大的异步编程工具，能够简化异步任务的处理和管理，提高代码的灵活性和并发性能。然而，需要开发者根据具体情况合理使用CompletableFuture，以确保异步编程的效率和可靠性！

MIT Licensed | Copyright © 2024-present God