本教程深入探讨了在reactor框架中如何高效、非阻塞地将多个独立的flux流的聚合结果合并为一个单一的mono对象。通过分析传统阻塞方法的不足，文章重点介绍了mono.zipwith操作符，并结合实际业务场景，提供了详细的代码示例，旨在帮助开发者掌握响应式编程中聚合操作的最佳实践，确保应用的非阻塞性和响应性。
响应式流聚合的挑战
在基于Project Reactor构建的响应式应用中，我们经常需要处理来自不同异步源的数据流，并将它们聚合到一个统一的响应式对象中。例如，您可能需要从两个独立的Flux流中收集元素列表，然后将这两个列表封装到一个最终的Mono对象中。然而，如果不正确地处理这些流，很容易引入阻塞操作，从而违背响应式系统的核心原则。
考虑以下业务场景和数据模型，我们需要收集成功账户和失败账户的列表，并将它们组合成一个Payments对象：
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;
}
}我们的目标是获取 Flux
传统阻塞方法的陷阱
初学者在处理此类聚合需求时，可能会尝试将每个 Flux 通过 collectList() 转换为 Mono>，然后立即订阅这些 Mono 来收集数据。例如：
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.List;
// 这是一个错误的阻塞示例
public class MainBlockingExample {
// 模拟获取成功账户的Flux
public static Flux<Payments.SuccessAccount> getAccountsSucceeded() { /* ... */ return Flux.empty(); }
// 模拟获取失败账户的Flux
public static Flux<Payments.FailedAccount> getAccountsFailed() { /* ... */ return Flux.empty(); }
public static Mono<Payments> getPaymentDataBlocking() {
Flux<Payments.SuccessAccount> accountsSucceeded = getAccountsSucceeded();
Flux<Payments.FailedAccount> accountsFailed = getAccountsFailed();
List<Payments.SuccessAccount> successAccounts = new ArrayList<>();
List<Payments.FailedAccount> failedAccounts = new ArrayList<>();
// 这里的subscribe方法在主线程中执行时，会阻塞当前线程直到数据收集完成
// 这与响应式编程的非阻塞原则相悖。在实际业务场景中，如果方法需要返回Mono
// 而不是在内部等待结果，这种方式是不可取的。
accountsFailed.collectList().subscribe(failedAccounts::addAll);
accountsSucceeded.collectList().subscribe(successAccounts::addAll);
return Mono.just(Payments.builder()
.failedAccounts(failedAccounts)
.successAccounts(successAccounts)
.build());
}
}尽管 collectList() 返回的是一个 Mono，但直接在其上调用 subscribe() 并试图在订阅回调中同步收集数据，在需要返回 Mono 的方法内部是典型的反模式。这会导致当前线程阻塞，等待 subscribe 操作完成，从而失去了响应式编程的非阻塞优势。在Spring WebFlux控制器等场景中，我们期望方法直接返回一个 Mono 或 Flux，而不是在方法内部阻塞等待结果。
解决方案：使用 Mono.zipWith 进行非阻塞聚合
Reactor 提供了 zip 操作符，专门用于将多个发布者（Publisher）的元素按照索引进行组合。当所有参与 zip 操作的发布者都发出一个元素时，zip 才会将这些元素组合成一个元组（Tuple）并向下游发出。对于聚合两个 Mono> 的场景，Mono.zipWith 是理想的选择。
1. 将 Flux 转换为 Mono
首先，我们需要将每个 Flux 转换为一个 Mono>。collectList() 操作符正是为此而生，它会收集 Flux 中的所有元素，并在 Flux 完成时，将这些元素作为一个 List 封装在 Mono 中发出。
// 获取成功的账户流 Flux<Payments.SuccessAccount> accountsSucceeded = getAccountsSucceeded(); // 获取失败的账户流 Flux<Payments.FailedAccount> accountsFailed = getAccountsFailed(); // 将Flux转换为Mono<List> Mono<List<Payments.FailedAccount>> failedAccountsMono = accountsFailed.collectList(); Mono<List<Payments.SuccessAccount>> successAccountsMono = accountsSucceeded.collectList();
2. 使用 Mono.zipWith 组合结果
现在，我们有了两个 Mono 对象，分别持有失败账户列表和成功账户列表。接下来，可以使用 Mono.zipWith 将它们组合起来。zipWith 方法接受另一个 Mono 和一个 BiFunction。BiFunction 定义了如何将这两个 Mono 发出的值组合成一个新值。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class Main {
// ... (getAccountsSucceeded 和 getAccountsFailed 方法定义)
/**
* 非阻塞地聚合成功账户和失败账户列表到Payments对象。
*
* @return 包含Payments对象的Mono
*/
public static Mono<Payments> getPaymentDataNonBlocking() {
Flux<Payments.SuccessAccount> accountsSucceeded = getAccountsSucceeded();
Flux<Payments.FailedAccount> accountsFailed = getAccountsFailed();
return accountsFailed.collectList() // 第一个Mono：Mono<List<FailedAccount>>
.zipWith(
accountsSucceeded.collectList(), // 第二个Mono：Mono<List<SuccessAccount>>
(failedList, successList) -> Payments.builder() // BiFunction组合两个列表
.failedAccounts(failedList)
.successAccounts(successList)
.build()
);
}
}在这个解决方案中，accountsFailed.collectList() 和 accountsSucceeded.collectList() 会并行地收集各自的账户列表。一旦两个 Mono 都完成了其内部的列表收集并发出值，zipWith 的 BiFunction 就会被调用，使用这两个列表构建 Payments 对象，并将其封装在一个新的 Mono
完整示例代码
为了更好地理解和实践，以下是包含 Payments 模型和 Main 类的完整示例：
package org.example;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
// Payments.java (与前面定义相同，此处省略重复代码)
/*
@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;
}
}
*/
public class Main {
public static void main(String[] args) {
// 订阅并打印结果，这是在应用启动时触发流的示例
getPaymentDataNonBlocking().subscribe(
System.out::println, // 成功处理
error -> System.err.println("Error: " + error.getMessage()), // 错误处理
() -> System.out.println("Processing complete.") // 完成信号
);
// 为了演示异步性， 
