java - 如何迭代 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() 但也许有一种响应式(Reactive)方法，而不是在 Flux 流中调用 block 方法？

Answer 1

此代码示例中有几个问题。我假设这是一个响应式(Reactive) Web 应用程序。

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

此外，Flux.fromIterable(userRepository.findAllByRole(Role.USER)) 建议您从响应式上下文中调用阻塞存储库。

您不应该永远在doOn*** 运算符中执行大量 I/O 操作。这些是为日志记录或轻微副作用操作而设计的。事实上，您需要对其进行 .block() ，这是您将阻塞整个响应式(Reactive)管道的另一个线索。

最后一个:您不应在 Web 应用程序中的任何位置调用 subscribe。如果这绑定(bind)到 HTTP 请求，那么您基本上会触发响应式(Reactive)管道，而无法保证资源或完成情况。调用 subscribe 会触发管道，但不会等到它完成(此方法返回 Disposable)。

一个更“典型”的示例如下所示:

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

如果您不知何故遇到了阻塞组件(例如 sendEmail 组件)，您应该在特定的调度程序上安排这些阻塞操作，以避免阻塞整个响应式(Reactive)管道。为此，请查看Schedulers section on the reactor reference documentation .