CompletableFuture怎么使用

目录

1. 引言
2. CompletableFuture简介
3. 创建CompletableFuture
4. 异步任务处理
5. 任务链式调用
6. 异常处理
7. 组合多个CompletableFuture
8. 超时处理
9. CompletableFuture的线程池
10. CompletableFuture的常见用例
11. CompletableFuture的局限性
12. 总结

引言

在现代软件开发中，异步编程变得越来越重要。随着多核处理器的普及，开发者需要编写能够充分利用硬件资源的并发程序。Java 8引入了CompletableFuture，它为异步编程提供了强大的支持。本文将详细介绍CompletableFuture的使用方法，帮助开发者更好地理解和应用这一工具。

CompletableFuture简介

CompletableFuture是Java 8引入的一个类，它实现了Future接口，并提供了更丰富的功能。CompletableFuture可以用于表示一个异步计算的结果，并且支持链式调用、组合多个异步任务、异常处理等高级功能。

与传统的Future相比，CompletableFuture具有以下优势： - 链式调用：可以将多个异步任务串联起来，形成一个任务链。 - 异常处理：可以方便地处理异步任务中的异常。 - 组合多个任务：可以将多个异步任务组合在一起，等待所有任务完成或任意一个任务完成。 - 超时处理：可以为异步任务设置超时时间，避免任务长时间阻塞。

创建CompletableFuture

1. 使用CompletableFuture.runAsync()

runAsync()方法用于执行一个没有返回值的异步任务。它接受一个Runnable对象作为参数，并返回一个CompletableFuture<Void>。

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // 异步任务
    System.out.println("Task is running asynchronously");
});

future.join(); // 等待任务完成

2. 使用CompletableFuture.supplyAsync()

supplyAsync()方法用于执行一个有返回值的异步任务。它接受一个Supplier<T>对象作为参数，并返回一个CompletableFuture<T>。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 异步任务
    return "Task result";
});

String result = future.join(); // 获取任务结果
System.out.println(result);

3. 使用CompletableFuture.completedFuture()

completedFuture()方法用于创建一个已经完成的CompletableFuture，并返回指定的结果。

CompletableFuture<String> future = CompletableFuture.completedFuture("Completed result");

String result = future.join(); // 获取任务结果
System.out.println(result);

异步任务处理

1. 使用thenApply()

thenApply()方法用于在异步任务完成后对结果进行处理。它接受一个Function<T, U>对象作为参数，并返回一个新的CompletableFuture<U>。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<String> resultFuture = future.thenApply(s -> s + " World");

String result = resultFuture.join(); // 获取处理后的结果
System.out.println(result); // 输出 "Hello World"

2. 使用thenAccept()

thenAccept()方法用于在异步任务完成后对结果进行消费。它接受一个Consumer<T>对象作为参数，并返回一个CompletableFuture<Void>。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<Void> resultFuture = future.thenAccept(s -> {
    System.out.println(s + " World");
});

resultFuture.join(); // 等待任务完成

3. 使用thenRun()

thenRun()方法用于在异步任务完成后执行一个没有参数和返回值的操作。它接受一个Runnable对象作为参数，并返回一个CompletableFuture<Void>。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<Void> resultFuture = future.thenRun(() -> {
    System.out.println("Task completed");
});

resultFuture.join(); // 等待任务完成

任务链式调用

CompletableFuture支持链式调用，可以将多个异步任务串联起来，形成一个任务链。每个任务的输出可以作为下一个任务的输入。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenApply(s -> s + " World")
  .thenApply(s -> s + "!");

String result = future.join(); // 获取最终结果
System.out.println(result); // 输出 "Hello World!"

异常处理

1. 使用exceptionally()

exceptionally()方法用于处理异步任务中的异常。它接受一个Function<Throwable, T>对象作为参数，并返回一个新的CompletableFuture<T>。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task failed");
});

CompletableFuture<String> resultFuture = future.exceptionally(ex -> {
    System.out.println("Exception: " + ex.getMessage());
    return "Fallback result";
});

String result = resultFuture.join(); // 获取处理后的结果
System.out.println(result); // 输出 "Fallback result"

2. 使用handle()

handle()方法用于在异步任务完成后对结果和异常进行处理。它接受一个BiFunction<T, Throwable, U>对象作为参数，并返回一个新的CompletableFuture<U>。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task failed");
});

CompletableFuture<String> resultFuture = future.handle((result, ex) -> {
    if (ex != null) {
        System.out.println("Exception: " + ex.getMessage());
        return "Fallback result";
    }
    return result;
});

String result = resultFuture.join(); // 获取处理后的结果
System.out.println(result); // 输出 "Fallback result"

