spring webflux使用的详细代码

package com.congge.reactors; import org.junit.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; import java.time.Duration; import java.time.temporal.ChronoUnit; import java.util.Arrays; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Stream; public class ReactorApi { @Test public void fluxJust() { Flux<String> phones = Flux.just("小米", "三星", "华为"); } @Test public void fluxFromIterable() { Flux<String> phones = Flux.fromIterable(Arrays.asList("小米", "三星", "华为")); } @Test public void fluxFromArray() { Flux<String> phones = Flux.fromArray(new String[]{"小米", "三星", "华为"}); } @Test public void fluxFromStream() { Flux<String> phones = Flux.fromStream(Stream.of(new String[]{"小米", "三星", "华为"})); phones.subscribe(); phones.subscribe(); //只能被订阅一次 } @Test public void fluxEmpty() { Flux<String> phones = Flux.empty(); //generic type still honored } @Test public void fluxRange() { Flux<Integer> phones = Flux.range(5, 3); } @Test public void fluxGenerate() { Flux<Long> flux = Flux.generate( AtomicLong::new, (state, sink) -> { long i = state.getAndIncrement(); sink.next(i); if (i == 10) sink.complete(); return state; }, (state) -> System.out.println("done") ); flux.subscribe(System.out::println); } @Test public void fluxCreate() { Flux<String> phones = Flux.create((t) -> { t.next("小米"); t.next("三星"); t.next("华为"); t.complete(); }); phones.subscribe(System.out::println); System.out.println("------------"); Flux<String> ownFluxListener = Flux.create(sink -> { //传入自定义的方法 new MyDataListener(){ public void onReceiveData(String str){ sink.next(str); } public void onComplete(){ sink.complete(); } }; }, FluxSink.OverflowStrategy.DROP); ownFluxListener.subscribe(System.out::println); } public class MyDataListener{ public void onReceiveData(String str){ System.out.println("收到数据:"+str); } public void onComplete(){ System.out.println("完成数据的消费处理"); } } @Test public void fluxDefer() { Flux.defer(() -> Flux.just("小米", "三星", "华为")) .subscribe(System.out::println); Flux<String> stockSeq4 = Flux.defer(() -> Flux.fromStream(Stream.of(new String[]{"小米", "三星", "华为"}))); stockSeq4.subscribe(); stockSeq4.subscribe(); } @Test public void fluxInterval() throws InterruptedException { //interval 定时发送元素 Flux.interval(Duration.of(1, ChronoUnit.SECONDS)) .subscribe((t) -> System.out.println((String.valueOf(t)))); Thread.sleep(1000000); } }