序
本文主要研究下Flux的compose与flatMap操作
compose
Flux.just("1","2","3")
.log()
.compose(stringFlux -> {
LOGGER.info("compose executed");
return stringFlux.map(e -> e + "$");
}).subscribe(LOGGER::info);
输出如下：
20:59:15.692 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework 20:59:15.702 [main] INFO com.example.demo.FluxTest - compose executed 20:59:15.707 [main] INFO reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription) 20:59:15.709 [main] INFO reactor.Flux.Array.1 - | request(unbounded) 20:59:15.709 [main] INFO reactor.Flux.Array.1 - | onNext(1) 20:59:15.709 [main] INFO com.example.demo.FluxTest - 1$ 20:59:15.710 [main] INFO reactor.Flux.Array.1 - | onNext(2) 20:59:15.710 [main] INFO com.example.demo.FluxTest - 2$ 20:59:15.710 [main] INFO reactor.Flux.Array.1 - | onNext(3) 20:59:15.710 [main] INFO com.example.demo.FluxTest - 3$ 20:59:15.710 [main] INFO reactor.Flux.Array.1 - | onComplete()
可以看到compose executed只打印一次
flatMap
Flux.just("1","3")
.log()
.flatMap(s -> {
LOGGER.info("flatMap executed");
return Flux.just(s+"$");
}).subscribe(LOGGER::info);
输出如下：
21:00:54.497 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework 21:00:54.516 [main] INFO reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription) 21:00:54.519 [main] INFO reactor.Flux.Array.1 - | request(256) 21:00:54.520 [main] INFO reactor.Flux.Array.1 - | onNext(1) 21:00:54.520 [main] INFO com.example.demo.FluxTest - flatMap executed 21:00:54.520 [main] INFO com.example.demo.FluxTest - 1$ 21:00:54.520 [main] INFO reactor.Flux.Array.1 - | onNext(2) 21:00:54.520 [main] INFO com.example.demo.FluxTest - flatMap executed 21:00:54.520 [main] INFO com.example.demo.FluxTest - 2$ 21:00:54.520 [main] INFO reactor.Flux.Array.1 - | onNext(3) 21:00:54.520 [main] INFO com.example.demo.FluxTest - flatMap executed 21:00:54.520 [main] INFO com.example.demo.FluxTest - 3$ 21:00:54.521 [main] INFO reactor.Flux.Array.1 - | onComplete()
可以看到flatMap executed在每次onNext的时候都打印一次
小结
compose操作是将整个flux作为一个整体输入一次性转换，而flatMap是针对flux的每个item作为输入来进行转换。