组合多个CompletableFuture

1. 使用thenCompose()

thenCompose()方法用于将一个CompletableFuture的结果作为另一个CompletableFuture的输入。它接受一个Function<T, CompletableFuture<U>>对象作为参数，并返回一个新的CompletableFuture<U>。

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<String> future2 = future1.thenCompose(s -> {
    return CompletableFuture.supplyAsync(() -> s + " World");
});

String result = future2.join(); // 获取最终结果
System.out.println(result); // 输出 "Hello World"

2. 使用thenCombine()

thenCombine()方法用于将两个CompletableFuture的结果进行组合。它接受一个CompletableFuture<U>和一个BiFunction<T, U, V>对象作为参数，并返回一个新的CompletableFuture<V>。

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return "World";
});

CompletableFuture<String> resultFuture = future1.thenCombine(future2, (s1, s2) -> {
    return s1 + " " + s2;
});

String result = resultFuture.join(); // 获取最终结果
System.out.println(result); // 输出 "Hello World"

3. 使用allOf()

allOf()方法用于等待多个CompletableFuture全部完成。它接受一个CompletableFuture<?>...数组作为参数，并返回一个新的CompletableFuture<Void>。

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return "World";
});

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

allFutures.join(); // 等待所有任务完成
System.out.println("All tasks completed");

4. 使用anyOf()

anyOf()方法用于等待多个CompletableFuture中的任意一个完成。它接受一个CompletableFuture<?>...数组作为参数，并返回一个新的CompletableFuture<Object>。

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Hello";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return "World";
});

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2);

Object result = anyFuture.join(); // 获取任意一个任务的结果
System.out.println(result); // 输出 "World"

超时处理

1. 使用orTimeout()

orTimeout()方法用于为CompletableFuture设置超时时间。如果任务在指定时间内未完成，CompletableFuture将被异常终止。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Hello";
}).orTimeout(1, TimeUnit.SECONDS);

try {
    String result = future.join(); // 获取任务结果
    System.out.println(result);
} catch (CompletionException ex) {
    System.out.println("Task timed out: " + ex.getMessage());
}

2. 使用completeOnTimeout()

completeOnTimeout()方法用于为CompletableFuture设置超时时间。如果任务在指定时间内未完成，CompletableFuture将被设置为指定的默认值。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Hello";
}).completeOnTimeout("Timeout result", 1, TimeUnit.SECONDS);

String result = future.join(); // 获取任务结果
System.out.println(result); // 输出 "Timeout result"

CompletableFuture的线程池

默认情况下，CompletableFuture使用ForkJoinPool.commonPool()作为线程池。如果需要自定义线程池，可以通过runAsync()和supplyAsync()方法的第二个参数指定。

ExecutorService executor = Executors.newFixedThreadPool(10);

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
}, executor);

String result = future.join(); // 获取任务结果
System.out.println(result);

executor.shutdown(); // 关闭线程池

CompletableFuture的常见用例

1. 并行执行多个任务

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "Task 1";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return "Task 2";
});

CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    return "Task 3";
});

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

allFutures.join(); // 等待所有任务完成
System.out.println("All tasks completed");

2. 任务依赖

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "Hello";
});

CompletableFuture<String> future2 = future1.thenApply(s -> s + " World");

CompletableFuture<String> future3 = future2.thenApply(s -> s + "!");

String result = future3.join(); // 获取最终结果
System.out.println(result); // 输出 "Hello World!"

3. 异常处理

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("Task failed");
});

CompletableFuture<String> resultFuture = future.exceptionally(ex -> {
    System.out.println("Exception: " + ex.getMessage());
    return "Fallback result";
});

String result = resultFuture.join(); // 获取处理后的结果
System.out.println(result); // 输出 "Fallback result"

CompletableFuture的局限性

尽管CompletableFuture提供了强大的异步编程支持，但它也有一些局限性： - 复杂性：CompletableFuture的API较为复杂，初学者可能需要花费一些时间来掌握。 - 调试困难：由于异步任务的执行顺序不确定，调试异步程序可能会比较困难。 - 线程池管理：默认使用ForkJoinPool.commonPool()，在某些场景下可能需要自定义线程池。

总结

CompletableFuture是Java 8引入的一个强大的异步编程工具，它提供了丰富的功能来支持异步任务的创建、链式调用、异常处理、任务组合等。通过合理使用CompletableFuture，开发者可以编写出高效、可维护的并发程序。然而，CompletableFuture也有一些局限性，开发者在使用时需要注意这些问题。

希望本文能够帮助读者更好地理解和应用CompletableFuture，在实际开发中发挥其强大的功能。