Reactor框架中非阻塞地聚合多个Flux流为单个Mono对象

本教程深入探讨了在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 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 和 FluxailedAccount>，并将它们聚合到一个 Mono 对象中，整个过程必须是非阻塞的。

传统阻塞方法的陷阱

初学者在处理此类聚合需求时，可能会尝试将每个 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 getAccountsSucceeded() { /* ... */ return Flux.empty(); }
    // 模拟获取失败账户的Flux
    public static Flux getAccountsFailed() { /* ... */ return Flux.empty(); }

    public static Mono getPaymentDataBlocking() {
        Flux accountsSucceeded = getAccountsSucceeded();
        Flux accountsFailed = getAccountsFailed();

        List successAccounts = new ArrayList<>();
        List 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

Quinvio AI

AI辅助下快速创建视频，虚拟代言人

下载

首先，我们需要将每个 Flux 转换为一个 Mono>。collectList() 操作符正是为此而生，它会收集 Flux 中的所有元素，并在 Flux 完成时，将这些元素作为一个 List 封装在 Mono 中发出。

// 获取成功的账户流
Flux accountsSucceeded = getAccountsSucceeded();
// 获取失败的账户流
Flux accountsFailed = getAccountsFailed();

// 将Flux转换为Mono
Mono> failedAccountsMono = accountsFailed.collectList();
Mono> 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 getPaymentDataNonBlocking() {
        Flux accountsSucceeded = getAccountsSucceeded();
        Flux accountsFailed = getAccountsFailed();

        return accountsFailed.collectList() // 第一个Mono：Mono>
                .zipWith(
                        accountsSucceeded.collectList(), // 第二个Mono：Mono>
                        (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 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;
    }
}
*/

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.") // 完成信号
        );

        // 为了演示异步性，