序
本文主要研究一下flux的map与flatMap的区别
map
@Test
public void testMap() throws InterruptedException {
Flux.just(1,2,3,4)
.log()
.map(i -> {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return i * 2;
})
.subscribe(e -> LOGGER.info("get:{}",e));
}
这里头的map是纯元素转换
10:53:57.058 [main] INFO reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription) 10:53:57.062 [main] INFO reactor.Flux.Array.1 - | request(unbounded) 10:53:57.063 [main] INFO reactor.Flux.Array.1 - | onNext(1) 10:53:58.067 [main] INFO com.example.demo.FluxTest - get:2 10:53:58.067 [main] INFO reactor.Flux.Array.1 - | onNext(2) 10:53:59.071 [main] INFO com.example.demo.FluxTest - get:4 10:53:59.071 [main] INFO reactor.Flux.Array.1 - | onNext(3) 10:54:00.076 [main] INFO com.example.demo.FluxTest - get:6 10:54:00.076 [main] INFO reactor.Flux.Array.1 - | onNext(4) 10:54:01.080 [main] INFO com.example.demo.FluxTest - get:8 10:54:01.081 [main] INFO reactor.Flux.Array.1 - | onComplete()
flatMap
@Test
public void testFaltMap() throws InterruptedException {
Flux.just(1,4)
.log()
.flatMap(e -> {
return Flux.just(e*2).delayElements(Duration.ofSeconds(1));
})
.subscribe(e -> LOGGER.info("get:{}",e));
TimeUnit.SECONDS.sleep(10);
}
这里的flatMap，将元素转为Mono或Flux，转换操作里头还可以进行异步操作
10:54:01.088 [main] INFO reactor.Flux.Array.2 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription) 10:54:01.089 [main] INFO reactor.Flux.Array.2 - | request(256) 10:54:01.089 [main] INFO reactor.Flux.Array.2 - | onNext(1) 10:54:01.137 [main] INFO reactor.Flux.Array.2 - | onNext(2) 10:54:01.137 [main] INFO reactor.Flux.Array.2 - | onNext(3) 10:54:01.138 [main] INFO reactor.Flux.Array.2 - | onNext(4) 10:54:01.138 [main] INFO reactor.Flux.Array.2 - | onComplete() 10:54:02.143 [parallel-1] INFO com.example.demo.FluxTest - get:2 10:54:02.143 [parallel-1] INFO com.example.demo.FluxTest - get:4 10:54:02.143 [parallel-1] INFO com.example.demo.FluxTest - get:6 10:54:02.143 [parallel-1] INFO com.example.demo.FluxTest - get:8
小结
flatMap的转换Function要求返回一个Publisher，这个Publisher代表一个作用于元素的异步的转换操作；而map仅仅是同步的元素转换操作。
