java - 返回在响应的特定条件下首先执行的 future

Question

我正在尝试使用 completablefutures 进行 3 个休息调用，并返回第一个与特定响应匹配的调用。下面是我为其编写的示例测试代码(减去其余调用)，但这似乎不起作用。即使有等待时间，我总是看到“future1”被返回，这意味着 test2 和 test3 正在阻塞。我如何实现要求？

我想过使用 CompletableFuture.anyOf 但它只返回第一个执行的 future 的结果。不是第一个与指定响应匹配的响应。请指教

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;
    }

}

Answer 1

您可以将 future 相互竞争，并将完成任务委托(delegate)给另一个 future :

static <T> CompletableFuture<T> first(Stream<CompletableFuture<T>> futures) {
    var delegate = new CompletableFuture<T>();

    runAsync(() ->
         futures.forEach(future ->
             future.handle(
                 (value, error) -> {
                     if (value == null) {
                         return delegate.completeExceptionally(error);
                     } else {
                         return delegate.complete(value);
                     }
                 })));

    return delegate;
}

每当 futures 参数中传递的第一个函数完成时，first 返回的 future 就会完成(成功或有错误)。

现在

CompletableFuture<String>  combinedFuture =
    first(Stream.of(test("future1", 10000), test2(), test3()));

System.out.println(combinedFuture.get());

打印“future2”或“future3”，具体取决于哪个先完成。