使用 CompletableFuture.allOf() 对方法进行单元测试

Question

我正在尝试测试一种使用

CompletableFuture.allOf()

的方法。这是我的方法：

    static CompletableFuture<byte[]> anySuccess(List<CompletableFuture<byte[]>> futures) {
        CompletableFuture<byte[]> delegateFuture = new CompletableFuture<>();

        CompletableFuture.allOf(futures.stream().map(s -> s.thenApply(t -> {
                    if (t == null) {
                        // Here we treat null as a failed lookup.
                        // The inner dummy exception serves as a signal for the caller to return null.
                        throw new RuntimeException();
                    }
                    return t;
                }).thenAccept(delegateFuture::complete)).toArray(CompletableFuture<?>[]::new))
                .exceptionally(ex -> {
                    delegateFuture.completeExceptionally(ex);
                    return null;
                });
        return delegateFuture;
    }

CompletableFutures 的输入集可以在不同时间完成/预计完成。这是我的单元测试的样子

    @Test
    void testAnySuccess() {
        final byte[] payload1 = "payload1".getBytes();
        final byte[] payload2 = "payload2".getBytes();

        List<CompletableFuture<byte[]>> futures = Arrays.asList(
                CompletableFuture.supplyAsync(
                        () -> null
                ),
            CompletableFuture.supplyAsync(
                () -> {
                    LockSupport.parkNanos(Duration.ofSeconds(5).toNanos());
                    System.out.println("Wait for 5 seconds");
                    return payload2;
                }),
                CompletableFuture.supplyAsync(
                        () -> {
                            LockSupport.parkNanos(Duration.ofMillis(100).toNanos());
                            System.out.println("Wait for 100ms");
                            return payload1;
                        })
        );
        long start = System.nanoTime();
        CompletableFuture<byte[]> res = anySuccess(futures);
        byte[] actualRes = res.join();
        Assert.assertEquals(actualRes, payload1);

据我了解 -

allOf()

方法不会阻塞 - 但它返回另一个 CompletableFuture，只有当其中的所有 future 都完成时，它才是完整的。

这是我的预期结果 - 返回第一个非空值，或者如果它们全部为空 - 则返回异常。

我很困惑在这种特殊情况下是否需要使用

allOf

-vs-

anyOf

方法。或者我是否需要更改大小写，以便不使用

.exceptionally(...)

方法，并尝试使用

thenAccept(...)

修改它

Answer 1

anyOf

不起作用。它将等待任何未来的完成，无论是成功的还是异常的。这意味着如果最快的未来失败，它也会失败。

我认为只要列表不为空，您当前的代码就可以正常工作。如果是，则不会调用

complete

或

completeExceptionally

，并且

delegateFuture

将永远不会完成。快速检查应该可以解决这个问题。

一旦确保列表不为空，调用

allOf

的结果不会返回，而是在后台保持活动状态，直到所有 future 完成。任何具有

null

结果的未来都将失败，但一旦具有非

null

值，它就会完成

delegateFuture

。由于如果

complete

已经完成，

completeExceptionally

和

delegateFuture

都不会执行任何操作，因此有两种选择：

1. 至少其中一份期货具有非

   null

   价值并将完成

   delegateFuture

   。所有其他具有非

   null

   值的期货将被有效地忽略。如果任何期货由于另一个异常的

   null

   值而失败，则调用

   completeExceptionally

   ，但这不会执行任何操作，因为

   delegateFuture

   已经完成。

2. 所有期货都会因

   null

   值或其他异常而失败。

   exceptionally

   块将调用

   completeExceptionally

   ，这将是

   delegateFuture

   的最终结果。您得到的异常是由第一次失败决定的。

---

未来的 Java 版本将提供更好的解决方案 - StructuredTaskScope，特别是使用

StructuredTaskScope.ShutdownOnSuccess

。一旦找到第一个成功，就会返回，如果没有成功，就会失败。与

allOf

的解决方案不同，一旦发现第一次成功，它就不会继续进行任何不必要的工作。