如何用CompletableFuture获取首个符合指定响应的REST调用结果？

尝试用CompletableFuture发起3个REST调用，期望返回首个符合特定响应的结果，但现有代码无法正常工作：即使设置了等待时间，始终返回"future1"，test2和test3存在阻塞问题。曾考虑用CompletableFuture.anyOf，但它仅返回首个执行完成的结果，而非符合指定响应的结果，寻求解决方案。

原测试代码：

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class Testing {
    public static void main(String args[]) throws InterruptedException, ExecutionException {
        CompletableFuture<String> combinedFuture = test("future1", 10000)
                .thenCompose(response1 -> test2()
                        .thenCompose(response2 -> test3()
                                .thenApply(response3 -> {
                                    return combine(response1, response2, response3);
                                })));
        System.out.println(combinedFuture.get());
    }

    private static CompletableFuture<String> test(String str, int i) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(i);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return str;
        });
    }

    private static CompletableFuture<String> test2() {
        return test("future2", 0);
    }

    private static CompletableFuture<String> test3() {
        return test("future3", 0);
    }

    private static String combine(String response1, String response2, String response3) {
        String responseString = null;
        if (response1 != null) {
            return response1;
        } else if (response2 != null) {
            return response2;
        } else if (response3 != null) {
            return response3;
        }
        return responseString;
    }
}

原代码问题分析 #

原代码使用thenCompose链式调用，本质是串行执行三个任务：必须等test("future1",10000)执行完成后才会启动test2，再等test2完成才启动test3。而且combine方法中response1永远不为null，直接返回，完全没机会判断后续结果，这才是问题根源。

正确实现思路 #

1. 并行启动所有任务：三个REST调用要同时发起，而非串行等待。
2. 添加结果过滤逻辑：对每个任务的结果进行判断，只保留符合要求的响应。
3. 获取首个符合条件的结果：将过滤后的任务组合，拿到第一个满足条件的结果，同时可以取消其他未完成的任务以节省资源。

修正后的代码示例 #

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class Testing {
    // 定义判断响应是否符合要求的方法（可根据实际需求修改）
    private static boolean isDesiredResponse(String response) {
        // 示例：假设我们想要"future2"或"future3"作为有效响应
        return "future2".equals(response) || "future3".equals(response);
    }

    public static void main(String args[]) throws InterruptedException, ExecutionException, TimeoutException {
        // 1. 并行启动三个任务
        CompletableFuture<String> future1 = test("future1", 10000);
        CompletableFuture<String> future2 = test2();
        CompletableFuture<String> future3 = test3();

        // 2. 对每个任务添加过滤：只有符合要求的结果才会触发后续流程
        CompletableFuture<String> filtered1 = future1.thenApply(response -> isDesiredResponse(response) ? response : null);
        CompletableFuture<String> filtered2 = future2.thenApply(response -> isDesiredResponse(response) ? response : null);
        CompletableFuture<String> filtered3 = future3.thenApply(response -> isDesiredResponse(response) ? response : null);

        // 3. 组合过滤后的任务，获取第一个非null的结果
        CompletableFuture<String> firstDesired = filtered1.applyToEither(filtered2, r -> r != null ? r : null)
                .applyToEither(filtered3, r -> r != null ? r : null);

        // 4. 等待结果，同时可设置超时；拿到结果后取消其他未完成任务
        String result = firstDesired.get(15, TimeUnit.SECONDS);
        // 取消未完成的任务（可选，避免资源浪费）
        future1.cancel(true);
        future2.cancel(true);
        future3.cancel(true);

        System.out.println("首个符合要求的结果：" + result);
    }

    private static CompletableFuture<String> test(String str, int i) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(i);
            } catch (InterruptedException e) {
                // 任务被取消时抛出异常，无需额外处理
                Thread.currentThread().interrupt();
            }
            return str;
        });
    }

    private static CompletableFuture<String> test2() {
        return test("future2", 0);
    }

    private static CompletableFuture<String> test3() {
        return test("future3", 0);
    }
}

关键代码说明 #

• 并行启动任务：直接创建三个CompletableFuture实例，它们会立即在异步线程中执行，实现并行调用。
• 结果过滤：通过thenApply对每个任务的结果进行判断，不符合要求则返回null，后续组合逻辑会忽略null结果。
• 组合任务：使用applyToEither依次组合过滤后的任务，它会等待两个任务中任意一个完成，返回其中的非null结果，最终得到第一个符合要求的响应。
• 取消任务：拿到结果后调用cancel(true)取消其他未完成的任务，避免不必要的资源消耗（比如已经找到符合要求的结果，就不需要继续等待其他REST调用完成）。

内容的提问来源于stack exchange，提问作者RTN9