解读CompletableFuture的底层原理

更新时间：2024年09月18日 16:08:50 作者：拾木200

本文探讨了Java8中CompletableFuture的原理和应用,详解其异步编程能力、工作机制及实际使用方法,CompletableFuture通过链式调用和状态管理优化异步任务,提高Java应用的效率和性能

引言

在现代 Java 编程中，异步编程变得越来越重要。为了实现高效和非阻塞的代码，Java 8 引入了 CompletableFuture，一个用于构建异步应用程序的强大工具。

本文将详细探讨 CompletableFuture 的底层原理，展示其工作机制，并通过代码示例说明如何在实际应用中使用它。

异步编程的背景

异步编程是指在程序运行过程中，不等待某个操作完成，而是继续执行其他操作，待异步操作完成后再处理其结果。这样可以提高程序的效率，特别是在 I/O 操作和网络请求等耗时操作中。

在 Java 8 之前，实现异步编程主要依赖于 Future 接口。然而，Future 存在一些局限性，例如无法手动完成、不能链式调用等。为了解决这些问题，Java 8 引入了 CompletableFuture。

什么是 CompletableFuture

CompletableFuture 是 Java 8 中新增的类，实现了 Future 和 CompletionStage 接口，提供了强大的异步编程能力。

CompletableFuture 允许以非阻塞的方式执行任务，并且可以通过链式调用来组合多个异步操作。

CompletableFuture 的特点

手动完成：可以手动设置 CompletableFuture 的结果或异常。
链式调用：支持多个 CompletableFuture 的链式调用，形成复杂的异步任务流。
组合操作：提供了丰富的方法来组合多个异步任务，例如 thenCombine、thenAcceptBoth 等。
异常处理：提供了灵活的异常处理机制，可以在任务链中处理异常。

CompletableFuture 的底层原理

工作机制

CompletableFuture 的核心是基于 ForkJoinPool 实现的。ForkJoinPool 是一种特殊的线程池，适用于并行计算任务。它采用了工作窃取算法，能够有效利用多核 CPU 的性能。

当我们提交一个任务给 CompletableFuture 时，它会将任务提交到默认的 ForkJoinPool.commonPool() 中执行。我们也可以指定自定义的线程池来执行任务。

状态管理

CompletableFuture 具有以下几种状态：

未完成（Pending）：任务尚未完成。
完成（Completed）：任务已经成功完成，并返回结果。
异常（Exceptionally Completed）：任务在执行过程中抛出了异常。

这些状态通过内部的 volatile 变量来管理，并使用 CAS（Compare-And-Swap） 操作保证线程安全。

任务调度

CompletableFuture 的任务调度机制基于 ForkJoinPool 的工作窃取算法。当一个线程完成当前任务后，会从其他线程的任务队列中窃取任务执行，从而提高 CPU 利用率。

下面我们通过一个简单的示例代码来理解 CompletableFuture 的基本用法。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 创建一个 CompletableFuture 实例
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Hello, World!";
        });

        // 阻塞等待结果
        String result = future.get();
        System.out.println(result);
    }
}

在上面的示例中，我们创建了一个 CompletableFuture 实例，并使用 supplyAsync 方法异步执行任务。

supplyAsync 方法会将任务提交到默认的 ForkJoinPool 中执行。最后，我们使用 get 方法阻塞等待结果并打印输出。

链式调用

CompletableFuture 的一个重要特性是支持链式调用。

通过链式调用，我们可以将多个异步任务组合在一起，形成一个任务流。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureChainExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            return "Hello, World!";
        }).thenApply(result -> {
            return result + " from CompletableFuture";
        }).thenApply(String::toUpperCase);

        String finalResult = future.get();
        System.out.println(finalResult);
    }
}

在这个示例中，我们使用 thenApply 方法对前一个任务的结果进行处理，并返回一个新的 CompletableFuture 实例。

通过链式调用，我们可以将多个任务串联在一起，形成一个任务流。

组合操作

