Java CompletableFuture 详细使用教程与实践

谁践踏了优雅 2024-03-24 23:48 30阅读 0赞

### 一、Java CompletableFuture 详细使用教程 ###

Java 8引入了一种强大的异步编程工具：`CompletableFuture`。它提供了一种处理异步计算的方式，使得你可以在计算完成时获取结果，或者将一个或多个 `CompletableFuture` 的结果组合在一起。本部分将详细解析 `CompletableFuture` 的各个方面，包括创建、组合、处理异常等，并通过示例来展示其使用方法。

#### 1.1 创建 CompletableFuture ####

创建 `CompletableFuture` 的最简单方法是使用无参数构造函数：

CompletableFuture<Void> future = new CompletableFuture<>();

这将创建一个未完成的 `CompletableFuture`。你可以通过 `complete` 方法来完成它：

future.complete(null);

如果你想创建一个已经完成的 `CompletableFuture`，你可以使用 `completedFuture` 方法：

CompletableFuture<String> future = CompletableFuture.completedFuture("Hello, world!");

此外，你还可以使用 `supplyAsync` 方法来创建一个异步计算的 `CompletableFuture`：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        // 这里是一些长时间运行的计算
        return "Hello, world!";
    });

#### 1.2 处理 CompletableFuture 的结果 ####

`CompletableFuture` 提供了一系列的方法来处理异步计算的结果。这些方法都返回一个新的 `CompletableFuture`，这样你可以将它们链接在一起形成一个处理管道。

例如，你可以使用 `thenApply` 方法来对结果进行转换：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
            .thenApply(s -> s + ", world!");

你还可以使用 `thenAccept` 方法来消费结果：

CompletableFuture.supplyAsync(() -> "Hello")
            .thenAccept(s -> System.out.println(s + ", world!"));

如果你不关心结果，只想在计算完成后执行一些操作，你可以使用 `thenRun` 方法：

CompletableFuture.supplyAsync(() -> "Hello")
            .thenRun(() -> System.out.println("Computation finished."));

#### 1.3 组合 CompletableFuture ####

`CompletableFuture` 提供了一系列的方法来组合多个异步计算。

例如，你可以使用 `thenCompose` 方法来将两个异步计算串联起来：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
            .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + ", world!"));

你还可以使用 `thenCombine` 方法来将两个异步计算并联起来：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> ", world!");
    
    CompletableFuture<String> future = future1.thenCombine(future2, String::concat);

如果你有多个 `CompletableFuture`，你可以使用 `allOf` 方法来等待它们全部完成：

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> ", world!");
    
    CompletableFuture<Void> future = CompletableFuture.allOf(future1, future2);

#### 1.4 处理 CompletableFuture 的异常 ####

`CompletableFuture` 提供了两个方法来处理异常：`exceptionally` 和 `handle`。

`exceptionally` 方法接受一个函数，这个函数将在计算抛出异常时被调用，它的返回值将作为新的结果：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (true) throw new RuntimeException("Exception!");
        return "Hello, world!";
    }).exceptionally(ex -> "Sorry, we have an error!");

`handle` 方法和 `exceptionally` 类似，但是它可以处理正常的结果和异常：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (true) throw new RuntimeException("Exception!");
        return "Hello, world!";
    }).handle((res, ex) -> {
        if (ex != null) {
            return "Sorry, we have an error!";
        } else {
            return res;
        }
    });

以上就是 `CompletableFuture` 的主要用法。通过组合这些方法，你可以创建出复杂的异步处理管道，大大提高程序的性能和响应性。

### 二、使用 CompletableFuture 结合 MyBatis 和线程池批量插入数据 ###

在处理大数据时，我们经常需要在数据库中插入大量的数据。如果我们使用传统的同步方法，可能会花费很长的时间。在这部分，我将展示如何使用 Java 的 `CompletableFuture` 进行异步处理，结合 MyBatis 和线程池来批量插入30万条数据，提高数据处理的效率。

#### 2.1 创建线程池 ####

首先，我们需要创建一个线程池，以便并发执行多个插入任务。我们可以使用 Java 的 `Executors` 类来创建线程池：

ExecutorService executor = Executors.newFixedThreadPool(10);

这将创建一个拥有10个线程的线程池。

#### 2.2 创建数据 ####

然后，我们需要创建要插入的数据。假设我们要插入一些用户数据，每个用户有一个名字和一个年龄：

class User {
        private String name;
        private int age;
    
        // getters and setters...
    }

我们可以创建一个方法来生成用户数据：

List<User> generateUsers(int count) {
        List<User> users = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            User user = new User();
            user.setName("User" + i);
            user.setAge(i % 100);
            users.add(user);
        }
        return users;
    }

#### 2.3 插入数据 ####

接下来，我们需要创建一个方法来插入数据。我们将使用 MyBatis 的 `SqlSession` 来执行插入操作：

void insertUsers(SqlSession sqlSession, List<User> users) {
        UserMapper userMapper = sqlSession.getMapper(UserMapper.class);
        for (User user : users) {
            userMapper.insert(user);
        }
        sqlSession.commit();
    }

注意，我们在插入所有用户后提交了事务。这是因为在大多数数据库中，事务提交是一个昂贵的操作，我们应该尽量减少事务提交的次数。

#### 2.4 使用 CompletableFuture 插入数据 ####

现在，我们可以使用 `CompletableFuture` 来并发插入数据。我们将数据分成多个批次，每个批次创建一个 `CompletableFuture` 来插入：

List<User> users = generateUsers(300000);
    int batchSize = 1000;
    List<CompletableFuture<Void>> futures = new ArrayList<>();
    
    for (int i = 0; i < users.size(); i += batchSize) {
        List<User> batch = users.subList(i, Math.min(users.size(), i + batchSize));
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            try (SqlSession sqlSession = sqlSessionFactory.openSession()) {
                insertUsers(sqlSession, batch);
            }
        }, executor);
        futures.add(future);
    }
    
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

在上面的代码中，我们首先生成了300000个用户数据，然后将这些数据分成大小为1000的批次。对于每个批次，我们创建一个 `CompletableFuture` 来插入数据，然后将这个 `CompletableFuture` 添加到一个列表中。最后，我们使用 `CompletableFuture.allOf` 来等待所有的 `CompletableFuture` 完成。

#### 2.5 关闭线程池 ####

最后，我们需要关闭线程池。我们可以使用 `ExecutorService.shutdown` 方法来关闭线程池：

executor.shutdown();

这将等待所有已提交的任务完成，然后关闭线程池。

以上就是使用 `CompletableFuture` 结合 MyBatis 和线程池批量插入数据的方法。通过这种方式，我们可以大大提高插入数据的效率。希望这篇文章能帮助大家更好地理解和使用 `CompletableFuture`。

![b94a03686c7348fa889d7bf3464cfdf2.png][]

[b94a03686c7348fa889d7bf3464cfdf2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/2be0471a61024c50938c5c2c2cfece7d.png