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

Java CompletableFuture组合操作中的依赖管理与优化

2021-01-241.5k 阅读

Java CompletableFuture 组合操作中的依赖管理与优化

CompletableFuture 基础回顾

在深入探讨依赖管理与优化之前，我们先来回顾一下 CompletableFuture 的基础知识。CompletableFuture 是 Java 8 引入的一个强大工具，用于异步编程。它实现了 Future 接口，同时提供了更为丰富的异步操作方法。

例如，创建一个简单的 CompletableFuture：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 模拟耗时操作
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Hello, CompletableFuture!";
});

future.thenAccept(System.out::println);

在上述代码中，supplyAsync 方法接受一个 Supplier 作为参数，并异步执行其中的任务。任务完成后，thenAccept 方法会处理返回的结果。

依赖关系的基本概念

在 CompletableFuture 的组合操作中，依赖关系是指一个 CompletableFuture 的执行依赖于另一个或多个 CompletableFuture 的完成。例如，假设有两个任务 A 和 B，任务 B 需要在任务 A 完成后才能开始执行，那么任务 B 就依赖于任务 A。

顺序依赖管理

thenApply 方法

thenApply 方法用于在一个 CompletableFuture 完成后，对其结果进行转换并返回一个新的 CompletableFuture。

CompletableFuture.supplyAsync(() -> "Hello")
       .thenApply(s -> s + ", World")
       .thenAccept(System.out::println);

在这个例子中，第一个 CompletableFuture 生成字符串 "Hello"，thenApply 方法接受这个结果并添加 ", World"，最后通过 thenAccept 打印出来。

thenCompose 方法

当 thenApply 返回的结果本身又是一个 CompletableFuture 时，就需要使用 thenCompose 方法来正确处理依赖关系。

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = future1.thenCompose(s -> {
    return CompletableFuture.supplyAsync(() -> s + ", World");
});
future2.thenAccept(System.out::println);

thenCompose 方法会将内层的 CompletableFuture 展开，使得整个操作链能够正确执行。

并行依赖管理

allOf 方法

allOf 方法用于等待一组 CompletableFuture 全部完成。它返回一个新的 CompletableFuture，当所有输入的 CompletableFuture 都完成时，这个新的 CompletableFuture 才会完成。

CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "A";
});
CompletableFuture<String> futureB = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "B";
});

CompletableFuture<Void> allFuture = CompletableFuture.allOf(futureA, futureB);
allFuture.thenRun(() -> {
    try {
        System.out.println("All futures completed. Result of A: " + futureA.get());
        System.out.println("Result of B: " + futureB.get());
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
});

在上述代码中，allOf 方法返回的 allFuture 会在 futureA 和 futureB 都完成后才完成，然后通过 thenRun 方法获取并打印两个任务的结果。

anyOf 方法

anyOf 方法则是只要一组 CompletableFuture 中有一个完成，它返回的 CompletableFuture 就会完成。

CompletableFuture<String> futureC = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "C";
});
CompletableFuture<String> futureD = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(1500);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "D";
});

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(futureC, futureD);
anyFuture.thenAccept(result -> System.out.println("First completed result: " + result));

在这个例子中，futureD 会先完成，所以 anyFuture 会在 futureD 完成后就完成，并打印出 futureD 的结果。

依赖管理中的异常处理

exceptionally 方法

在 CompletableFuture 的依赖链中，异常处理至关重要。exceptionally 方法用于处理 CompletableFuture 执行过程中抛出的异常。

CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("Simulated exception");
    }
    return "Normal result";
})
       .exceptionally(ex -> {
            System.out.println("Caught exception: " + ex.getMessage());
            return "Default value";
        })
       .thenAccept(System.out::println);

在上述代码中，如果 supplyAsync 中的任务抛出异常，exceptionally 方法会捕获并处理异常，返回一个默认值。

handle 方法

handle 方法可以同时处理正常结果和异常情况。

CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("Simulated exception");
    }
    return "Normal result";
})
       .handle((result, ex) -> {
            if (ex != null) {
                System.out.println("Caught exception: " + ex.getMessage());
                return "Default value";
            }
            return result;
        })
       .thenAccept(System.out::println);

handle 方法接受一个 BiFunction，第一个参数是正常结果，第二个参数是异常。根据是否有异常，返回相应的处理结果。

依赖管理的优化策略

减少不必要的异步操作

在使用 CompletableFuture 时，要避免不必要的异步操作。如果一个操作非常短，同步执行可能比异步执行更高效，因为异步操作会引入线程池调度等开销。

// 同步操作
String result = performShortOperation();
System.out.println(result);

// 不必要的异步操作
CompletableFuture.supplyAsync(() -> performShortOperation())
       .thenAccept(System.out::println);

在上述代码中，performShortOperation 如果是一个非常短的操作，直接同步执行会更好。

