本文旨在详细阐述在project reactor框架中，如何优雅且非阻塞地将两个独立的flux流处理后的结果聚合为一个单一的mono对象。通过分析传统阻塞式操作的弊端，我们将重点介绍并演示mono.zipwith操作符的正确使用方法，以实现高效、响应式的并发数据聚合，从而避免在异步流程中引入阻塞点。

1. 理解响应式流中的非阻塞聚合需求

在响应式编程中，我们经常需要从多个独立的异步源获取数据，并将这些数据组合成一个统一的结果对象。例如，一个支付服务可能需要同时从不同的子系统获取成功交易列表和失败交易列表，然后将它们封装在一个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流：

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());
}

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());
}

一个常见的误区是尝试通过订阅这些Flux流并将结果收集到可变列表中，然后构建最终对象。例如：

// 这是一个阻塞的、不推荐的做法
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() 会立即触发流的执行，并在当前线程等待结果，导致阻塞
    accountsFailed.collectList().subscribe(failedAccounts::addAll);
    accountsSucceeded.collectList().subscribe(successAccounts::addAll);

    return Mono.just(Payments.builder()
            .failedAccounts(failedAccounts)
            .successAccounts(successAccounts)
            .build());
}

上述代码中的subscribe()调用是阻塞的，因为它会在当前线程等待collectList()操作完成，这违背了Reactor非阻塞的原则。在实际的Web服务或异步处理场景中，这种阻塞操作会导致线程池资源耗尽，严重影响系统吞吐量和响应性。

2. 使用Mono.zipWith 实现非阻塞聚合

为了在Reactor中实现真正的非阻塞聚合，我们需要利用其提供的组合操作符。Mono.zipWith（或Mono.zip）是解决此类问题的理想选择。它允许我们将两个Mono（或多个Mono）的结果组合起来，一旦所有源Mono都完成了并产生了它们的值，就会使用一个提供的BiFunction（或Function）来处理这些值，并生成一个新的Mono结果。

Tunee AI

新一代AI音乐智能体

1104 查看详情

具体步骤如下：

1. 将Flux转换为Mono： 首先，我们需要将每个Flux流通过collectList()操作符转换为一个发出单个List的Mono。这个Mono将在原始Flux完成并收集所有元素后发出其列表。
2. 使用zipWith组合： 接下来，将第一个Mono与第二个Mono使用zipWith操作符进行组合。
3. 提供组合函数： zipWith需要一个BiFunction作为参数，该函数接收两个Mono发出的值（即两个List），并返回我们期望的最终结果（即Payments对象）。

下面是使用Mono.zipWith实现的非阻塞解决方案：

package org.example;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import j*a.util.ArrayList;
import j*a.util.List;

public class Main {
    public static void main(String[] args) {
        // 订阅并打印结果，这是在应用程序入口点进行的操作，不会阻塞核心业务逻辑
        getPaymentData().subscribe(System.out::println);

        // 为了在main方法中观察异步结果，通常需要一些延迟或等待机制
        // 在实际应用中，例如Spring WebFlux控制器，Mono会被框架自动订阅和处理
        try {
            Thread.sleep(1000); // 简单等待，仅用于演示
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static Mono<Payments> getPaymentData() {
        Flux<Payments.SuccessAccount> accountsSucceededFlux = getAccountsSucceeded();
        Flux<Payments.FailedAccount> accountsFailedFlux = getAccountsFailed();

        // 将Flux转换为Mono<List>
        Mono<List<Payments.SuccessAccount>> successAccountsMono = accountsSucceededFlux.collectList();
        Mono<List<Payments.FailedAccount>> failedAccountsMono = accountsFailedFlux.collectList();

        // 使用 zipWith 组合两个 Mono 的结果
        Mono<Payments> combinedPaymentsMono = failedAccountsMono.zipWith(
                successAccountsMono,
                (failedAccounts, successAccounts) -> Payments.builder()
                        .failedAccounts(failedAccounts)
                        .successAccounts(successAccounts)
                        .build()
        );

        return combinedPaymentsMono;
    }

    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());
    }

    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());
    }
}

在这个改进后的getPaymentData()方法中：

• accountsSucceededFlux.collectList()和accountsFailedFlux.collectList()各自返回一个Mono。这两个Mono会并行地收集它们各自Flux中的所有元素。
• failedAccountsMono.zipWith(successAccountsMono, ...)操作符会等待这两个Mono都完成并发出它们的结果（即两个List）。
• 一旦两个List都可用，zipWith会调用提供的BiFunction，将这两个List作为参数传入，然后使用它们来构建并发出最终的Payments对象。
• 整个过程都是非阻塞的，getPaymentData()方法会立即返回一个Mono，而实际的数据处理和对象构建则会在背后的Reactor调度器上异步执行。

3. 注意事项与最佳实践

• 避免中间订阅： 在响应式链中，除了最终的消费者（如REST控制器返回Mono或在main方法中打印结果），应尽量避免使用subscribe()来获取中间结果。subscribe()会触发流的执行，并且其副作用（如修改外部变量）在异步环境中难以管理，也容易引入阻塞。
• 利用组合操作符： Reactor提供了丰富的组合操作符（如zip、merge、concat、when等），它们是处理多个响应式流的强大工具。选择正确的操作符取决于你希望如何组合这些流的行为（例如，并行等待所有完成、按顺序合并、或只关心第一个完成的）。
• 错误处理： zipWith操作符具有短路特性。如果其中任何一个源Mono发出错误，那么zipWith返回的Mono也会立即发出相同的错误，而不会等待其他源完成。这对于快速失败和错误传播非常有用。
• 可读性和可维护性： 保持响应式链的流畅性，避免将异步操作拆分为多个独立的阻塞步骤，可以显著提高代码的可读性和可维护性。

