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 Main {

    public static Mono<Payments> getPaymentDataIncorrect() {
        Flux<Payments.SuccessAccount> accountsSucceeded = getAccountsSucceeded();
        Flux<Payments.FailedAccount> accountsFailed = getAccountsFailed();

        List<Payments.SuccessAccount> successAccounts = new ArrayList<>();
        List<Payments.FailedAccount> failedAccounts = new ArrayList<>();

        // 这种方式是阻塞的，因为它试图在异步操作完成前同步地填充列表
        accountsFailed.collectList().subscribe(failedAccounts::addAll);
        accountsSucceeded.collectList().subscribe(successAccounts::addAll);

        // 在此处，successAccounts和failedAccounts可能还未被填充
        return Mono.just(Payments.builder()
                .failedAccounts(failedAccounts)
                .successAccounts(successAccounts)
                .build());
    }
    // ... getAccountsSucceeded() 和 getAccountsFailed() 方法省略，与原始问题相同
}

package org.example;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;

public class Main {

    public static void main(String[] args) {
        getPaymentData().subscribe(System.out::println);
    }

    public static Mono<Payments> getPaymentData() {
        // 1. 获取两个独立的Flux流
        Flux<Payments.SuccessAccount> accountsSucceededFlux = getAccountsSucceeded();
        Flux<Payments.FailedAccount> accountsFailedFlux = getAccountsFailed();

        // 2. 将每个Flux转换为一个Mono<List>
        // collectList() 会收集Flux中的所有元素，并在Flux完成时发出一个包含这些元素的List
        Mono<List<Payments.FailedAccount>> failedAccountsMono = accountsFailedFlux.collectList();
        Mono<List<Payments.SuccessAccount>> successAccountsMono = accountsSucceededFlux.collectList();

        // 3. 使用 Mono.zipWith() 组合两个 Mono<List>
        // zipWith 会等待两个Mono都发出其结果，然后使用提供的BiFunction进行组合
        Mono<Payments> combined = failedAccountsMono.zipWith(
                successAccountsMono,
                (failedAccounts, successAccounts) -> Payments.builder()
                        .failedAccounts(failedAccounts)
                        .successAccounts(successAccounts)
                        .build()
        );
        return combined;
    }

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

    // 模拟获取失败账户的Flux流
    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());
    }
}