project-reactor - 在项目 react 器中处理 ListenableFuture

Question

我已经开始使用项目 react 器，并希望将我们的 API 之一移动到响应式做事方式。我想知道处理像 ListenableFuture 这样的事情是什么。

在我的例子中，我使用的是 Cassandra，当我调用 session.executeAsync() 时，它会返回一个扩展了 ListenableFuture 的 ResultSetFuture。下面是我现在编写的代码示例，我似乎不喜欢将 ListenableFuture 暴露给客户。

public Mono<ListenableFuture<Void>> save(Publisher<AccountDTO> accountPublisher) {
    return Mono.just(accountPublisher)
            .map(accountDTO -> {
                Account accountEntity = modelMapper.map(accountDTO, Account.class);
                return mappingManager.mapper(Account.class).saveAsync(accountEntity);
            })
            .retry(1)
            .doOnError(throwable -> log.error("Unable to create account "))
            .mapError(throwable -> new MyCustomException(""));
}

我的问题是:

Is it a good practice to expose ListenableFuture, I personally don't want to give anything like this back to the client where they can block. Is there a better way to handle this in project reactor where I can just return a Mono?

Answer 1

您可以轻松桥接 ListenableFuture<Void>异步 API 改为公开 Mono<Void> , 通过使用 Mono.create()工厂方法。该方法采用 Consumer<Sink> ，您将其作为 lambda 提供:

1. 为调用sink.success() 的 future 添加一个成功监听器(因为没有实际值，或者你也可以用监听器收到的success(aVoid)值调用Void)
2. 为调用sink.error(failure) 的 future 添加一个失败监听器

差不多就这些了!请参阅有关 create 的引用文档(虽然这个提到了 Flux 版本，由于必须处理多个值，它有点复杂):http://projectreactor.io/docs/core/release/reference/docs/index.html#producing.create