MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java 中 CompletableFuture 任务异步回调 thenAccept 方法

2021-10-272.8k 阅读

Java 中 CompletableFuture 任务异步回调 thenAccept 方法

在 Java 并发编程领域，CompletableFuture 是一个强大的工具，它提供了一种异步执行任务并处理结果的优雅方式。thenAccept 方法作为 CompletableFuture 众多方法中的一员，在处理异步任务结果时扮演着重要角色。

thenAccept 方法的基本概念

thenAccept 方法用于在 CompletableFuture 任务完成时，接受任务的结果并进行处理，但不返回新的结果。它的定义如下：

public CompletableFuture<Void> thenAccept(Consumer<? super U> action)

其中，action 是一个 Consumer 类型的函数式接口，它接受 CompletableFuture 任务完成后的结果作为参数，并对其进行处理。该方法返回一个新的 CompletableFuture，当 action 执行完毕后，这个新的 CompletableFuture 就会完成，并且其结果为 null。

简单示例

下面通过一个简单的代码示例来展示 thenAccept 方法的基本用法：

import java.util.concurrent.CompletableFuture;

public class ThenAcceptExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            // 模拟一个耗时操作
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Hello, CompletableFuture!";
        }).thenAccept(result -> {
            System.out.println("Received result: " + result);
        });

        // 主线程不要立即退出
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，supplyAsync 方法创建了一个异步任务，该任务模拟了一个耗时 2 秒的操作，并返回一个字符串结果。thenAccept 方法接收到这个结果，并将其打印出来。注意，为了确保主线程在异步任务完成之前不退出，我们在主线程中添加了一个 3 秒的 Thread.sleep。

深入理解 thenAccept 的执行流程

当 CompletableFuture 任务完成时，thenAccept 方法注册的 Consumer 会被提交到一个线程池中执行（默认情况下，使用 ForkJoinPool.commonPool()）。如果 CompletableFuture 任务已经完成，thenAccept 方法会立即将 Consumer 提交到线程池中执行。

假设我们有一个更复杂的场景，需要在异步任务完成后进行一系列的数据库操作。例如，我们从数据库中查询用户信息，然后更新用户的登录时间：

import java.util.concurrent.CompletableFuture;

public class DatabaseOperationExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            // 模拟查询用户信息
            System.out.println("Querying user information from database...");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "User: John Doe";
        }).thenAccept(userInfo -> {
            System.out.println("Received user information: " + userInfo);
            // 模拟更新用户登录时间
            System.out.println("Updating user login time...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        // 主线程不要立即退出
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，supplyAsync 模拟了从数据库查询用户信息的操作，thenAccept 则模拟了更新用户登录时间的操作。这两个操作都是异步执行的，通过 CompletableFuture 和 thenAccept，我们可以将它们串联起来，使得整个流程更加清晰和高效。

thenAccept 与异常处理

在实际应用中，异步任务可能会抛出异常。CompletableFuture 提供了多种方式来处理这些异常，thenAccept 也不例外。如果 CompletableFuture 任务在执行过程中抛出异常，thenAccept 注册的 Consumer 不会被执行，而是会触发异常处理机制。

我们可以通过 exceptionally 方法来处理异常。例如：

import java.util.concurrent.CompletableFuture;

public class ExceptionHandlingExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            // 模拟一个可能抛出异常的操作
            if (Math.random() < 0.5) {
                throw new RuntimeException("Simulated exception");
            }
            return "Success result";
        }).thenAccept(result -> {
            System.out.println("Received result: " + result);
        }).exceptionally(ex -> {
            System.out.println("Caught exception: " + ex.getMessage());
            return null;
        });

        // 主线程不要立即退出
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，supplyAsync 任务有 50% 的概率抛出一个运行时异常。thenAccept 方法在任务正常完成时处理结果，而 exceptionally 方法在任务抛出异常时捕获并处理异常。

thenAccept 与其他 CompletableFuture 方法的组合使用

CompletableFuture 提供了丰富的方法，thenAccept 可以与其他方法组合使用，以实现更复杂的异步操作。例如，我们可以结合 thenApply 方法，先对结果进行转换，再使用 thenAccept 进行处理。

import java.util.concurrent.CompletableFuture;

public class MethodChainingExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> "Hello")
                .thenApply(String::toUpperCase)
                .thenAccept(System.out::println);

        // 主线程不要立即退出
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，supplyAsync 方法返回一个字符串 "Hello"，thenApply 方法将其转换为大写形式，最后 thenAccept 方法将转换后的结果打印出来。

