Java Reactor - 条件流执行

Question

我想知道如何使用 Reactor 创建“逻辑流”。

让我们假设我想实现以下场景:

作为输入，我有对象要保存在数据库中。作为输出，我想获得 Mono 表示执行消息。

选项 1:如果要保存的对象已填充所有字段，则执行其他操作，将其保存到数据库中，最后返回“Success”消息

选项 2:如果要保存的对象至少有一个字段未填写，我返回“Error”

我已经创建了这样的代码:

Mono<String> message = Mono.just(new User("Bob the Reactor master")) // user with name = can be saved
    .flatMap(user -> {
        if(user.getName() != null && user.getName().length() > 1){
            // Perform additional operations e.g. user.setCreatedDate(new Date())
            // Save to repository e.g. repository.save(user)
            return Mono.just("Success!");
        }
        else{
            return Mono.just("Error!");
        }
    })
    .doOnNext(System.out::println); // print stream result

message.subscribe();

这段代码是 100% react 性的吗(有它的所有好处)？如果没有，那么它会是什么样子？

Answer 1

答案取决于您评论的存储库。

存储库是非阻塞的并返回 Mono 或 Flux

您应该订阅它然后返回 Success Mono。在您的 if 语句中:

return repository.save(user).then(Mono.just("Success!"));

存储库被阻塞

您应该让您的存储库调用非阻塞将其执行移动到单独的线程。 Reactor 的方法是用 Mono 包装它并订阅弹性调度程序或您的自定义调度程序。

Mono<String> message = Mono.just(new User("Bob the Reactor master"))
        .filter(user -> user.getName() != null && user.getName().length() > 1)
        .map(user -> user) // Perform additional operations e.g. user.setCreatedDate(new Date())
        .flatMap(user -> Mono.fromRunnable(() -> repository.save(user))
                .subscribeOn(Schedulers.elastic())
                .then(Mono.just("Success!")))
        .switchIfEmpty(Mono.just("Error!"));