总结

通过Mono.zipWith操作符，我们能够优雅且高效地在Project Reactor中聚合来自多个Flux流的异步结果，并将其封装成一个单一的Mono对象。这种模式是构建高性能、非阻塞响应式应用程序的关键，它确保了在处理并发数据源时，应用程序能够充分利用资源并保持出色的响应能力。理解并正确运用这些组合操作符，是掌握Reactor响应式编程范式的核心。

以上就是Reactor响应式编程：非阻塞地聚合两个Flux流的结果为单个Mono对象的详细内容，更多请关注其它相关文章！

# java  # seo理论教学视频  # 夏令营推广营销视频文案  # 江门seo标准  # 在这个  # 也会  # 就会  # 这是  # 并将  # 应用程序  # 第一个  # 转换为  # 这两个  # 多个  # 响应式编程  # ai  # 工具  # react  # 肺seo是什么病  # 淘宝引流seo教程  # 普宁网站优化软件有哪些  # 嘉兴网站推广多少钱  # 安徽互联网推广推荐网站  # 百度seo优化论文  # 三水整站seo优化 

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

相关推荐： QQ邮箱登录官网首页 腾讯QQ邮箱网页入口  12306选座如何查看座位示意图_12306座位示意图解读与使用  一加 14R 快充无反应_一加 14R 充电优化  一加手机电池耗电快怎么办_一加手机电池耗电快的解决方法  如何有效阻止外部脚本意外修改内联样式的高度属性  优化 Python 函数中的条件逻辑：解决 if-else 嵌套与参数选择问题  蛙漫限时开放最深处链接_蛙漫全站漫画会员同款秒开地址  搜狗浏览器如何使用密码生成器创建强密码 搜狗浏览器内置密码安全工具  c++如何实现单例设计模式_c++线程安全的单例模式写法  斑马英语APP如何开启夜间护眼阅读_斑马英语APP夜间模式与低蓝光设置教程  在VS Code中配置和运行Dart程序的完整步骤  谷歌学术网站直达地址 谷歌学术搜索网页版一键进入  为什么简单的XML文件也会解析失败？ 检查隐藏的非打印字符（如BOM）的方法  小米Civi 4录制视频过暗_小米Civi 4亮度优化  J*aScript中安全有效地处理localStorage字符串数据  微博网页版首页入口 微博电脑端官网登录链接  在Blazor WebAssembly应用中动态注入客户端特定指标代码的策略  俄罗斯浏览器官网直达链接 俄罗斯浏览器最新在线入口导航  C++如何操作大型数据集_使用C++流式处理(Streaming)技术避免一次性加载大文件  百度浏览器字体显示异常偏小_百度浏览器字体渲染修复方案  css滚动区域卡顿如何改善_css滚动问题用will-change优化渲染  冬*霸灯泡不亮怎么办_浴霸取暖灯一盏不亮的灯座清洁修复法  解决Bootstrap卡片顶部边距导致背景图下移的问题  Tabulator表格中精确实现日期时间排序的指南  优酷会员付费后没到账怎么办_优酷会员充值异常及解决方法  windows10怎么查看本机ip_windows10命令提示符ipconfig使用  消息称三星明年 2 月正式发布 HBM4，与 SK 海力士同台竞技  Go语言中Map值调用指针接收器方法的限制与应对  微信网页版扫码登录入口 微信网页版二维码登录入口  魅族17怎样用浏览器译外语网页_iPhone魅族17浏览器译外语网页【即时翻译】  Angular响应式表单：实现提交后表单及按钮的禁用与只读化  Excel如何用迷你图显趋势_Excel用迷你图显趋势【趋势小图】  Adobe PDF表单中利用J*aScript解析与格式化日期组件的教程  uc手机浏览器网页版入口 uc浏览器手机版便捷登录首页  React中useState与局部变量：理解组件状态管理与渲染机制  在J*a中如何捕获IndexOutOfBoundsException_索引越界异常防护方法说明  Odoo 16：在表单视图中基于当前记录动态修改Tree视图属性  神庙逃亡小游戏在线玩 神庙逃亡小游戏入口  在J*a中如何在J*a中使用异常机制记录错误日志_异常日志实践经验  AO3官方在线访问地址 Archive of Our Own最新镜像合集  vivo手机互传视频怎么操作_vivo手机互传视频详细传输方法  在J*a中如何隐藏复杂性_使用门面模式组织对象交互  随机参数递归函数的基准调用次数与时间复杂度探究  12306选座怎么选到商务座_12306商务座选择与配置说明  ArchiveofOurOwn小说阅读-ArchiveofOurOwn同人作品访问链接  TikTok评论显示延迟如何处理 TikTok评论刷新优化方法  QQ邮箱正确登录入口_QQ邮箱官方网站使用地址  微博网页版怎么开启两步验证_微博网页版账号安全两步验证设置方法  怎么去除衣服上的口红印_生活小妙招教你用酒精轻松擦除  解决深度学习模型训练初期异常高损失与完美验证准确率问题