序
本文主要展示一下reactive streams的一些transform操作
mergeWith
@Test public void testMerge(){ Flux flux1 = Flux.interval(Duration.ofSeconds(1)) .take(3) .map(e -> "[flux1]:"+e); Flux mergeFlux = Flux.interval(Duration.ofSeconds(1)) .delayElements(Duration.ofSeconds(1)) .take(3) .map(e -> "[flux2]:"+e) .mergeWith(flux1); mergeFlux.subscribe(e -> { LOGGER.info("subscribe:{}",e); }); mergeFlux.blockLast(); } 输出实例
21:18:07.583 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework21:18:08.618 [parallel-2] INFO com.example.demo.TransformTest - subscribe:[flux1]:021:18:09.619 [parallel-2] INFO com.example.demo.TransformTest - subscribe:[flux1]:121:18:09.645 [parallel-6] INFO com.example.demo.TransformTest - subscribe:[flux2]:021:18:10.619 [parallel-2] INFO com.example.demo.TransformTest - subscribe:[flux1]:221:18:10.649 [parallel-8] INFO com.example.demo.TransformTest - subscribe:[flux2]:121:18:11.654 [parallel-2] INFO com.example.demo.TransformTest - subscribe:[flux2]:2
可以发现，他们是交叉合并的。
concatWith
@Test public void testConcat(){ Flux flux1 = Flux.interval(Duration.ofSeconds(1)) .take(3) .map(e -> "[flux1]:"+e); Flux concatFlux = Flux.interval(Duration.ofSeconds(1)) .delayElements(Duration.ofSeconds(1)) .take(3) .map(e -> "[flux2]:"+e) .concatWith(flux1); concatFlux.subscribe(e -> { LOGGER.info("subscribe:{}",e); }); concatFlux.blockLast(); } 输出
21:19:00.779 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework21:19:02.832 [parallel-4] INFO com.example.demo.TransformTest - subscribe:[flux2]:021:19:03.836 [parallel-6] INFO com.example.demo.TransformTest - subscribe:[flux2]:121:19:04.840 [parallel-8] INFO com.example.demo.TransformTest - subscribe:[flux2]:221:19:05.845 [parallel-2] INFO com.example.demo.TransformTest - subscribe:[flux1]:021:19:06.845 [parallel-2] INFO com.example.demo.TransformTest - subscribe:[flux1]:121:19:07.844 [parallel-2] INFO com.example.demo.TransformTest - subscribe:[flux1]:2
可以发现concatWith只是连接两个flux的数据，并不是按emit的顺序交叉来
zipWith
@Test public void testZip(){ List firstList = Lists.newArrayList("a","b","c","d","e","a","b"); List secondList = Lists.newArrayList("1","2","3","4","5"); Flux > zipFlux = Flux.fromIterable(firstList) .zipWith(Flux.fromIterable(secondList)); zipFlux.subscribe(e -> { LOGGER.info("subscribe:{}",e); }); } 输出如下
21:20:59.506 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework21:20:59.516 [main] INFO com.example.demo.TransformTest - subscribe:[a,1]21:20:59.517 [main] INFO com.example.demo.TransformTest - subscribe:[b,2]21:20:59.517 [main] INFO com.example.demo.TransformTest - subscribe:[c,3]21:20:59.517 [main] INFO com.example.demo.TransformTest - subscribe:[d,4]21:20:59.517 [main] INFO com.example.demo.TransformTest - subscribe:[e,5]
可以发现flux1相比flux2多余的数据没有被zip
flatMap
@Test public void testFlatMap(){ List secondList = Lists.newArrayList("1","2","3","4","5"); Flux flatMapFlux = Flux.fromIterable(secondList) .flatMap((str) ->{ return Mono.just(str).repeat(2).map(String::toUpperCase).delayElements(Duration.ofMillis(1)); }); flatMapFlux.subscribe(e -> { LOGGER.info("subscribe:{}",e); }); flatMapFlux.blockLast(); Flux mapFlux = Flux.fromIterable(secondList) .repeat(2) .map(String::toUpperCase); mapFlux.subscribe(e -> { LOGGER.info("map subscribe:{}",e); }); mapFlux.blockLast(); } 输出
21:33:46.904 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework21:33:46.958 [parallel-1] INFO com.example.demo.TransformTest - subscribe:121:33:46.959 [parallel-1] INFO com.example.demo.TransformTest - subscribe:221:33:46.959 [parallel-1] INFO com.example.demo.TransformTest - subscribe:321:33:46.960 [parallel-1] INFO com.example.demo.TransformTest - subscribe:421:33:46.960 [parallel-1] INFO com.example.demo.TransformTest - subscribe:521:33:46.960 [parallel-1] INFO com.example.demo.TransformTest - subscribe:221:33:46.960 [parallel-7] INFO com.example.demo.TransformTest - subscribe:321:33:46.960 [parallel-8] INFO com.example.demo.TransformTest - subscribe:421:33:46.960 [parallel-1] INFO com.example.demo.TransformTest - subscribe:521:33:46.961 [parallel-6] INFO com.example.demo.TransformTest - subscribe:121:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:121:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:221:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:321:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:421:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:521:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:121:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:221:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:321:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:421:33:46.963 [main] INFO com.example.demo.TransformTest - map subscribe:5
flatMap是异步的
reduce
@Test public void testReduce(){ List secondList = Lists.newArrayList("1","2","3","4","5"); Mono reduceMono = Flux.fromIterable(secondList) .flatMap(e -> Mono.just(e).map(item -> Integer.valueOf(item))) .reduce((total, e) -> total + e); reduceMono.subscribe(e -> { LOGGER.info("subscribe:{}",e); }); } 输出
21:36:29.978 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework21:36:30.014 [main] INFO com.example.demo.TransformTest - subscribe:15
groupBy
@Test public void testGroup(){ List firstList = Lists.newArrayList("a","b","c","d","e","a","b"); Flux > groupFlux = Flux.fromIterable(firstList) .map(String::toUpperCase) .groupBy(key -> key); groupFlux.subscribe(e -> { LOGGER.info("subscribe:{}",e.collectList().subscribe(item -> { LOGGER.info("item:{}",item); })); }); } 输出
21:37:00.912 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework21:37:00.949 [main] INFO com.example.demo.TransformTest - subscribe:reactor.core.publisher.LambdaMonoSubscriber@5faeada121:37:00.951 [main] INFO com.example.demo.TransformTest - subscribe:reactor.core.publisher.LambdaMonoSubscriber@1563da521:37:00.951 [main] INFO com.example.demo.TransformTest - subscribe:reactor.core.publisher.LambdaMonoSubscriber@2bbf4b8b21:37:00.951 [main] INFO com.example.demo.TransformTest - subscribe:reactor.core.publisher.LambdaMonoSubscriber@30a3107a21:37:00.951 [main] INFO com.example.demo.TransformTest - subscribe:reactor.core.publisher.LambdaMonoSubscriber@33c7e1bb21:37:00.951 [main] INFO com.example.demo.TransformTest - item:[A, A]21:37:00.952 [main] INFO com.example.demo.TransformTest - item:[B, B]21:37:00.952 [main] INFO com.example.demo.TransformTest - item:[C]21:37:00.952 [main] INFO com.example.demo.TransformTest - item:[D]21:37:00.952 [main] INFO com.example.demo.TransformTest - item:[E]
first
@Test public void testFirst(){ List firstList = Lists.newArrayList("a","b","c","d","e","a","b"); List secondList = Lists.newArrayList("1","2","3","4","5"); Flux firstFlux = Flux.fromIterable(firstList) .delayElements(Duration.ofMillis(200)); Flux secondFlux = Flux.fromIterable(secondList) .take(2); Flux result = Flux.first(firstFlux, secondFlux); result.subscribe(e -> { LOGGER.info("subscribe:{}",e); }); } toIterable
@Test public void testToIterable(){ List firstList = Lists.newArrayList("a","b","c","d","e","a","b"); Iterable itr = Flux.fromIterable(firstList) .map(String::toUpperCase) .toIterable(); itr.forEach(e -> LOGGER.info(e)); } 输出
21:39:35.031 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework21:39:35.045 [main] INFO com.example.demo.TransformTest - A21:39:35.045 [main] INFO com.example.demo.TransformTest - B21:39:35.045 [main] INFO com.example.demo.TransformTest - C21:39:35.045 [main] INFO com.example.demo.TransformTest - D21:39:35.045 [main] INFO com.example.demo.TransformTest - E21:39:35.045 [main] INFO com.example.demo.TransformTest - A21:39:35.045 [main] INFO com.example.demo.TransformTest - B
小结
reactive streams的操作相当于在jdk的streams的基础上实现了reactive化，可以参照着了解。