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

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

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

// 获取成功的账户流
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();

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

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

        // 为了演示异步性，