1

有没有办法在超时后不取消未来的情况下尝试等待一段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 s 并完成它的未来。它返回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 传递给第二个，这不是很干净。

#2Object.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 瓦夫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. 需要一个未来中的未来以避免await取消内部未来。
  2. 将第一个 Future 移到第二个会破坏依赖ThreadLocals 的 [legacy] 代码。
  3. recoverWith并捕捉TimeoutException不是那么优雅。

#4CompletableFuture.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. 异常处理不好。

#5CompletableFuture.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不同的状态返回怎么办？

所有这些解决方案都有缺点并且需要额外的代码，但这感觉像是应该由CompletableFutureVavrFuture开箱即用的东西来支持。有一个更好的方法吗？

4

1 回答 1

3

值得首先指出的是，如何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不能以任何方式影响运行操作，这包括取消时中断等。正如文档CompletableFuture所说：

方法取消具有相同的效果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相同的设施。当取消速度更快时，它不会在指定时间之前评估指定的或根本不评估。将提供更快可用的任何结果。orTimeoutcompleteOnTimeoutSupplierfuture.whenCompleteapplyToEither

这不会完成futureon 超时，但如前所述，它的完成不会影响原始计算，所以这也可以工作：

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

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

于 2020-01-20T17:17:30.960 回答