本教程深入探讨了在reactor框架中如何高效、非阻塞地将多个独立的flux流的聚合结果合并为一个单一的mono对象。通过分析传统阻塞方法的不足，文章重点介绍了mono.zipwith操作符，并结合实际业务场景，提供了详细的代码示例，旨在帮助开发者掌握响应式编程中聚合操作的最佳实践，确保应用的非阻塞性和响应性。

响应式流聚合的挑战

在基于Project Reactor构建的响应式应用中，我们经常需要处理来自不同异步源的数据流，并将它们聚合到一个统一的响应式对象中。例如，您可能需要从两个独立的Flux流中收集元素列表，然后将这两个列表封装到一个最终的Mono对象中。然而，如果不正确地处理这些流，很容易引入阻塞操作，从而违背响应式系统的核心原则。

考虑以下业务场景和数据模型，我们需要收集成功账户和失败账户的列表，并将它们组合成一个Payments对象：

package org.example;

import lombok.Builder;
import lombok.Getter;
import lombok.ToString;

import j*a.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 和 FluxailedAccount>，并将它们聚合到一个 Mono 对象中，整个过程必须是非阻塞的。

传统阻塞方法的陷阱

初学者在处理此类聚合需求时，可能会尝试将每个 Flux 通过 collectList() 转换为 Mono>，然后立即订阅这些 Mono 来收集数据。例如：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import j*a.util.ArrayList;
import j*a.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

TapNow

新一代AI视觉创作引擎

407 查看详情

首先，我们需要将每个 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 j*a.util.List;

// Payments.j*a (与前面定义相同，此处省略重复代码)
/*
@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.") // 完成信号
        );

        // 为了演示异步性，

以上就是Reactor框架中非阻塞地聚合多个Flux流为单个Mono对象的详细内容，更多请关注其它相关文章！

# 的是  # 高埗高端网站建设  # 根河模板网站建设  # 泉州网站建设配置  # 免费b站推广网站在线看  # 搜狐营销公众号推广方案  # 哪里有网络营销推广在线咨询  # 奶茶店营销新店推广方案  # 江西做网站建设  # 沈阳网站建站推广  # 青铜峡门户网站建设  # 就会  # 这是  # react  # 装在  # 象中  # 并将  # 这两个  # 转换为  # 中非  # 多个  # 响应式编程  # ai  # ssl  # java 

相关栏目： 【 科技资讯46185 】 【 网络学院92790 】

相关推荐： 内存检查：在VS Code中调试C++时的内存视图  Composer的 "conflict" 字段有什么用_如何声明不兼容的包以避免依赖冲突  一加手机拍照效果不好怎么办 一加哈苏影像调校与专业模式使用教程【高手篇】  如何使 Jest 模拟函数默认抛出错误以提高测试效率  C++如何进行游戏物理模拟_使用Box2D库为C++游戏添加2D物理效果  mysql如何设置表访问权限_mysql表访问权限配置  智慧团建扫码登录入口 智慧团建扫码登录入口官网版​  mc.js免安装版 mc.js一键畅玩入口  如何在离线环境中使用Composer_Composer离线安装依赖包的技巧与策略  C++如何使用AddressSanitizer(ASan)_C++调试工具中检测内存访问错误的利器  德邦快递查询平台 德邦快递物流信息查询入口  word中如何让数字纵向排列_Word数字纵向排列方法  在J*a中如何使用BigDecimal进行高精度计算_BigDecimal类应用指南  漫蛙官网正版漫画入口 漫蛙2官方网页登录地址  html怎么运行外部js文件中的函数_运html外js文件函数法【技巧】  J*a应用程序首次运行自动创建文件与目录的最佳实践  单射、满射与双射的关系 一文理清所有逻辑  俄罗斯Yandex搜索引擎入口_Yandex官网免登录一键访问  GemBox Document HTML转PDF垂直文本渲染问题及解决方案  离线运行Go语言之旅：本地部署与GOPATH配置指南  React列表渲染与独立状态管理：避免全局状态影响局部更新  Win10怎么设置静态IP地址 Win10手动配置IP地址步骤【指南】  163邮箱登录密码 163邮箱忘记密码找回  抖音极速版最新版本 抖音极速版官方下载地址  css滚动动画效果怎么实现_使用Animate.css滚动触发动画类  DLsite中文平台入口 DLsite官网内容在线查看  Golang如何安装Swagger工具_GoSwagger文档生成环境  理解Python模块与全局变量的作用域管理  从J*aScript对象中精确提取指定属性的教程  为什么简单的XML文件也会解析失败？ 检查隐藏的非打印字符（如BOM）的方法  qq浏览器如何查看和导出已保存的密码 qq浏览器密码管理器数据备份教程  快手网页版在线登录 快手网页版官网入口快速访问  探索高级语言到C/C++的转译路径：以Go为例及内存管理策略  Win11怎么设置开机NumLock亮 Win11修改注册表InitialKeyboardIndicators值  QQ邮箱在线登录平台 QQ邮箱个人邮箱网页版入口  J*a递归快速排序中静态变量导致数据累积的陷阱与解决方案  网站内容防复制粘贴的实现策略与局限性  C++ explicit关键字防止隐式转换_C++构造函数安全规范  蛙漫漫画官网在线入口 蛙漫全本漫画免费阅读平台  漫蛙Manwa2官网入口地址分享 漫蛙漫画PC版永久访问通道  Yandex官方入口网址 Yandex俄罗斯搜索引擎最新在线地址  QQ网页版官方账号入口 QQ网页版网页版登录指南  修复二维数组索引越界异常：一维循环到二维坐标的正确映射  京东单号查询入口_京东快递订单追踪入口  Lar*el Excel导入时生成自定义递增ID的策略与实践  韩剧圈正版入口页面_韩剧圈官网登录链接  ExcelARRAYTOTEXT函数怎么自定义分隔符输出数组文本_ARRAYTOTEXT实现动态生成SQL语句  vivo浏览器怎么扫描二维码 vivo浏览器内置扫一扫功能使用方法  composer的"require-dev"部分是用来做什么的？  在J*a中如何捕获IndexOutOfBoundsException_索引越界异常防护方法说明