Spring WebFlux：如何将Flux<T>合并至包含Collection<T>的类型——基于Flux<A>与Flux<B>的合并场景

Kotlin 实现方案 #

首先修正下你的数据类（B类需要加上val关键字），然后写合并逻辑：

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

// 定义数据类
data class A(val b: List<B>, val value_in_b: String)
data class B(val value_in_b: String)

fun mergeFluxes(fluxA: Flux<A>, fluxB: Flux<B>): Flux<A> {
    // 步骤1：将Flux<B>按value_in_b分组，生成键值对映射
    val groupedBs: Mono<Map<String, List<B>>> = fluxB
        .groupBy(B::value_in_b)  // 按关联键分组
        .flatMap { group -> 
            // 收集每个分组的B实例为列表，转成(key, list)的Pair
            group.collectList().map { group.key() to it } 
        }
        .collectMap({ it.first }, { it.second })  // 最终转为Map

    // 步骤2：将每个A与对应的B列表合并
    return fluxA.flatMap { a ->
        groupedBs.map { bMap ->
            // 获取匹配的B列表，无匹配则返回空列表
            val matchedBs = bMap.getOrDefault(a.value_in_b, emptyList())
            // 复制A实例，替换b字段为匹配到的Bs
            a.copy(b = matchedBs)
        }
    }
}

Java 实现方案 #

如果用Java的话，数据类可以用普通类或者Java 16+的record（更简洁），这里给出两种写法参考：

方案1：用普通类（兼容Java 8+）

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

// 定义数据类
class A {
    private final List<B> b;
    private final String value_in_b;

    // 基础构造器
    public A(List<B> b, String value_in_b) {
        this.b = b;
        this.value_in_b = value_in_b;
    }

    // 用于复制替换b列表的构造器
    public A(A original, List<B> newBs) {
        this.b = newBs;
        this.value_in_b = original.value_in_b;
    }

    // Getter方法
    public String getValue_in_b() { return value_in_b; }
    public List<B> getB() { return b; }
}

class B {
    private final String value_in_b;

    public B(String value_in_b) {
        this.value_in_b = value_in_b;
    }

    public String getValue_in_b() { return value_in_b; }
}

public class FluxMerger {
    public static Flux<A> mergeFluxes(Flux<A> fluxA, Flux<B> fluxB) {
        // 步骤1：分组Flux<B>为Map
        Mono<Map<String, List<B>>> groupedBs = fluxB
                .groupBy(B::getValue_in_b)
                .flatMap(group -> 
                    group.collectList().map(list -> Map.entry(group.key(), list))
                )
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        // 步骤2：合并A与对应的B列表
        return fluxA.flatMap(a ->
                groupedBs.map(bMap -> {
                    List<B> matchedBs = bMap.getOrDefault(a.getValue_in_b(), List.of());
                    return new A(a, matchedBs);
                })
        );
    }
}

方案2：用Java 16+ Record（更简洁）

把A和B换成record，省去构造器和Getter：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

// 用Record定义数据类
record A(List<B> b, String value_in_b) {}
record B(String value_in_b) {}

public class FluxMerger {
    public static Flux<A> mergeFluxes(Flux<A> fluxA, Flux<B> fluxB) {
        Mono<Map<String, List<B>>> groupedBs = fluxB
                .groupBy(B::value_in_b)
                .flatMap(group -> 
                    group.collectList().map(list -> Map.entry(group.key(), list))
                )
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        return fluxA.flatMap(a ->
                groupedBs.map(bMap -> {
                    List<B> matchedBs = bMap.getOrDefault(a.value_in_b(), List.of());
                    // Record的copy方法（Java 16+支持）
                    return a.withB(matchedBs);
                })
        );
    }
}

额外注意事项 #

• 顺序问题：如果需要保持Flux原有的顺序，把flatMap换成concatMap即可——不过concatMap是串行处理，性能会略低于flatMap，根据业务需求选择。
• 无限流场景：如果Flux是无限流（比如持续的消息流），上述方案不适用（因为collectMap会一直等待流结束），这时候需要用join或者窗口操作来做实时关联。
• 空值处理：如果value_in_b可能为null，建议提前过滤掉null的A/B实例，避免分组时出现异常。