如何迭代Flux并与Mono混合

Question

我有一个应该向用户发送电子邮件的用例。首先，我创建电子邮件正文。

Mono<String> emailBody = ...cache();

然后我选择用户并向他们发送电子邮件：

Flux.fromIterable(userRepository.findAllByRole(Role.USER))
            .map(User::getEmail)
            .doOnNext(email -> sendEmail(email, emailBody.block(), massSendingSubject))
            .subscribe();

我不喜欢什么

1. 没有cache（）方法，emailBody Mono会在每个迭代步骤中进行计算
2. 要获得emailBody值，我使用emailBody。block（），但可能有一种反应方式，而不是在通量流中调用block方法

Answer 1

此代码示例中有几个问题。我假设这是一个反应式web应用程序。

首先，不清楚您是如何创建电子邮件正文的；您是从数据库还是远程服务获取东西？如果它主要是CPU绑定的（而不是I/O），那么您不需要将其包装为反应式类型。现在，如果它应该是Publisher中的包装器并且所有用户的电子邮件内容都是相同的，那么使用cache运算符并不是一个糟糕的选择。

此外，Flux.fromIterable（userRepository.findAllByRole（Role. USER））建议您从反应式上下文调用阻塞存储库。

千万不要在***操作符中执行繁重的I/O操作。这些都是为日志记录或轻微的副作用操作而设计的。事实上，您需要。block（）这是另一条线索，表明您将阻塞整个反应管道。

最后一点：您不应该在web应用程序中的任何位置调用subscribe。如果这是绑定到HTTP请求的，则基本上是触发反应性管道，而不保证资源或完成。调用subscribe会触发管道，但不会等到管道完成（此方法返回一个一次性的）。

更“典型”的示例如下所示：

Flux<User> users = userRepository.findAllByRole(Role.USER);
String emailBody = emailContentGenerator.createEmail();


// sendEmail() should return Mono<Void> to signal when the send operation is done
Mono<Void> sendEmailsOperation = users
     .flatMap(user -> sendEmail(user.getEmail(), emailBody, subject))
     .then();

// something else should subscribe to that reactive type,
// you could plug that as a return value of a Controller for example

如果您在某种程度上被阻塞组件（例如sendmail）所困扰，您应该在特定的调度器上安排这些阻塞操作，以避免阻塞整个反应管道。为此，请参阅Reactor参考文档中的Schedulers部分。