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

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

// 这是一个阻塞的、不推荐的做法
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());
}

package org.example;

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

import java.util.ArrayList;
import java.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());
    }
}