如何在外部Flux中消费内部Flux元素并正确赋值？

Flux<A> outsideFlux = groupedFlux.map(element -> {

    // 将element转换为A对象的操作
    A a = ...;

    Flux<Double> insideFlux = someOtherCallThatReturnsThisFluxOfDouble;

    insideFlux.subscribe(val -> a.setVal(val));
    return a;
})

执行后发现a.setVal(val)并未生效，a的val属性仍为null。

Reactor中map是同步操作符，它会立即执行逻辑并返回结果对象。而insideFlux.subscribe()是异步触发的，map不会等待内部流执行完成就会返回创建好的A对象，此时val还没被赋值，自然是null。

需要用flatMap替代map——flatMap支持处理嵌套的异步流，会等待内部流执行完成后再向下游发射处理好的对象。根据内部流的发射情况，有几种常见处理方式：

场景1：内部流仅发射一个Double值 #

如果insideFlux只会返回一个值，用single()确保流中只有一个元素，完成赋值后返回A对象：

Flux<A> outsideFlux = groupedFlux.flatMap(element -> {
    // 转换element为A对象
    A a = convertElementToA(element);
    
    Flux<Double> insideFlux = someOtherCallThatReturnsThisFluxOfDouble;
    
    return insideFlux
            .single() // 确保流中仅有一个元素，否则抛出异常
            .doOnNext(val -> a.setVal(val))
            .thenReturn(a); // 赋值完成后返回A对象
});

场景2：内部流发射多个值，取最后一个 #

如果insideFlux会发射多个值，需要取最后一个值赋值：

Flux<A> outsideFlux = groupedFlux.flatMap(element -> {
    A a = convertElementToA(element);
    Flux<Double> insideFlux = someOtherCallThatReturnsThisFluxOfDouble;
    
    return insideFlux
            .last() // 取流的最后一个元素
            .doOnNext(val -> a.setVal(val))
            .thenReturn(a);
});

场景3：合并内部流的所有值 #

如果需要将内部流的所有Double值做聚合（比如求和、平均）后再赋值：

Flux<A> outsideFlux = groupedFlux.flatMap(element -> {
    A a = convertElementToA(element);
    Flux<Double> insideFlux = someOtherCallThatReturnsThisFluxOfDouble;
    
    return insideFlux
            .reduce(0.0, Double::sum) // 累加所有值，也可以用average()等聚合操作
            .doOnNext(total -> a.setVal(total))
            .thenReturn(a);
});

永远不要在map这类同步操作符中直接订阅异步流，而是用flatMap/concatMap（需要顺序执行时用）这类操作符来衔接嵌套流，确保外部流能等待内部流的异步操作完成后，再拿到已经完成赋值的对象。

内容的提问来源于stack exchange，提问作者Harsh Singh