并发场景下的 thenAccept

在并发场景中，thenAccept 方法的行为需要特别注意。由于 thenAccept 注册的 Consumer 是在一个线程池中执行的，可能会出现多个 CompletableFuture 任务的 thenAccept 操作并发执行的情况。

例如，我们有多个异步任务，每个任务完成后都使用 thenAccept 来更新一个共享资源：

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

public class ConcurrentThenAcceptExample {
    private static AtomicInteger sharedCounter = new AtomicInteger(0);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture[] futures = new CompletableFuture[10];
        for (int i = 0; i < 10; i++) {
            futures[i] = CompletableFuture.supplyAsync(() -> 1)
                    .thenAccept(result -> sharedCounter.addAndGet(result));
        }

        CompletableFuture.allOf(futures).get();
        System.out.println("Final counter value: " + sharedCounter.get());
    }
}

在这个示例中，我们创建了 10 个异步任务，每个任务都返回 1，并使用 thenAccept 来更新一个共享的 AtomicInteger。通过 CompletableFuture.allOf 方法，我们等待所有任务完成，然后打印出最终的计数器值。由于 AtomicInteger 是线程安全的，所以在并发更新时不会出现数据竞争问题。

thenAccept 方法的性能考虑

在使用 thenAccept 方法时，性能是一个需要考虑的因素。由于 thenAccept 注册的 Consumer 是在一个线程池中执行的，如果 Consumer 执行的操作比较耗时，可能会导致线程池中的线程被长时间占用，从而影响其他任务的执行。

为了优化性能，我们可以考虑以下几点：

  1. 尽量减少 Consumer 中的耗时操作：将复杂的操作分解为多个步骤，或者将部分操作移到异步任务的前期执行。
  2. 调整线程池大小：根据实际应用场景，合理调整线程池的大小，以确保有足够的线程来处理 thenAccept 操作。
  3. 使用合适的线程池：对于不同类型的任务，可以选择不同的线程池实现，例如 ThreadPoolExecutor 可以根据任务的特性进行更精细的配置。

实际应用场景

thenAccept 方法在实际应用中有很多场景，以下是一些常见的例子：

  1. 异步日志记录：在一个异步任务完成后，使用 thenAccept 方法将任务的结果记录到日志中。
  2. 数据处理流水线：在数据处理的流水线中，一个任务的结果作为下一个任务的输入，thenAccept 可以用于在任务之间传递和处理数据。
  3. 缓存更新：当从数据库中获取数据后，使用 thenAccept 方法更新缓存，以确保缓存数据的一致性。

例如，在一个电商系统中，当用户下单后，我们可以使用 CompletableFuture 和 thenAccept 来异步更新库存和记录订单日志：

import java.util.concurrent.CompletableFuture;

public class EcommerceExample {
    public static void main(String[] args) {
        CompletableFuture.supplyAsync(() -> {
            // 模拟下单操作
            System.out.println("Placing order...");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Order placed successfully";
        }).thenAccept(orderResult -> {
            // 更新库存
            System.out.println("Updating inventory...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).thenAccept(unused -> {
            // 记录订单日志
            System.out.println("Logging order...");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        // 主线程不要立即退出
        try {
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在这个示例中，下单操作完成后，通过 thenAccept 方法依次执行库存更新和订单日志记录操作，整个流程都是异步执行的，提高了系统的响应性能。

总结

CompletableFuture 的 thenAccept 方法是 Java 并发编程中处理异步任务结果的重要工具。通过深入理解其基本概念、执行流程、异常处理以及与其他方法的组合使用，我们可以更加灵活和高效地编写异步代码。在实际应用中，根据不同的场景和性能需求，合理使用 thenAccept 方法，可以提升系统的并发处理能力和响应性能。无论是简单的异步任务处理，还是复杂的并发场景，thenAccept 都能为我们提供强大的支持。同时，在使用过程中，要注意性能优化和异常处理，确保代码的稳定性和可靠性。通过不断地实践和总结，我们可以更好地掌握 thenAccept 方法的使用技巧，为开发高性能的 Java 应用程序打下坚实的基础。