CompletableFuture方法大全和使用详解（一步到位）

CompletableFuture原理与实践详解

前言

在现代软件开发中，随着系统复杂度和并发需求的增加，异步编程变得尤为重要。Java 8引入的CompletableFuture类，为开发者提供了强大的工具，以简化异步任务的管理和组合。本文将探讨CompletableFuture的使用方法。

一、CompletableFuture的背景与定义

1.1 Future的局限性

在Java 8之前，异步编程主要依赖于Future接口。然而，Future存在以下局限性：

• 阻塞获取结果：调用get()方法时，线程会被阻塞，直到结果可用。这会导致线程资源的浪费，影响系统的并发性能。
• 缺乏回调机制：无法在任务完成后自动执行后续操作，需要手动检查任务状态，代码复杂且容易出错。
• 难以组合多个任务：对于需要等待多个异步任务完成或组合多个任务结果的场景，Future提供的支持不够完善。

1.2 CompletableFuture的优势

为了解决这些问题，Java 8引入了CompletableFuture，它实现了CompletionStage和Future接口，提供了更丰富的API，支持异步回调和任务组合。CompletableFuture具有以下优势：

• 非阻塞获取结果：通过回调函数，在任务完成后自动触发后续操作，无需阻塞等待。
• 丰富的任务组合：提供了多种方法来组合多个异步任务，如thenApply、thenCombine、thenCompose,allOf、anyOf等。
• 完善的异常处理：内置了异常处理机制，如exceptionally、handle等方法，方便地处理异步任务中的异常情况。
• 灵活的执行控制：支持自定义线程池，可以更精确地控制任务的执行环境。

二、CompletableFuture的核心方法

2.1 创建异步任务

CompletableFuture提供了两种方法来创建异步任务：

2.1.1 supplyAsync – 创建有返回值的异步任务

// 使用默认线程池（ForkJoinPool.commonPool()）
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    // 执行异步操作
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return \"异步任务结果\";
});

// 使用自定义线程池
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
    return \"使用自定义线程池的结果\";
}, executor);

2.1.2 runAsync – 创建无返回值的异步任务

// 使用默认线程池
CompletableFuture future = CompletableFuture.runAsync(() -> {
    // 执行异步操作，不返回结果
    System.out.println(\"执行异步任务\");
});

// 使用自定义线程池
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture future2 = CompletableFuture.runAsync(() -> {
    System.out.println(\"使用自定义线程池执行\");
}, executor);

注意：在实际生产环境中，建议使用自定义线程池，因为默认的ForkJoinPool.commonPool()在高并发场景下可能无法满足需求。

2.2 异步回调处理

CompletableFuture提供了丰富的回调方法，以便在任务完成后执行特定操作：

2.2.1 thenApply – 对结果进行转换

thenApply方法接收一个函数，将前一个阶段的结果转换为新的结果。它有两个版本：

• thenApply：在完成当前任务的同一个线程中执行
• thenApplyAsync：在新的线程中执行（使用默认线程池或自定义线程池）

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

// 同步版本
CompletableFuture result1 = future.thenApply(s -> {
    return s + \" World\";
});

// 异步版本（使用默认线程池）
CompletableFuture result2 = future.thenApplyAsync(s -> {
    return s + \" World\";
});

// 异步版本（使用自定义线程池）
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture result3 = future.thenApplyAsync(s -> {
    return s + \" World\";
}, executor);

2.2.2 thenAccept – 消费结果，无返回值

thenAccept方法接收一个消费者函数，用于消费前一个阶段的结果，但不返回新值。

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    return \"Hello World\";
});

// 同步版本
future.thenAccept(result -> {
    System.out.println(\"消费结果: \" + result);
});

// 异步版本
future.thenAcceptAsync(result -> {
    System.out.println(\"消费结果: \" + result);
});

// 异步版本（使用自定义线程池）
ExecutorService executor = Executors.newFixedThreadPool(10);
future.thenAcceptAsync(result ->{
    System.out.println(\"消费结果：\" + result);
},executor);

2.2.3 thenRun – 不关心结果，执行后续操作

thenRun方法接收一个Runnable，在前一个阶段完成后执行，但不关心前一个阶段的结果。

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    return \"Hello World\";
});

// 同步版本
future.thenRun(() -> {
    System.out.println(\"任务完成后执行清理操作\");
});

// 异步版本
ExecutorService executor = Executors.newFixedThreadPool(10);
future.thenRunAsync(() -> {
    System.out.println(\"任务完成后执行清理操作\");
},executor);

2.2.4 thenCompose – 将两个异步操作串联

thenCompose方法用于将两个异步操作串联起来，前一个操作的结果作为后一个操作的输入。这类似于流式处理。(与下面的thenCombine区别在于：这个方法主要用于异步任务的串行编排。简单来说，thenCompose() 用于将一个异步任务的结果作为输入，继续发起一个新的异步任务，形成一条异步流水线。thenCompose() 把两个 Future 合并成一个连续的 Future，而不是thenCombine()形成的嵌套 CompletableFuture)

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

CompletableFuture future2 = future1.thenCompose(result -> {
    return CompletableFuture.supplyAsync(() -> {
        return result + \" World\";
    });
});

// 结果：future2 的值为 \"Hello World\"

2.2.5 thenCombine – 合并两个并行任务的结果，返回参数

thenCombine方法用于合并两个并行的异步任务的结果。

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

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

CompletableFuture combined = future1.thenCombine(future2, (result1, result2) -> {
    return result1 + \" \" + result2;
});

// 结果：combined 的值为 \"Hello World\"

2.2.6thenAcceptBoth – 方法用于合并两个并行任务的结果，不返回参数

  CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + \" cf1 do something....\");
            return 1;
        });
 
        CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + \" cf2 do something....\");
            return 2;
        });
        
        CompletableFuture cf3 = cf1.thenAcceptBoth(cf2, (a, b) -> {
            System.out.println(Thread.currentThread() + \" cf3 do something....\");
            System.out.println(a + b);
        });

2.2.7runAfterBoth

runAfterBoth没有入参，也没有返回值。两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。

     CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + \" cf1 do something....\");
            return 1;
        });
 
        CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + \" cf2 do something....\");
            return 2;
        });
 
        CompletableFuture cf3 = cf1.runAfterBoth(cf2, () -> {
            System.out.println(Thread.currentThread() + \" cf3 do something....\");
        });

2.3 组合多个异步任务

2.3.1 allOf – 等待所有任务完成

allOf方法创建一个新的CompletableFuture，当所有给定的CompletableFuture都完成时，它才会完成。

CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
    return \"结果1\";
});

CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
    return \"结果2\";
});

CompletableFuture future3 = CompletableFuture.supplyAsync(() -> {
    return \"结果3\";
});

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

// 等待所有任务完成
allFutures.thenRun(() -> {
    System.out.println(\"所有任务已完成\");
    // 获取各个任务的结果
    String result1 = future1.join();
    String result2 = future2.join();
    String result3 = future3.join();
    System.out.println(\"结果1: \" + result1);
    System.out.println(\"结果2: \" + result2);
    System.out.println(\"结果3: \" + result3);
});

2.3.2 anyOf – 任意一个任务完成即可继续

anyOf方法创建一个新的CompletableFuture，当任意一个给定的CompletableFuture完成时，它就会完成。

CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return \"结果1\";
});

CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    return \"结果2\";
});

CompletableFuture