CompletableFuture 提供了多种方法来组合多个异步任务。以下是一些常用的组合操作示例：

1.thenCombine：组合两个 CompletableFuture，并将两个任务的结果进行处理。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureCombineExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 5);
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 10);

        CompletableFuture<Integer> combinedFuture = future1.thenCombine(future2, Integer::sum);

        System.out.println(combinedFuture.get());  // 输出 15
    }
}

2. thenAcceptBoth：组合两个 CompletableFuture，并对两个任务的结果进行消费处理。

import java.util.concurrent.CompletableFuture;

public class CompletableFutureAcceptBothExample {
    public static void main(String[] args) {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 5);
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 10);

        future1.thenAcceptBoth(future2, (result1, result2) -> {
            System.out.println("Result: " + (result1 + result2));
        }).join();
    }
}

3. allOf：组合多个 CompletableFuture，并在所有任务完成后执行操作。

import java.util.concurrent.CompletableFuture;

public class CompletableFutureAllOfExample {
    public static void main(String[] args) {
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            System.out.println("Task 1 completed");
        });

        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
            System.out.println("Task 2 completed");
        });

        CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(future1, future2);

        combinedFuture.join();
        System.out.println("All tasks completed");
    }
}

异常处理

在异步任务中处理异常是非常重要的。CompletableFuture 提供了多种方法来处理任务执行过程中的异常。

1.exceptionally：在任务抛出异常时，提供一个默认值。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureExceptionallyExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (true) {
                throw new RuntimeException("Exception occurred");
            }
            return "Hello, World!";
        }).exceptionally(ex -> {
            System.out.println("Exception: " + ex.getMessage());
            return "Default Value";
        });

        System.out.println(future.get());  // 输出 Default Value
    }
}

2. handle：无论任务是否抛出异常，都进行处理。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureHandleExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (true) {
                throw new RuntimeException("Exception occurred");
            }
            return "Hello, World!";
        }).handle((result, ex) -> {
            if (ex != null) {
                return "Default Value";
            }
            return result;
        });

        System.out.println(future.get());  // 输出 Default Value
    }
}

实战案例：构建异步数据处理管道

为了更好地理解 CompletableFuture 的实际应用，我们来构建一个异步数据处理管道。

假设我们有一个数据源，需要对数据进行一系列的处理操作，并将处理结果输出到文件中。

数据源模拟

我们首先模拟一个数据源，该数据源会生成一系列数据。

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class DataSource {
    public List<Integer> getData() {
        return IntStream.range(0, 10).boxed().collect(Collectors.toList());
    }
}

数据处理

接下来，我们定义数据处理操作。

假设我们需要对数据进行两步处理：首先对每个数据乘以 2，然后对结果进行累加。

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class DataProcessor {
    public List<Integer> processStep1(List<Integer> data) {
        return data.stream().map(x -> x * 2).collect(Collectors.toList());
    }

    public Integer processStep2(List<Integer> data) {
        return data.stream().reduce(0, Integer::sum);
    }

    public CompletableFuture<List<Integer>> processStep1Async(List<Integer> data) {
        return CompletableFuture.supplyAsync(() -> processStep1(data));
    }

    public CompletableFuture<Integer> processStep2Async(List<Integer> data) {
        return CompletableFuture.supplyAsync(() -> processStep2(data));
    }
}

结果输出

我们定义一个方法将处理结果输出到文件中。

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.CompletableFuture;

public class ResultWriter {
    public void writeResult(String fileName, Integer result) throws IOException {
        Files.write(Paths.get(fileName), result.toString().getBytes());
    }

    public CompletableFuture<Void> writeResultAsync(String fileName, Integer result) {
        return CompletableFuture.runAsync(() -> {
            try {
                writeResult(fileName, result);
            } catch (IOException e) {
                throw new IllegalStateException(e);
            }
        });
    }
}

主程序

最后，我们在主程序中将上述组件组合在一起，构建异步数据处理管道。

import java.util.List;
import java.util.concurrent.CompletableFuture;