合理使用线程池

CompletableFuture 默认使用 ForkJoinPool.commonPool() 来执行异步任务。对于一些需要特定线程池配置的场景，应该创建并使用自定义线程池。

ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture.supplyAsync(() -> "Task result", executor)
       .thenAccept(System.out::println);
executor.shutdown();

在这个例子中，我们创建了一个固定大小为 10 的线程池，并使用它来执行异步任务。

优化依赖链长度

尽量缩短 CompletableFuture 的依赖链长度。过长的依赖链可能导致性能问题和调试困难。如果可能，将复杂的依赖链拆分成多个较小的部分。

// 过长的依赖链
CompletableFuture.supplyAsync(() -> step1())
       .thenApply(result1 -> step2(result1))
       .thenApply(result2 -> step3(result2))
       .thenApply(result3 -> step4(result3))
       .thenAccept(System.out::println);

// 拆分后的依赖链
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> step1());
CompletableFuture<String> future2 = future1.thenApply(result1 -> step2(result1));
CompletableFuture<String> future3 = future2.thenApply(result2 -> step3(result2));
CompletableFuture<String> future4 = future3.thenApply(result3 -> step4(result3));
future4.thenAccept(System.out::println);

虽然拆分后的代码看起来更繁琐，但在维护和性能优化方面可能更有优势。

利用缓存

如果 CompletableFuture 执行的任务结果是可缓存的，可以使用缓存来避免重复计算。

Map<String, CompletableFuture<String>> cache = new HashMap<>();
CompletableFuture<String> getResultFromCacheOrCompute(String key) {
    return cache.computeIfAbsent(key, k -> CompletableFuture.supplyAsync(() -> {
        // 模拟耗时计算
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Computed result for " + key;
    }));
}

在上述代码中，cache 用于缓存 CompletableFuture 的计算结果，下次再请求相同的 key 时，直接从缓存中获取 CompletableFuture，避免重复计算。

高级依赖管理技巧

动态依赖构建

在实际应用中，依赖关系可能不是静态确定的，而是需要根据运行时的条件动态构建。

List<CompletableFuture<String>> futures = new ArrayList<>();
if (condition1) {
    futures.add(CompletableFuture.supplyAsync(() -> "Result from condition1"));
}
if (condition2) {
    futures.add(CompletableFuture.supplyAsync(() -> "Result from condition2"));
}

CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
allFutures.thenRun(() -> {
    futures.forEach(future -> {
        try {
            System.out.println(future.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    });
});

在上述代码中，根据 condition1 和 condition2 的值动态决定添加哪些 CompletableFuture 到 futures 列表中，然后使用 allOf 方法等待所有动态添加的任务完成。

依赖的优先级管理

有时候，不同的 CompletableFuture 任务可能具有不同的优先级。可以通过自定义线程池和任务调度策略来实现优先级管理。

PriorityBlockingQueue<Runnable> taskQueue = new PriorityBlockingQueue<>(10, Comparator.comparingInt((Runnable r) -> {
    if (r instanceof PrioritizedTask) {
        return ((PrioritizedTask) r).getPriority();
    }
    return 0;
}));
ExecutorService priorityExecutor = new ThreadPoolExecutor(
        5,
        10,
        10L,
        TimeUnit.SECONDS,
        taskQueue);

class PrioritizedTask implements Runnable {
    private final int priority;
    private final Runnable task;

    public PrioritizedTask(int priority, Runnable task) {
        this.priority = priority;
        this.task = task;
    }

    public int getPriority() {
        return priority;
    }

    @Override
    public void run() {
        task.run();
    }
}

CompletableFuture.supplyAsync(() -> "High priority task", new PrioritizedTask(1, () -> {}), priorityExecutor)
       .thenAccept(System.out::println);
CompletableFuture.supplyAsync(() -> "Low priority task", new PrioritizedTask(2, () -> {}), priorityExecutor)
       .thenAccept(System.out::println);

在上述代码中，我们创建了一个带有优先级的任务队列和线程池。PrioritizedTask 类实现了 Runnable 接口，并带有优先级属性。通过这种方式，高优先级的任务会优先被执行。

依赖的重试机制

在 CompletableFuture 的依赖操作中，有时候任务可能会因为临时的网络问题或其他原因失败，需要进行重试。

CompletableFuture<String> retryableFuture = CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) {
        throw new RuntimeException("Simulated failure");
    }
    return "Success";
});

int maxRetries = 3;
CompletableFuture<String> finalFuture = retryableFuture;
for (int i = 0; i < maxRetries; i++) {
    final int attempt = i;
    finalFuture = finalFuture.exceptionally(ex -> {
        System.out.println("Attempt " + (attempt + 1) + " failed: " + ex.getMessage());
        if (attempt < maxRetries - 1) {
            return CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (Math.random() > 0.5) {
                    throw new RuntimeException("Simulated failure");
                }
                return "Success";
            });
        } else {
            System.out.println("Max retries reached. Giving up.");
            return "Failed after " + maxRetries + " attempts";
        }
    });
}

