尝试等待 CompletableFuture

Question

有没有办法尝试等待CompletableFuture 一段时间，然后返回不同的结果而不在超时后取消未来？

我有一个服务（我们称它为expensiveService），它跑去做自己的事。它返回一个结果：

enum Result {
    COMPLETED,
    PROCESSING,
    FAILED
}

我愿意 [阻止并] 等待它一小段时间（比如说 2 秒）。如果它没有完成，我想返回不同的结果，但我希望服务继续做自己的事情。然后询问服务是否完成（例如通过 websockets 或其他）将是客户的工作。

即我们有以下几种情况：

• expensiveService.processAndGet() 需要 1 秒并完成它的未来。它返回COMPLETED。
• expensiveService.processAndGet() 1 秒后失败。它返回FAILED。
• expensiveService.processAndGet() 需要 5 秒并完成它的未来。它返回PROCESSING。如果我们向其他服务询问结果，我们会得到COMPLETED。
• expensiveService.processAndGet() 5 秒后失败。它返回PROCESSING。如果我们向其他服务询问结果，我们会得到FAILED。

在这种特定情况下，我们实际上需要在超时时获取当前结果对象，从而导致以下额外的边缘情况。这会导致以下建议的解决方案出现一些问题：

• expensiveService.processAndGet() 需要 2.01 秒并完成它的未来。它返回PROCESSING 或COMPLETED。

我也在使用 Vavr，并愿意接受使用 Vavr 的Future 的建议。

我们创建了三种可能的解决方案，它们都有各自的优点和缺点：

#1 等待另一个未来

CompletableFuture<Result> f = expensiveService.processAndGet();
return f.applyToEither(Future.of(() -> {
            Thread.sleep(2000);
            return null;
        }).map(v -> resultService.get(processId)).toCompletableFuture(),
        Function.identity());

问题

1. 总是调用第二个resultService。
2. 我们占用整个线程 2 秒。

#1a 等待另一个 Future 检查第一个 Future

CompletableFuture<Result> f = expensiveService.processAndGet();
return f.applyToEither(Future.of(() -> {
            int attempts = 0;
            int timeout = 20;
            while (!f.isDone() && attempts * timeout < 2000) {
                Thread.sleep(timeout);
                attempts++;
            }
            return null;
        }).map(v -> resultService.get(processId)).toCompletableFuture(),
        Function.identity());

问题

1. 第二个resultService 仍然被调用。
2. 我们需要将第一个 Future 传递给第二个，这不是很干净。

#2 Object.notify

Object monitor = new Object();
CompletableFuture<Upload> process = expensiveService.processAndGet();
synchronized (monitor) {
    process.whenComplete((r, e) -> {
        synchronized (monitor) {
            monitor.notifyAll();
        }
    });
    try {
        int attempts = 0;
        int timeout = 20;
        while (!process.isDone() && attempts * timeout < 2000) {
            monitor.wait(timeout);
            attempts++;
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}
if (process.isDone()) {
    return process.toCompletableFuture();
} else {
    return CompletableFuture.completedFuture(resultService.get(processId));
}

问题

1. 复杂的代码（可能存在错误，可读性差）。

#3 Vavr 的Future.await

return Future.of(() -> expensiveService.processAndGet()
        .await(2, TimeUnit.SECONDS)
        .recoverWith(e -> {
            if (e instanceof TimeoutException) {
                return Future.successful(resultService.get(processId));
            } else {
                return Future.failed(e);
            }
        })
        .toCompletableFuture();

问题

1. 需要 Future in a Future 以避免 await 取消内部 Future。
2. 将第一个 Future 移入第二个会破坏依赖于 ThreadLocals 的 [遗留] 代码。
3. recoverWith 和捕捉 TimeoutException 并不是那么优雅。

#4 CompletableFuture.orTimeout

return expensiveService.processAndGet()
        .orTimeout(2, TimeUnit.SECONDS)
        .<CompletableFuture<Upload>>handle((u, e) -> {
            if (u != null) {
                return CompletableFuture.completedFuture(u);
            } else if (e instanceof TimeoutException) {
                return CompletableFuture.completedFuture(resultService.get(processId));
            } else {
                return CompletableFuture.failedFuture(e);
            }
        })
        .thenCompose(Function.identity());

问题

1. 虽然在我的情况下，processAndGet 未来没有被取消，但根据文档，它应该是。
2. 异常处理不好。

#5 CompletableFuture.completeOnTimeout

return expensiveService.processAndGet()
        .completeOnTimeout(null, 2, TimeUnit.SECONDS)
        .thenApply(u -> {
            if (u == null) {
                return resultService.get(processId);
            } else {
                return u;
            }
        });

问题

1. 虽然在我的情况下，processAndGet 未来尚未完成，但根据文档，它应该是。
2. 如果processAndGet 想将null 作为不同的状态返回怎么办？

所有这些解决方案都有缺点并且需要额外的代码，但这感觉就像是 CompletableFuture 或 Vavr 的开箱即用的 Future 应该支持的东西。有没有更好的方法来做到这一点？

Answer 1

首先值得指出的是，CompletableFuture 是如何工作的（或者为什么会这样命名）：

CompletableFuture<?> f = CompletableFuture.supplyAsync(supplier, executionService);

基本上等价于

CompletableFuture<?> f = new CompletableFuture<>();
executionService.execute(() -> {
    if(!f.isDone()) {
        try {
            f.complete(supplier.get());
        }
        catch(Throwable t) {
            f.completeExceptionally(t);
        }
    }
});

CompletableFuture 与 Executor 正在执行的代码之间没有任何联系，事实上，我们可以有任意数量的持续完成尝试。特定代码旨在完成 CompletableFuture 实例的事实只有在调用完成方法之一时才会变得明显。

因此，CompletableFuture 不能以任何方式影响正在运行的操作，这包括取消时中断等。正如the documentation of CompletableFuture 所说：

方法cancel和completeExceptionally(new CancellationException())效果一样

所以取消只是另一个完成尝试，如果它是第一个，它将获胜，但不会影响任何其他完成尝试。

所以orTimeout(long timeout, TimeUnit unit) 在这方面并没有太大的不同。超时后，它将执行相当于completeExceptionally(new TimeoutException())，如果没有其他完成尝试更快，它将获胜，这将影响依赖阶段，但不会影响其他正在进行的完成尝试，例如expensiveService.processAndGet() 在您的案例中发起了什么。

您可以像这样实现所需的操作

CompletableFuture<Upload> future = expensiveService.processAndGet();
CompletableFuture<Upload> alternative = CompletableFuture.supplyAsync(
    () -> resultService.get(processId), CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS));
return future.applyToEither(alternative, Function.identity())
    .whenComplete((u,t) -> alternative.cancel(false));

对于delayedExecutor，我们使用与orTimeout 和completeOnTimeout 相同的功能。当future.whenComplete 中的取消更快时，它不会在指定时间之前评估指定的Supplier 或根本不评估。 applyToEither 将提供更快的可用结果。

这不会在超时时完成future，但正如前面所说，它的完成不会影响原始计算，所以这也可以：

CompletableFuture<Upload> future = expensiveService.processAndGet();
CompletableFuture.delayedExecutor(2, TimeUnit.SECONDS)
    .execute(() -> {
        if(!future.isDone()) future.complete(resultService.get(processId));
    });
return future;

如前所述，这会在超时后完成未来，不会影响正在进行的计算，但会向调用者提供替代结果，但不会将resultService.get(processId) 抛出的异常传播到返回的未来。