public class Main {
    public static void main(String[] args) {
        DataSource dataSource = new DataSource();
        DataProcessor dataProcessor = new DataProcessor();
        ResultWriter resultWriter = new ResultWriter();

        List<Integer> data = dataSource.getData();

        CompletableFuture<List<Integer>> step1Future = dataProcessor.processStep1Async(data);
        CompletableFuture<Integer> step2Future = step1Future.thenCompose(dataProcessor::processStep2Async);
        CompletableFuture<Void> writeFuture = step2Future.thenCompose(result -> resultWriter.writeResultAsync("result.txt", result));

        writeFuture.join();
        System.out.println("Data processing completed");
    }
}

在这个例子中，我们使用 CompletableFuture 将数据处理步骤和结果输出串联在一起，形成了一个完整的异步数据处理管道。

通过 thenCompose 方法，我们将前一个任务的结果传递给下一个异步任务，从而实现了链式调用。

总结

本文深入探讨了 CompletableFuture 的底层原理，展示了其工作机制，并通过多个代码示例说明了如何在实际应用中使用 CompletableFuture。通过理解 CompletableFuture 的异步编程模型、状态管理、任务调度和异常处理机制，我们可以更好地利用这一强大的工具构建高效、非阻塞的 Java 应用程序。

希望这篇文章能够帮助你全面理解 CompletableFuture，并在实际开发中灵活应用。这些仅为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

JVM调优参数的设置
Java虚拟机的调优是一个复杂而关键的任务,可以通过多种参数来实现,本文就来介绍一下JVM调优参数的设置,具有一定的参考价值,感兴趣的可以了解一下
2024-03-03
SpringBoot项目中的视图解析器问题(两种)
SpringBoot官网推荐使用HTML视图解析器,但是根据个人的具体业务也有可能使用到JSP视图解析器，所以本文介绍了两种视图解析器，感兴趣的可以了解下
2020-06-06
Maven包冲突导致NoSuchMethodError错误的解决办法
web 项目能正常编译,运行时也正常启动,但执行到需要调用 org.codehaus.jackson 包中的某个方法时,产生运行异常,这篇文章主要介绍了Maven包冲突导致NoSuchMethodError错误的解决办法,需要的朋友可以参考下
2024-05-05
Java JDK动态代理（AOP）的实现原理与使用详析
所谓代理，就是一个人或者一个机构代表另一个人或者另一个机构采取行动。下面这篇文章主要给大家介绍了关于Java JDK动态代理（AOP）实现原理与使用的相关资料，文中通过示例代码介绍的非常详细，需要的朋友可以参考借鉴，下面来一起看看吧。
2017-07-07
Java Spring的数据库开发详解
这篇文章主要介绍了Spring的数据库开发，主要围绕SpringJDBC和Spring Jdbc Template两个技术来讲解，文中有详细的代码示例，需要的小伙伴可以参考一下
2023-04-04
SpringSecurity 自定义认证登录的项目实践
本文主要介绍了SpringSecurity 自定义认证登录的项目实践,以手机验证码登录为例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2024-08-08
java如何确定一个链表有环及入口节点
这篇文章主要介绍了java如何确定一个链表有环及入口节点，想了解数据结构的同学可以参考下
2021-04-04
java比较器Comparable接口与Comaprator接口的深入分析
本篇文章是对java比较器Comparable接口与Comaprator接口进行了详细的分析介绍，需要的朋友参考下
2013-06-06
SpringBoot中实现代理方式
这篇文章主要介绍了SpringBoot中实现代理方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-06-06
Java 设计模式以虹猫蓝兔的故事讲解简单工厂模式
简单工厂模式是属于创建型模式，又叫做静态工厂方法（Static Factory Method）模式，但不属于23种GOF设计模式之一。简单工厂模式是由一个工厂对象决定创建出哪一种产品类的实例。简单工厂模式是工厂模式家族中最简单实用的模式，可以理解为是不同工厂模式的一个特殊实现
2022-03-03