本文将深入探讨在Project Reactor框架中，如何高效且非阻塞地将多个独立的Flux流的聚合结果合并为一个单一的Mono对象。通过详细分析常见的错误模式，并引入Reactor提供的zip操作符，我们将展示如何优雅地实现这一目标，确保应用程序的响应性和并发性。
引言：响应式数据流聚合的挑战
在响应式编程中，我们经常需要从不同的异步源获取数据流，并在所有数据都可用后将它们组合成一个统一的结果对象。例如，您可能需要从两个不同的服务获取成功账户列表和失败账户列表，然后将它们封装在一个Payments对象中。
一个常见的错误尝试是，在获取到Flux流后，立即调用collectList().subscribe()来获取数据，并尝试在订阅回调外部构建结果。然而，这种做法通常会导致阻塞，因为它试图在响应式流完成之前，同步地访问其结果。在Reactor中，subscribe()方法是非阻塞的，但如果您在订阅回调之外立即依赖其副作用来构建一个同步对象，那么在异步操作完成之前，您将无法获得所需的数据，从而引入阻塞或不确定的行为。
考虑以下数据模型和初始的错误尝试：
package org.example;
import lombok.Builder;
import lombok.Getter;
import lombok.ToString;
import java.util.List;
@Getter
@Builder
@ToString
public class Payments {
private List<SuccessAccount> successAccounts;
private List<FailedAccount> failedAccounts;
@Getter
@Builder
@ToString
public static class SuccessAccount {
private String name;
private String accountNumber;
}
@Getter
@Builder
@ToString
public static class FailedAccount {
private String name;
private String accountNumber;
private String errorCode;
}
}以及一个试图聚合的错误方法：
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.List;
public class Main {
public static Mono<Payments> getPaymentDataIncorrect() {
Flux<Payments.SuccessAccount> accountsSucceeded = getAccountsSucceeded();
Flux<Payments.FailedAccount> accountsFailed = getAccountsFailed();
List<Payments.SuccessAccount> successAccounts = new ArrayList<>();
List<Payments.FailedAccount> failedAccounts = new ArrayList<>();
// 这种方式是阻塞的，因为它试图在异步操作完成前同步地填充列表
accountsFailed.collectList().subscribe(failedAccounts::addAll);
accountsSucceeded.collectList().subscribe(successAccounts::addAll);
// 在此处，successAccounts和failedAccounts可能还未被填充
return Mono.just(Payments.builder()
.failedAccounts(failedAccounts)
.successAccounts(successAccounts)
.build());
}
// ... getAccountsSucceeded() 和 getAccountsFailed() 方法省略，与原始问题相同
}上述代码中的accountsFailed.collectList().subscribe(failedAccounts::addAll)和accountsSucceeded.collectList().subscribe(successAccounts::addAll)虽然subscribe本身是非阻塞的，但它不会立即填充failedAccounts和successAccounts。当Mono.just()被调用时，这两个列表很可能仍然是空的，因为订阅的回调是异步执行的。这导致了逻辑上的错误，并且如果强制同步等待，则会引入阻塞。
使用 zip 操作符实现非阻塞聚合
Project Reactor提供了zip操作符来解决这种场景。zip操作符能够将多个Publisher（例如Mono或Flux）的元素按照索引进行组合，当所有参与的Publisher都发出一个元素时，zip操作符会将这些元素组合成一个新的元素。
在我们的案例中，我们需要将两个Flux流的最终聚合结果（即List）组合起来。首先，我们可以使用collectList()操作符将每个Flux转换为一个Mono，表示该流所有元素的列表。然后，我们就可以使用Mono.zipWith()来组合这两个Mono
。
Mono.zipWith()接受另一个Mono作为参数，以及一个BiFunction（或更高阶的函数，如zip有多个重载），该函数定义了如何将两个Mono发出的结果组合成一个新的结果。
以下是使用zipWith操作符的正确实现：
package org.example;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
public class Main {
public static void main(String[] args) {
getPaymentData().subscribe(System.out::println);
}
public static Mono<Payments> getPaymentData() {
// 1. 获取两个独立的Flux流
Flux<Payments.SuccessAccount> accountsSucceededFlux = getAccountsSucceeded();
Flux<Payments.FailedAccount> accountsFailedFlux = getAccountsFailed();
// 2. 将每个Flux转换为一个Mono<List>
// collectList() 会收集Flux中的所有元素，并在Flux完成时发出一个包含这些元素的List
Mono<List<Payments.FailedAccount>> failedAccountsMono = accountsFailedFlux.collectList();
Mono<List<Payments.SuccessAccount>> successAccountsMono = accountsSucceededFlux.collectList();
// 3. 使用 Mono.zipWith() 组合两个 Mono<List>
// zipWith 会等待两个Mono都发出其结果，然后使用提供的BiFunction进行组合
Mono<Payments> combined = failedAccountsMono.zipWith(
successAccountsMono,
(failedAccounts, successAccounts) -> Payments.builder()
.failedAccounts(failedAccounts)
.successAccounts(successAccounts)
.build()
);
return combined;
}
// 模拟获取成功账户的Flux流
public static Flux<Payments.SuccessAccount> getAccountsSucceeded() {
return Flux.just(Payments.SuccessAccount.builder()
.accountNumber("1234345")
.name("Payee1")
.build(),
Payments.SuccessAccount.builder()
.accountNumber("83673674")
.name("Payee2")
.build());
}
// 模拟获取失败账户的Flux流
public static Flux<Payments.FailedAccount> getAccountsFailed() {
return Flux.just(Payments.FailedAccount.builder()
.accountNumber("12234345")
.name("Payee3")
.errorCode("8938")
.build(),
Payments.FailedAccount.builder()
.accountNumber("3342343")
.name("Payee4")
.errorCode("8938")
.build());
}
}在这个修正后的实现中：
- getAccountsSucceeded() 和 getAccountsFailed() 方法返回了两个独立的 Flux 流。
- accountsFailedFlux.collectList() 和 accountsSucceededFlux.collectList() 将这两个 Flux 转换为两个 Mono
- 。这些 Mono 会在各自的 Flux 完成收集所有元素后发出一个 List。
- failedAccountsMono.zipWith(successAccountsMono, ...) 操作符会等待 failedAccountsMono 和 successAccountsMono 都发出它们的 List 结果。一旦两个结果都可用，zipWith 会调用提供的 BiFunction（在本例中是一个Lambda表达式），将这两个 List 作为参数传入，并使用它们构建一个 Payments 对象。
- 最终，zipWith 操作符返回一个 Mono
，它会在 Payments 对象成功构建后发出该对象。整个过程是非阻塞的，并且完全符合响应式编程范式。
关键概念与优势
- 非阻塞性: zip 操作符是完全非阻塞的。它不会在等待上游Publisher发出元素时阻塞当前线程。相反，它会注册订阅，并在元素可用时异步地处理它们。
- 并发执行: zip 操作符的两个上游Publisher（在本例中是两个 collectList() 操作）可以并发地执行。这意味着获取成功账户和失败账户的数据流可以同时进行，从而提高整体效率。
- 结果组合的原子性: zip 确保只有当所有参与的Publisher都准备好发出一个元素时，组合函数才会被调用。这保证了在创建 Payments 对象时，所需的两个 List 数据都是完整且可用的。
- 错误处理: 如果任何一个上游 Mono 在发出其 List 之前失败，zip 操作符将立即传播该错误，而不会等待其他 Mono 完成。
总结
在Project Reactor中，当需要将多个独立的异步数据流（Flux或Mono）的最终结果聚合成一个单一的响应式对象时，zip操作符是首选的非阻塞解决方案。通过将每个Flux首先转换为一个Mono（使用collectList()），然后利用Mono.zipWith()结合一个自定义的组合函数，可以优雅且高效地实现复杂的聚合逻辑，同时保持应用程序的响应性和并发性。避免在响应式流中进行同步阻塞操作是构建高性能、可伸缩的响应式系统的关键。
