本文旨在探讨如何在project reactor框架中，以非阻塞的方式将两个独立的`flux`数据流的聚合结果合并为一个单一的`mono`对象。通过分析传统阻塞方法的不足，文章将重点介绍`mono.zipwith`操作符及其与`flux.collectlist()`的结合使用，以构建一个完全响应式、高效且易于维护的数据聚合解决方案，并提供详细的代码示例和最佳实践建议。
引言：响应式数据流聚合的挑战
在现代异步编程中，尤其是在基于Project Reactor等响应式框架构建的系统中，我们经常面临需要从多个独立的异步源获取数据，并将这些数据聚合成一个单一的、结构化的复合对象的场景。例如，从不同的服务或数据库查询成功账户列表和失败账户列表，然后将它们合并到一个支付结果对象中。
这种聚合操作的关键在于保持整个处理流程的非阻塞性。如果在此过程中引入任何阻塞操作，将可能导致线程资源浪费、系统吞吐量下降，并违背响应式编程的核心理念。
领域模型概览
为了更好地理解问题和解决方案，我们首先定义一个示例领域模型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 successAccounts;
private List 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流（一个产生SuccessAccount，另一个产生FailedAccount），然后将它们各自收集成列表，最终封装进一个Payments对象，并且整个过程是非阻塞的。
问题剖析：为何传统方法会阻塞
初学者在尝试聚合多个响应式流时，可能会不自觉地引入阻塞操作。考虑以下尝试将两个Flux收集为列表并构建Payments对象的代码：
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.List;
public class Main {
public static void main(String[] args) {
getPaymentData().subscribe(System.out::println);
}
public static Mono getPaymentData() {
Flux accountsSucceeded = getAccountsSucceeded();
Flux accountsFailed = getAccountsFailed();
List successAccounts = new ArrayList<>();
List failedAccounts = new ArrayList<>();
// 这里的subscribe调用是问题所在
accountsFailed.collectList().subscribe(failedAccounts::addAll); // 阻塞或导致竞态条件
accountsSucceeded.collectList().subscribe(successAccounts::addAll); // 阻塞或导致竞态条件
return Mono.just(Payments.builder()
.failedAccounts(failedAccounts)
.successAccounts(successAccounts)
.build());
}
// ... getAccountsSucceeded() 和 getAccountsFailed() 方法省略 ...
} 上述代码段中，accountsFailed.collectList().subscribe(failedAccounts::addAll); 和 accountsSucceeded.collectList().subscribe(successAccounts::addAll); 是问题的根源。
- subscribe() 的作用: subscribe() 是一个终端操作，它会触发响应式流的执行。然而，它本身是异步的，意味着当subscribe()被调用时，流的元素并不会立即被收集到failedAccounts或successAccounts列表中。
- 打破响应式链: 在subscribe()调用之后，紧接着的return Mono.just(...)会立即执行。此时，failedAccounts和successAccounts列表很可能仍然是空的，因为它们的填充是在异步的subscribe回调中进行的。这导致Payments对象被构建时包含了空列表，或者如果流处理需要时间，则会因为尝试同步获取异步结果而导致阻塞（尽管此处代码本身不会阻塞主线程，但它无法正确获取异步结果）。
- 非响应式: 这种模式实际上将响应式流的异步结果拉回到命令式代码中处理，破坏了整个操作的响应式特性。为了正确等待结果，开发者可能会引入block()操作，从而彻底失去了非阻塞的优势。
解决方案：使用Mono.zipWith实现非阻塞聚合
Project Reactor提供了zip系列操作符来解决这种并发聚合问题。Mono.zipWith（或静态方法Mono.zip）是专门用于将两个或多个Mono的结果合并成一个新Mono的强大工具。
其核心思想是：当所有参与zip操作的源Mono都成功发出其元素时，zip操作符会收集这些元素，并将它们作为参数传递给一个提供的BiFunction（或Function），该函数负责将这些元素组合成一个新的结果，然后由zip返回的Mono发出这个新结果。
在我们的场景中，我们需要将两个Flux转换为Mono，然后对这两个Mono
进行zip操作。Flux.collectList()操作符正是为此而生，它将一个Flux
>。
以下是使用Mono.zipWith的正确实现：
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.ArrayList;
import java.util.List;
public class Main {
public static void main(String[] args) {
getPaymentData().subscribe(System.out::println);
}
public static Mono getPaymentData() {
Flux accountsSucceededFlux = getAccountsSucceeded();
Flux accountsFailedFlux = getAccountsFailed();
// 1. 将Flux转换为Mono
Mono> failedAccountsMono = accountsFailedFlux.collectList();
Mono> successAccountsMono = accountsSucceededFlux.collectList();
// 2. 使用zipWith合并两个Mono的结果
return failedAccountsMono.zipWith(
successAccountsMono,
(failedAccounts, successAccounts) -> Payments.builder()
.failedAccounts(failedAccounts)
.successAccounts(successAccounts)
.build()
);
}
// 模拟获取成功账户的Flux
public static Flux getAccountsSucceeded() {
return Flux.just(Payments.SuccessAccount.builder()
.accountNumber("1234345")
.name("Payee1")
.build(),
Payments.SuccessAccount.builder()
.accountNumber("83673674")
.name("Payee2")
.build());
}
// 模拟获取失败账户的Flux
public static Flux getAccountsFailed() {
return Flux.just(Payments.FailedAccount.builder()
.accountNumber("12234345")
.name("Payee3")
.errorCode("8938")
.build(),
Payments.FailedAccount.builder()
.accountNumber("3342343")
.name("Payee4")
.errorCode("8938")
.build());
}
}
代码解析：
-
failedAccountsFlux.collectList(): 这个操作符将Flux
转换成一个Mono - >。这意味着当原始的Flux发出所有元素并完成时，collectList()会收集这些元素到一个列表中，并将这个列表作为单一元素由返回的Mono发出。
-
failedAccountsMono.zipWith(successAccountsMono, ...):
- failedAccountsMono是第一个源Mono，它会发出失败账户列表。
- successAccountsMono是第二个源Mono，它会发出成功账户列表。
- BiFunction (failedAccounts, successAccounts) -> ... 是一个组合函数。当两个源Mono都成功发出它们的列表时，zipWith会调用这个函数，将两个列表作为参数传入。
- 在这个函数内部，我们使用Payments.builder()来构建最终的Payments对象，将两个列表分别设置到failedAccounts和successAccounts字段中。
-
非阻塞性: 整个链条都是非阻塞的。zipWith会等待两个上游Mono都完成后才执行组合函数，并且它本身返回一个Mono
，允许消费者在它发出结果时进行订阅和处理，而无需在中间环节阻塞。
总结与最佳实践
- 保持响应式链条完整: 在Reactor中，避免在中间操作中调用subscribe()。subscribe()应该作为整个响应式链的最后一个操作，用于触发执行并处理最终结果。在链条中间，应使用各种操作符（如map、flatMap、filter、zip等）来转换和组合流。
- 利用zip系列操作符: 当需要将多个独立的异步结果聚合成一个复合结果时，Mono.zip或Flux.zip是理想的选择。它们确保所有依赖的异步操作都完成后才进行组合，同时保持非阻塞。
-
collectList()的重要性: Flux.collectList()是将Flux转换为Mono
- 的关键操作。这在需要将一个元素流聚合成一个集合，并进一步参与Mono操作（如zipWith）时非常有用。
- 错误处理: zip操作符具有“快速失败”的特性。如果任何一个参与zip的源Mono发出错误，那么整个zip操作返回的Mono也会立即发出该错误，而不会等待其他源完成。在实际应用中，应考虑如何使用onErrorResume、doOnError等操作符进行错误处理。
通过遵循这些原则，开发者可以构建出高效、健壮且完全响应式的应用程序，充分利用Project Reactor带来的并发和非阻塞优势。
