CompletableFuture异步编程实战

JDK8 CompletableFuture异步编程实战

目录

• 一、CompletableFuture概述
• 二、创建CompletableFuture
• 三、结果处理与转换
• 四、多任务组合
• 五、异常处理
• 六、线程池配置
• 七、实战案例:电商系统
• 八、实战案例:批量任务处理
• 九、性能优化与最佳实践
• 十、常见问题与陷阱
• 总结

一、CompletableFuture概述

1.1 为什么需要CompletableFuture

在JDK 8之前，Java异步编程主要依赖Future和Callable，但存在诸多不足：

Future的局限性:
+---------------------------+
|  无法手动完成             |
|  无法链式调用             |
|  无法组合多个Future       |
|  无法处理异常             |
|  阻塞式获取结果           |
+---------------------------+

CompletableFuture的优势:
+---------------------------+
|  支持手动完成             |
|  支持链式调用             |
|  多个Future组合          |
|  完善的异常处理           |
|  非阻塞式编程             |
+---------------------------+

package com.example.completablefuture;

import java.util.concurrent.*;

/**
 * Future vs CompletableFuture对比
 */
public class FutureVsCompletableFuture {

    public static void main(String[] args) throws Exception {

        // === 1. 传统Future的问题 ===
        System.out.println(\"=== 传统Future ===\");

        ExecutorService executor = Executors.newFixedThreadPool(2);

        // 提交任务
        Future future = executor.submit(() -> {
            Thread.sleep(1000);
            return \"Hello Future\";
        });

        // 只能阻塞等待结果
        System.out.println(\"阻塞等待结果...\");
        String result = future.get(); // 阻塞
        System.out.println(\"结果: \" + result);

        // === 2. CompletableFuture的优势 ===
        System.out.println(\"n=== CompletableFuture ===\");

        // 异步执行
        CompletableFuture cf = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return \"Hello CompletableFuture\";
        });

        // 非阻塞式处理结果
        cf.thenAccept(res -> System.out.println(\"结果: \" + res));

        // 链式调用
        CompletableFuture chain = CompletableFuture
            .supplyAsync(() -> \"Step 1\")
            .thenApply(s -> s + \" -> Step 2\")
            .thenApply(s -> s + \" -> Step 3\");

        System.out.println(\"链式调用结果: \" + chain.get());

        executor.shutdown();

        // 等待异步任务完成
        Thread.sleep(2000);
    }
}

1.2 CompletableFuture核心方法分类

方法分类:
+------------------+------------------------+--------------------+
|    类型          |       方法             |      说明          |
+------------------+------------------------+--------------------+
| 创建             | supplyAsync            | 有返回值           |
|                  | runAsync               | 无返回值           |
+------------------+------------------------+--------------------+
| 转换             | thenApply              | 同步转换           |
|                  | thenApplyAsync         | 异步转换           |
+------------------+------------------------+--------------------+
| 消费             | thenAccept             | 消费结果           |
|                  | thenRun                | 不接收结果         |
+------------------+------------------------+--------------------+
| 组合             | thenCompose            | 串行组合           |
|                  | thenCombine            | 并行组合           |
|                  | allOf                  | 等待所有完成       |
|                  | anyOf                  | 等待任一完成       |
+------------------+------------------------+--------------------+
| 异常处理         | exceptionally          | 异常处理           |
|                  | handle                 | 正常+异常处理      |
|                  | whenComplete           | 完成时回调         |
+------------------+------------------------+--------------------+

二、创建CompletableFuture

2.1 基本创建方式

package com.example.completablefuture;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * CompletableFuture创建方式
 */
public class CreateCompletableFuture {

    public static void main(String[] args) throws Exception {

        // === 1. supplyAsync - 有返回值的异步任务 ===
        System.out.println(\"=== supplyAsync ===\");

        CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(\"执行异步任务: \" + Thread.currentThread().getName());
            return \"Hello\";
        });

        System.out.println(\"结果: \" + future1.get());

        // === 2. runAsync - 无返回值的异步任务 ===
        System.out.println(\"n=== runAsync ===\");

        CompletableFuture future2 = CompletableFuture.runAsync(() -> {
            System.out.println(\"执行无返回值任务: \" + Thread.currentThread().getName());
        });

        future2.get(); // 等待完成

        // === 3. 使用自定义线程池 ===
        System.out.println(\"n=== 自定义线程池 ===\");

        ExecutorService executor = Executors.newFixedThreadPool(5);

        CompletableFuture future3 = CompletableFuture.supplyAsync(() -> {
            System.out.println(\"自定义线程池: \" + Thread.currentThread().getName());
            return \"Custom Thread Pool\";
        }, executor);

        System.out.println(\"结果: \" + future3.get());

        // === 4. completedFuture - 已完成的Future ===
        System.out.println(\"n=== completedFuture ===\");

        CompletableFuture completed = CompletableFuture.completedFuture(\"立即完成\");
        System.out.println(\"结果: \" + completed.get());

        // === 5. 手动完成 ===
        System.out.println(\"n=== 手动完成 ===\");

        CompletableFuture manual = new CompletableFuture();

        // 在另一个线程中完成
        new Thread(() -> {
            try {
                Thread.sleep(1000);
                manual.complete(\"手动完成的结果\");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        System.out.println(\"等待手动完成...\");
        System.out.println(\"结果: \" + manual.get());

        executor.shutdown();
    }
}

三、结果处理与转换

3.1 thenApply、thenAccept、thenRun

package com.example.completablefuture;

import java.util.concurrent.CompletableFuture;

/**
 * 结果处理与转换
 */
public class ResultProcessing {

    public static void main(String[] args) throws Exception {

        // === 1. thenApply - 转换结果(有返回值) ===
        System.out.println(\"=== thenApply ===\");

        CompletableFuture future1 = CompletableFuture
            .supplyAsync(() -> {
                System.out.println(\"步骤1: 返回10\");
                return 10;
            })
            .thenApply(num -> {
                System.out.println(\"步骤2: \" + num + \" * 2\");
                return num * 2;
            })
            .thenApply(num -> {
                System.out.println(\"步骤3: \" + num + \" + 5\");
                return num + 5;
            });

        System.out.println(\"最终结果: \" + future1.get()); // 25

        // === 2. thenAccept - 消费结果(无返回值) ===
        System.out.println(\"n=== thenAccept ===\");

        CompletableFuture future2 = CompletableFuture
            .supplyAsync(() -> \"Hello World\")
            .thenAccept(str -> {
                System.out.println(\"消费结果: \" + str);
            });

        future2.get();

        // === 3. thenRun - 不关心结果(无返回值) ===
        System.out.println(\"n=== thenRun ===\");

        CompletableFuture future3 = CompletableFuture
            .supplyAsync(() -> \"Some Result\")
            .thenRun(() -> {
                System.out.println(\"任务完成，执行后续操作\");
            });

        future3.get();

        // === 4. 同步 vs 异步 ===
        System.out.println(\"n=== 同步 vs 异步 ===\");

        // thenApply: 同步执行(使用相同线程)
        CompletableFuture.supplyAsync(() -> {
            System.out.println(\"supplyAsync: \" + Thread.currentThread().getName());
            return \"Result\";
        }).thenApply(s -> {
            System.out.println(\"thenApply: \" + Thread.currentThread().getName());
            return s.toUpperCase();
        }).get();

        // thenApplyAsync: 异步执行(使用不同线程)
        CompletableFuture.supplyAsync(() -> {
            System.out.println(\"supplyAsync: \" + Thread.currentThread().getName());
            return \"Result\";
        }).thenApplyAsync(s -> {
            System.out.println(\"thenApplyAsync: \" + Thread.currentThread().getName());
            return s.toUpperCase();
        }).get();

        // === 5. 链式调用示例 ===
        demonstrateChaining();
    }

    /**
     * 链式调用示例
     */
    private static void demonstrateChaining() throws Exception {
        System.out.println(\"n=== 链式调用示例 ===\");

        String result = CompletableFuture
            .supplyAsync(() -> \"hello\")
            .thenApply(String::toUpperCase)
            .thenApply(s -> s + \" WORLD\")
            .thenApply(s -> s + \"!\")
            .get();

        System.out.println(\"链式调用结果: \" + result);
    }
}

四、多任务组合

4.1 thenCompose与thenCombine

package com.example.completablefuture;

import java.util.concurrent.CompletableFuture;

/**
 * 多任务组合
 */
public class TaskCombination {

    public static void main(String[] args) throws Exception {

        // === 1. thenCompose - 串行组合(依赖) ===
        System.out.println(\"=== thenCompose (串行) ===\");

        CompletableFuture compose = CompletableFuture
            .supplyAsync(() -> {
                System.out.println(\"任务1: 查询用户ID\");
                return \"USER_001\";
            })
            .thenCompose(userId -> CompletableFuture.supplyAsync(() -> {
                System.out.println(\"任务2: 根据\" + userId + \"查询用户详情\");
                return \"User{name=\'张三\', id=\'\" + userId + \"\'}\";
            }));

        System.out.println(\"结果: \" + compose.get());

        // === 2. thenCombine - 并行组合(独立) ===
        System.out.println(\"n=== thenCombine (并行) ===\");

        CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            System.out.println(\"任务1: 查询用户信息\");
            return \"User: 张三\";
        });

        CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            System.out.println(\"任务2: 查询订单信息\");
            return \"Order: ORD001\";
        });

        CompletableFuture combined = future1.thenCombine(future2, (user, order) -> {
            System.out.println(\"合并结果\");
            return user + \" | \" + order;
        });

        System.out.println(\"组合结果: \" + combined.get());

        // === 3. allOf - 等待所有任务完成 ===
        System.out.println(\"n=== allOf ===\");

        CompletableFuture task1 = CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            return \"任务1完成\";
        });

        CompletableFuture task2 = CompletableFuture.supplyAsync(() -> {
            sleep(2000);
            return \"任务2完成\";
        });

        CompletableFuture task3 = CompletableFuture.supplyAsync(() -> {
            sleep(1500);
            return \"任务3完成\";
        });

        long start = System.currentTimeMillis();

        CompletableFuture allOfFuture = CompletableFuture.allOf(task1, task2, task3);

        allOfFuture.get(); // 等待所有完成

        long end = System.currentTimeMillis();

        System.out.println(\"所有任务完成，耗时: \" + (end - start) + \"ms\");
        System.out.println(\"结果1: \" + task1.get());
        System.out.println(\"结果2: \" + task2.get());
        System.out.println(\"结果3: \" + task3.get());

        // === 4. anyOf - 等待任一任务完成 ===
        System.out.println(\"n=== anyOf ===\");

        CompletableFuture slow = CompletableFuture.supplyAsync(() -> {
            sleep(3000);
            return \"慢任务\";
        });

        CompletableFuture fast = CompletableFuture.supplyAsync(() -> {
            sleep(1000);
            return \"快任务\";
        });

        CompletableFuture