运行顺序异步操作

Question

我有一系列 I/O 操作（数据库，I/O 设备...）我需要按顺序运行。

@SafeVarargs
public final CompletableFuture<Boolean> execute(final Supplier<Boolean>... methods)
{
    CompletableFuture<Boolean> future = null;

    for (Supplier<Boolean> method : methods)
    {
        if (future == null)
        {
            future = CompletableFuture.supplyAsync(method, threadPool);
        }
        else
        {
            future.thenCombineAsync(CompletableFuture.supplyAsync(method, threadPool), (result, currentResult) -> result && currentResult,
                    threadPool);
        }
    }

    return future.exceptionally(this::onException);
}

我的代码随机执行。

我该怎么做才能保证秩序？
最后怎么合并结果？例如，如果一切都是真的？
要在一切完成后应用回调以检查结果？

Answer 1

您当前的解决方案会立即调用 supplyAsync()，稍后会尝试合并结果。

如果要保证顺序执行，应该使用thenApply()或thenCompose()而不是thenCombine():

for (Supplier<Boolean> method : methods)
{
    if (future == null)
    {
        future = CompletableFuture.supplyAsync(method, threadPool);
    }
    else
    {
        future.thenApplyAsync(result -> result && method.get(), threadPool);
    }
}

请注意，如果其中任何 returns 为假，这将不会调用下一个供应商的 method.get()，因为 && 正在短路。您可以使用单个 & 强制调用，或者交换参数。

这已经在最后合并了所有布尔结果。您可以在循环后的结果 future 上添加任何内容，例如更多 thenApply() 调用，或阻塞 join() 调用以检索 Boolean.

请注意，此循环也可以使用流轻松重写：

future = Arrays.stream(methods)
        .reduce(CompletableFuture.completedFuture(true),
                (f, method) -> f.thenApplyAsync(result -> result && method.get()),
                (f1, f2) -> f1.thenCombine(f2, (result1, result2) -> result1 && result2));

Answer 2

您可以使用 Spotify CompletableFutures 库轻松地做到这一点：https://github.com/spotify/completable-futures

他们为此提供了一些非常有用的工具，例如 allAsList returns CompletableFuture<List<T>>。

运行顺序异步操作

Run async actions in sequence

java

concurrency

completable-future

运行 顺序异步操作

Run async actions in sequence

java

concurrency

completable-future

运行顺序异步操作