java - 如何在 CompletionStage.exceptionally 中链接非阻塞操作

Question

我正在用 Java 编写一个 Play2 应用程序服务方法，它应该执行以下操作。异步调用A方法，失败再异步调用B方法。

为了说明，假设此接口(interface)用于服务调用的后端:

public interface MyBackend {
    CompletionStage<Object> tryWrite(Object foo);
    CompletionStage<Object> tryCleanup(Object foo);
}

所以在我的服务方法中，我想返回一个可以完成这些的 future :

• tryWrite成功完成
• tryWrite 失败和 tryCleanup 成功完成并失败，但 tryWrite() 除外

(注意:当然 tryWrite() 可以自己做任何清理，这是一个说明问题的简化示例)

像这样调用后端的服务的实现对我来说似乎很难，因为 CompletionStage.exceptionally() 方法不允许 Composing。

版本 1:

public class MyServiceImpl {
    public CompletionStage<Object> tryWriteWithCleanup(Object foo) {

        CompletionStage<Object> writeFuture = myBackend.tryWrite(foo)
            .exceptionally((throwable) -> {
                CompletionStage<Object> cleanupFuture = myBackend.tryCleanup(foo);
                throw new RuntimeException(throwable);
        });
        return writeFuture;
    }
}

因此版本 1 以非阻塞方式调用 tryCleanup(foo)，但 tryWriteWithCleanup() 返回的 CompletionStage 不会等待 cleanupFuture 完成。如何更改此代码以从同样等待 cleanupFuture 完成的服务返回 future ？

版本 2:

public class MyServiceImpl {
    public CompletionStage<Object> tryWriteWithCleanup(Object foo) {

        final AtomicReference<Throwable> saveException = new AtomicReference<>();
        CompletionStage<Object> writeFuture = myBackend
            .tryWrite(foo)
            .exceptionally(t -> {
                saveException.set(t);
                // continue with cleanup
                return null;
            })
            .thenCompose((nil) -> {
                // if no cleanup necessary, return
                if (saveException.get() == null) {
                    return CompletableFuture.completedFuture(null);
                }
                return CompletionStage<Object> cleanupFuture = myBackend.tryCleanup(foo)
                    .exceptionally(cleanupError -> {
                        // log error
                        return null;
                    })
                    .thenRun(() -> {
                        throw saveException.get();
                    });
        });
        return writeFuture;
    }
}

版本 2 使用外部 AtomicReference 来存储失败，如果失败，则在另一个 thenCompose() block 中进行异步第二次调用。

我所有其他的尝试都以失败告终，我不想将它们粘贴在这里。

Answer 1

不幸的是CompletionStage/CompletableFuture不提供具有组合的异常处理 API。

你可以通过依赖 handle() 来解决这个问题用BiFunction返回 CompletionStage .这将为您提供嵌套阶段 ( CompletionStage<CompletionStage<Object>> )，您可以使用 compose(identity()) 来“unnest” :

public CompletionStage<Object> tryWriteWithCleanup(Object foo) {
    return myBackend.tryWrite(foo)
            .handle((r, e) -> {
                if (e != null) {
                    return myBackend.tryCleanup(foo)
                            .handle((r2, e2) -> {
                                // Make sure we always return the original exception
                                // but keep track of new exception if any,
                                // as if run in a finally block
                                if (e2 != null) {
                                    e.addSuppressed(e2);
                                }
                                // wrapping in CompletionException  behaves as if
                                // we threw the original exception
                                throw new CompletionException(e);
                            });
                }
                return CompletableFuture.completedFuture(r);
            })
            .thenCompose(Function.identity());
}