finalFuture.thenAccept(System.out::println);

在上述代码中，retryableFuture 模拟了一个可能失败的任务。通过 exceptionally 方法，在任务失败时进行重试，最多重试 maxRetries 次。

性能分析与监控

使用 Java 自带工具

Java 提供了一些工具来分析和监控 CompletableFuture 的性能。例如，使用 jconsole 可以查看线程池的使用情况，包括线程池的活跃线程数、任务队列大小等。

  1. 启动 jconsole：在命令行中输入 jconsole 并回车。
  2. 选择要监控的 Java 进程：jconsole 会列出当前运行的 Java 进程，选择包含 CompletableFuture 代码的进程。
  3. 在 jconsole 中查看线程池相关信息：在 MBeans 标签页中，找到 java.util.concurrent 相关的 MBean，可以查看线程池的各种指标。

自定义性能监控

除了使用 Java 自带工具，还可以通过自定义代码来监控 CompletableFuture 的性能。例如，记录任务开始和结束时间，计算任务执行时间。

long startTime = System.currentTimeMillis();
CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "Task result";
})
       .thenAccept(result -> {
            long endTime = System.currentTimeMillis();
            System.out.println("Task execution time: " + (endTime - startTime) + " ms");
        });

在上述代码中，通过记录任务开始和结束的时间戳，计算出任务的执行时间并打印出来。

与其他异步框架的比较

与 Guava 的 ListenableFuture 比较

Guava 的 ListenableFuture 是 Java 8 之前常用的异步编程工具。与 CompletableFuture 相比，ListenableFuture 功能相对较少，例如没有提供像 thenApply、thenCompose 这样方便的链式操作方法。而且，ListenableFuture 对异常处理的支持也不如 CompletableFuture 强大。

与 RxJava 比较

RxJava 是一个基于观察者模式的异步编程框架，功能非常强大。与 CompletableFuture 相比，RxJava 更侧重于事件流的处理，提供了丰富的操作符来处理异步数据流。然而，RxJava 的学习曲线相对较陡，对于简单的异步任务，使用 CompletableFuture 可能更加简洁明了。

实际应用场景

微服务调用

在微服务架构中，经常需要调用多个微服务并处理它们的返回结果。CompletableFuture 的依赖管理功能可以方便地处理微服务调用之间的依赖关系。例如，一个微服务可能需要先调用用户服务获取用户信息，然后根据用户信息调用订单服务获取订单列表。

CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> userService.getUserById(1));
CompletableFuture<List<Order>> orderFuture = userFuture.thenCompose(user -> CompletableFuture.supplyAsync(() -> orderService.getOrdersByUser(user)));
orderFuture.thenAccept(orders -> System.out.println("Orders for user: " + orders));

大数据处理

在大数据处理中，可能需要并行处理多个数据块，然后合并结果。CompletableFuture 的 allOf 方法可以方便地实现这种并行依赖管理。

List<CompletableFuture<DataChunkResult>> chunkFutures = new ArrayList<>();
for (DataChunk chunk : dataChunks) {
    chunkFutures.add(CompletableFuture.supplyAsync(() -> processDataChunk(chunk)));
}

CompletableFuture<Void> allChunkFutures = CompletableFuture.allOf(chunkFutures.toArray(new CompletableFuture[0]));
allChunkFutures.thenRun(() -> {
    List<DataChunkResult> results = chunkFutures.stream()
           .map(future -> {
                try {
                    return future.get();
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                    return null;
                }
            })
           .collect(Collectors.toList());
    DataSummary summary = summarizeResults(results);
    System.out.println("Data summary: " + summary);
});

实时数据分析

在实时数据分析场景中，可能需要根据不同的数据源进行实时计算。CompletableFuture 的动态依赖构建和优先级管理功能可以根据实时数据的特点进行灵活处理。例如，对于重要的数据源数据优先处理，对于一些次要数据源可以稍后处理或根据资源情况进行调度。

总结

通过深入理解 CompletableFuture 的依赖管理和优化策略，我们可以在异步编程中更高效地利用系统资源，提高程序的性能和可靠性。从基本的顺序和并行依赖管理，到异常处理、优化策略以及高级技巧，每个方面都对构建健壮的异步应用至关重要。同时，通过性能分析与监控以及与其他异步框架的比较，我们可以更好地选择适合具体场景的技术方案。在实际应用场景中，如微服务调用、大数据处理和实时数据分析，CompletableFuture 都展现出了强大的功能和灵活性。希望通过本文的介绍，读者能够更加熟练地运用 CompletableFuture 进行高效的异步编程。