Java多线程工具CompletableFuture的使用教程

介绍

在 Java 1.8 版本中，加入了 CompletableFuture 类，它是一种新的 Future 类型，用于异步计算任务的完成（无需调用线程池提供的线程）。CompletableFuture 可以将异步操作串行化，也可以将多个异步操作组合和并为一个结果。本文将全面介绍 CompletableFuture 的使用方法，并提供示例代码。

CompletableFuture 的创建

要使用 CompletableFuture，可以通过以下两种方式创建：

1. 调用静态工厂方法 CompletableFuture.supplyAsync()，传入实现 Supplier 接口的逻辑代码，返回 CompletableFuture 对象：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello, CompletableFuture");

1. 调用 CompletableFuture 构造器，手动编写 CompletableFuture 的调用逻辑：

CompletableFuture<String> future = new CompletableFuture<>();
future.complete("Hello, CompletableFuture");

CompletableFuture 的调用

阻塞等待

调用 CompletableFuture 的 get() 方法可以阻塞等待异步操作的完成，并获取操作的结果。例如：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello, CompletableFuture");
String result = future.get();
System.out.println(result);

在 get() 调用处，如果异步操作没有完成，get() 方法会阻塞等待操作完成。如果操作已经完成，get() 方法会立即返回异步操作的结果。

异步计算

CompletableFuture 通过 applyAsync() 方法实现异步计算。例如：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 10)
    .thenApplyAsync(result -> result * 2);
int result = future.get();
System.out.println(result);

首先，定义了一个数字 10。在 thenApplyAsync() 调用时，将 10 作为参数传递给 lambda 表达式进行加 2 的运算。这个计算过程并不会阻塞当前线程，而是异步进行，输出结果为 20。

异步消费

CompletableFuture 通过 thenAcceptAsync() 方法实现异步消费。例如：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 10)
    .thenAcceptAsync(result -> {
        System.out.println(result);
    });

调用 thenAcceptAsync() 方法，将结果输出。

异步组合

可以通过 CompletableFuture 的 thenCompose()、thenCombine()、whenComplete() 和 exceptionally() 方法实现 CompletableFuture 的异步组合。

• thenCompose() 方法允许你将两个异步操作进行串联，并返回组合操作的结果。

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> 2)
  .thenCompose(res -> CompletableFuture.supplyAsync(() -> res * 10))
  .thenCompose(res -> CompletableFuture.supplyAsync(() -> res + 5));

在这个例子中，首先用 supplyAsync() 方法来提供一个整数 2。然后将此整数与另一个异步操作的结果进行组合。最后，将上一次操作的结果与 5 相加。

• thenCombine() 方法类似于 thenCompose() 方法，但是它并非是将两个异步操作进行串联，而是将两个异步操作进行并联，并返回组合操作的结果。

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 2);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 3);
CompletableFuture<Integer> future = future1.thenCombine(future2, (i1, i2) -> i1 + i2);
int result = future.get();
System.out.println(result);

在这个例子中，创建了两个异步操作 future1 和 future2，分别返回 2 和 3。然后使用 thenCombine() 方法将两个操作进行并联，将两个操作的结果相加，最后输出结果为 5。

• whenComplete() 方法在异步操作完成时进行回调，类似于使用 try-catch 结构。

CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> "Hello, CompletableFuture")
  .thenApplyAsync(String::toUpperCase)
  .whenComplete((res, throwable) -> {
    if (throwable != null) {
      throwable.printStackTrace();
    } else {
      System.out.println(res);
    }
  });

在这个例子中，首先使用 supplyAsync() 方法来提供字符串 "Hello, CompletableFuture"。然后使用 thenApplyAsync() 方法将字符串转换为大写形式。最后，在 whenComplete() 方法中，针对某个异常展开异常处理器。

• exceptionally() 方法用来处理异步操作执行过程中的异常。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
  if (true) {
    throw new RuntimeException("RuntimeException in CompletableFuture");
  }
  return "Hello, CompletableFuture";
}).exceptionally(ex -> {
  System.out.println("Exception occurred: " + ex.getMessage());
  return "null";
});
String result = future.get();
System.out.println(result);

在这个例子中，如果出现异常就输出异常，并用 "null" 代替原来的字符串。

示例

示例1

本示例中，定义了一个 List 存储 XML 文件的路径，使用 CompletableFuture 和 SAXParser 进行异步解析。

List<String> fileList = Arrays.asList("file1.xml", "file2.xml", "file3.xml", "file4.xml");

List<CompletableFuture<List<String>>> parsers = fileList.stream().map(file -> {
    return CompletableFuture.supplyAsync(() -> {
        SAXParserFactory factory = SAXParserFactory.newInstance();
        SAXParser parser = factory.newSAXParser();
        TestHandler handler = new TestHandler();
        parser.parse(new File(file), handler);
        return handler.getData();
    });
}).collect(Collectors.toList());

CompletableFuture<List<String>> results = CompletableFuture.supplyAsync(() -> {
    return parsers.stream().flatMap(parser -> parser.join().stream()).collect(Collectors.toList());
});

List<String> strings = results.join();
System.out.println(strings);

在这个例子中，首先定义一个文件列表（所需的 XML 文件）。然后通过 stream() 方法将文件列表转换为 CompletableFuture，一次异步解析每一个文件，将解析结果存储在 TestHandler 对象中。最后也是使用 CompletableFuture，将所有文件的解析结果合并在一起。

示例2

本示例中，使用 CompletableFuture 和线程池进行异步运算。

ExecutorService executor = Executors.newFixedThreadPool(3);

CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return 3;
    }, executor)
    .thenApplyAsync(result -> {
        return result * 2;
    }, executor)
    .thenApplyAsync(result -> {
        return result + 1;
    }, executor);

System.out.println(completableFuture.get());
executor.shutdown();

在这个例子中，首先定义了一个大小为 3 的线程池。然后创建了 CompletableFuture 对象，在异步操作中，让当前线程睡眠 1 秒钟，然后返回数字 3。最后，计算数字 3 的加减，输出结果为 7。

总结

CompletableFuture 类是 Java 1.8 版本中新引入的多线程工具类，它具有高度灵活性、易用性和可组合性。通过调用不同的 API 方法，可以使用 CompletableFuture 来完成各种异步操作，包括延迟操作、异步计算、异步消费以及异步组合。本文从多个角度详细讲述了 CompletableFuture 的使用方法，提供了多个示例，希望对您理解和使用 CompletableFuture 有所帮助。