Reactor响应式例子

蔚落 2022-01-26 07:19 501阅读 0赞

package com.example.reactive;
    
    import java.io.IOException;
    import java.time.Duration;
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.Collections;
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Optional;
    import java.util.concurrent.Callable;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    import java.util.concurrent.atomic.AtomicBoolean;
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.concurrent.atomic.AtomicLong;
    import java.util.concurrent.atomic.LongAdder;
    import java.util.function.BiConsumer;
    import java.util.function.Consumer;
    import java.util.function.Function;
    import java.util.function.Supplier;
    import java.util.stream.Stream;
    
    import org.junit.Before;
    import org.junit.Test;
    import org.reactivestreams.Publisher;
    import org.reactivestreams.Subscription;
    
    import reactor.core.Disposable;
    import reactor.core.Exceptions;
    import reactor.core.publisher.BaseSubscriber;
    import reactor.core.publisher.ConnectableFlux;
    import reactor.core.publisher.Flux;
    import reactor.core.publisher.Mono;
    import reactor.core.publisher.SignalType;
    import reactor.core.publisher.UnicastProcessor;
    import reactor.core.scheduler.Schedulers;
    import reactor.test.StepVerifier;
    import reactor.test.StepVerifierOptions;
    import reactor.test.publisher.PublisherProbe;
    import reactor.util.context.Context;
    import reactor.util.function.Tuple2;
    
    /**
     * https://blog.csdn.net/weweeeeeeee/article/details/82885449
     * https://blog.51cto.com/liukang/2090191
     * 
     * Reactor中提供了非常丰富的操作符，除了以上几个常见的，还有：
     * 
     * 用于编程方式自定义生成数据流的create和generate等及其变体方法；
     * 用于“无副作用的peek”场景的doOnNext、doOnError、doOncomplete、doOnSubscribe、doOnCancel等及其变体方法；
     * 用于数据流转换的when、and/or、merge、concat、collect、count、repeat等及其变体方法；
     * 用于过滤/拣选的take、first、last、sample、skip、limitRequest等及其变体方法；
     * 用于错误处理的timeout、onErrorReturn、onErrorResume、doFinally、retryWhen等及其变体方法；
     * 用于分批的window、buffer、group等及其变体方法； 用于线程调度的publishOn和subscribeOn方法。
     * 使用这些操作符，你几乎可以搭建出能够进行任何业务需求的数据处理管道/流水线
     */
    public class ReactorReactiveMain {
    	ReactorReactiveMain reactor = null;
    
    	@Before
    	public void createThis() {
    		reactor = new ReactorReactiveMain();
    	}
    
    	@Test
    	public <T> void testExcutor() throws InterruptedException, ExecutionException {
    		String threadExcutor = reactor.<String>threadExcutor("hh");
    		System.out.println(threadExcutor);
    	}
    
    	public <T> T threadExcutor(String msg) throws InterruptedException, ExecutionException {
    		ExecutorService threadPool = Executors.newFixedThreadPool(8);
    
    		final List<T> batches = new ArrayList<T>();
    
    		Callable<T> t = new Callable<T>() { // *1Callable 分配 -- 可能导致 GC 压力。特别是当任务运行时间过长,每次 GC 暂停都会影响全局性能
    
    			public T call() {
    				synchronized (batches) { // *2同步过程强制每个线程执行停-检查操作。
    					T result = callDatabase(msg); // *3消息的消费可能比生产慢。,积压虽然不会直接导致内存泄漏，但会带来严重副作用：GC 暂停时要扫描更多的对象；有丢失重要数据位的风险；
    					batches.add(result);
    					return result;
    				}
    			}
    
    			@SuppressWarnings("unchecked")
    			private T callDatabase(String msg) {
    				return (T) msg;
    			}
    		};
    
    		Future<T> f = threadPool.submit(t); // *4使用线程池(ThreadPool)将任务传递给目标线程 -- 通过 FutureTask 方式肯定会产生 GC 压力。
    		T result = f.get();// *5阻塞直至 callDatabase()
    							// 回调。阻塞回调容易产生恶性循环。阻塞回调会降低消息生产者的效率。在实践中，任务提交后需要等待结果返回，此时流式过程几乎演变为同步的了
    
    		return result;
    	}
    
    	// 2.X
    	// public static void main(String[] args) throws InterruptedException {
    	// Environment env = Environment.initialize();
    	// Processor<String, String> p = RingBufferProcessor.create("testProcessor",
    	// 32);
    	// Stream<String> s1 = Streams.wrap(p);
    	// s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev));
    	// s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev));
    	// s1.consume(ev -> System.out.println(Thread.currentThread() + " data=" + ev));
    	// p.onNext("One");
    	// p.onNext("Two");
    	// p.onNext("Three");
    	// p.onComplete();
    	//
    	// Environment.terminate();
    	// }
    
    	public void test() {
    		Consumer<String> consumer = new Consumer<String>() {
    			@Override
    			public void accept(String value) {
    				System.out.println(value);
    			}
    		};
    
    		// 为了简约，现在用 Java 8 风格
    		Function<Integer, String> transformation = integer -> "" + integer;
    
    		Supplier<Integer> supplier = () -> 123;
    
    		BiConsumer<Consumer<String>, String> biConsumer = (callback, value) -> {
    			for (int i = 0; i < 10; i++) {
    				// 对要运行的最后逻辑运行做惰性求值
    				callback.accept(value);
    			}
    		};
    
    		// 注意生产者到双向消费者执行过程
    		biConsumer.accept(consumer, transformation.apply(supplier.get()));
    
    		Consumer<Tuple2<Consumer<String>, String>> biConsumer2 = tuple -> {
    			for (int i = 0; i < 10; i++) {
    				// 类型正确，开启编译器
    				tuple.getT1().accept(tuple.getT2());
    			}
    		};
    	}
    
    	// ---------------------------简单的创建和订阅 Flux 或 Mono
    	// 的方法---------------------------
    	@Test
    	public void testCreateSe() {
    		// 简单的创建和订阅 Flux 或 Mono 的方法
    		// 最简单的上手 Flux 和 Mono 的方式就是使用相应类提供的多种工厂方法之一。
    		// 比如，如果要创建一个 String 的序列，你可以直接列举它们，或者将它们放到一个集合里然后用来创建 Flux，如下：
    		Flux<String> seq1 = Flux.just("foo", "bar", "foobar");
    		List<String> iterable = Arrays.asList("foo", "bar", "foobar");
    		Flux<String> seq2 = Flux.fromIterable(iterable);
    		seq1.subscribe(System.out::println);
    		seq2.subscribe(System.out::println);
    
    		Mono<String> noData = Mono.empty(); // 即使没有值，工厂方法仍然采用通用的返回类型
    		Mono<String> data = Mono.just("foo");
    		Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3); // 第一个参数是 range 的开始，第二个参数是要生成的元素个数
    		noData.subscribe(System.out::println);
    		data.subscribe(System.out::println);
    		numbersFromFiveToSeven.subscribe(System.out::println);
    	}
    
    	@Test
    	public void testPubSub() {
    		// 在订阅（subscribe）的时候，Flux 和 Mono 使用 Java 8 lambda 表达式。 .subscribe()
    		// 方法有多种不同的方法签名，你可以传入各种不同的 lambda 形式的参数来定义回调
    		Flux<Integer> ints = Flux.range(1, 3);
    		ints.subscribe();// 最简单的订阅方式
    
    		ints.subscribe(i -> System.out.println(i)); // 订阅并消费
    		ints.subscribe(System.out::println); // 订阅并消费
    	}
    
    	/**
    	 * subscribe();
    	 * 
    	 * subscribe(Consumer<? super T> consumer);
    	 * 
    	 * subscribe(Consumer<? super T> consumer, Consumer<? super Throwable>
    	 * errorConsumer);
    	 * 
    	 * subscribe(Consumer<? super T> consumer, Consumer<? super Throwable>
    	 * errorConsumer, Runnable completeConsumer);
    	 * 
    	 * subscribe(Consumer<? super T> consumer, Consumer<? super Throwable>
    	 * errorConsumer, Runnable completeConsumer, Consumer<? super Subscription>
    	 * subscriptionConsumer);
    	 */
    	// 为了演示下一个方法签名，我们故意引入一个错误
    	@Test
    	public void testPubSubOnError() {
    		Flux<Integer> ints = Flux.range(1, 4).map(i -> {
    			if (i <= 3) {
    				return i;
    			}
    			throw new RuntimeException("Got to 4");
    		});
    		// 现在我们有两个 lambda 表达式：一个是用来处理正常数据，一个用来处理错误
    		ints.subscribe(i -> System.out.println(i), error -> System.err.println("Error: " + error));
    	}
    
    	@Test
    	public void testPubSubOnErrorAndCallBack() {
    		Flux<Integer> ints = Flux.range(1, 4);
    		// 错误和完成信号都是终止信号，并且二者只会出现其中之一。为了能够最终全部正常完成，你必须处理错误信号。
    		// 用于处理完成信号的 lambda 是一对空的括号，因为它实际上匹配的是 Runalbe 接口中的 run 方法， 不接受参数
    		ints.subscribe(i -> System.out.println(i), error -> System.err.println("Error " + error), () -> {
    			System.out.println("Done");
    		});
    	}
    
    	// 最后一个 subscribe 方法签名包含一个自定义的 subscriber
    	@Test
    	public void testPubSubCustomizeSub() {
    		SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
    		Flux<Integer> ints = Flux.range(1, 4);
    		ints.subscribe(i -> System.out.println(i), error -> System.err.println("Error " + error), () -> {
    			System.out.println("Done");
    		}, s -> ss.request(10));// 自定义的 Subscriber 作为 subscribe 方法的最后一个参数
    		ints.subscribe(ss);
    	}
    
    	/**
    	 * 响应式流规范定义了另一个 subscribe 方法的签名，它只接收一个自定义的 Subscriber， 没有其他的参数
    	 * subscribe(Subscriber<? super T> subscriber);
    	 * 
    	 * 如果你已经有一个 Subscriber，那么这个方法签名还是挺有用的。况且，你可能还会用到它
    	 * 来做一些订阅相关（subscription-related）的回调。比如，你想要自定义“背压（backpressure）” 并且自己来触发请求。
    	 * 
    	 * 在这种情况下，使用 BaseSubscriber 抽象类就很方便，因为它提供了很好的配置“背压” 的方法
    	 */
    	@Test
    	public void testSubscriber() {
    		Flux<String> source = Flux.just("a", "b", "c");// someStringSource();
    
    		source.map(String::toUpperCase).subscribe(new BaseSubscriber<String>() { // BaseSubscriber 是一个抽象类，所以我们创建一个匿名内部类
    			@Override
    			protected void hookOnSubscribe(Subscription subscription) {
    				// request(n) 就是这样一个方法。它能够在任何 hook 中，通过 subscription 向上游传递
    				// 背压请求。这里我们在开始这个流的时候请求1个元素值。
    				request(1);
    			}
    
    			@Override
    			protected void hookOnNext(String value) {
    				// 随着接收到新的值，我们继续以每次请求一个元素的节奏从源头请求值
    				request(1);
    			}
    			// BaseSubscriber 还提供了 requestUnbounded() 方法来切换到“无限”模式（等同于
    			// request(Long.MAX_VALUE)）。
    		});
    	}
    
    	// ---------------------------可编程式地创建一个序列---------------------------
    	/**
    	 * 通过定义相对应的事件（onNext、onError`和`onComplete） 创建一个 Flux 或 Mono。所有这些方法都通过 API
    	 * 来触发我们叫做 sink（池） 的事件
    	 * 
    	 * 最简单的创建 Flux 的方式就是使用 generate 方法, 这是一种 同步地， 逐个地 产生值的方法意味着 sink 是一个
    	 * SynchronousSink 而且其 next() 方法在每次回调的时候最多只能被调用一次。你也可以调用 error(Throwable) 或者
    	 * complete()，不过是可选的。
    	 * 
    	 * 最有用的一种方式就是同时能够记录一个状态值（state），从而在使用 sink 发出下一个元素的时候能够
    	 * 基于这个状态值去产生元素。此时生成器（generator）方法就是一个 BiFunction<S, SynchronousSink<T>, S>， 其中
    	 * <S> 是状态对象的类型。你需要提供一个 Supplier<S> 来初始化状态值，而生成器需要 在每一“回合”生成元素后返回新的状态值（供下一回合使用）。
    	 */
    	@Test
    	public void testGenerateState() {
    		// 基于状态值的 generate 示例
    		// 初始化状态值（state）为0。
    		Flux<String> flux = Flux.generate(() -> 0, (state, sink) -> {
    			// 我们基于状态值 state 来生成下一个值（state 乘以 3）。
    			sink.next("3 x " + state + " = " + 3 * state);
    			if (state == 10) {
    				// 我们也可以用状态值来决定什么时候终止序列。
    				sink.complete();
    			}
    			return state + 1; // 返回一个新的状态值 state，用于下一次调用。
    		});
    		// 面的代码生成了“3 x”的乘法表：在subscribe订阅之前，数据是不流动的
    		flux.subscribe(System.out::println);
    	}
    
    	/**
    	 * 也可以使用可变（mutable）类型（译者注：如上例，原生类型及其包装类，以及String等属于不可变类型） 的 <S>。上边的例子也可以用
    	 * AtomicLong 作为状态值，在每次生成后改变它的值
    	 */
    	@Test
    	public void testGenerateStateWithMutable() {
    		// 这次我们初始化一个可变类型的状态值。第一个参数：Callable<AtomicLong>
    		// stateSupplier、Callable就是一个无入有出的run方法，也就是一个Supplier【提供器】
    		Flux<String> flux = Flux.generate(AtomicLong::new, (state, sink) -> {
    			long i = state.getAndIncrement();
    			sink.next("3 x " + i + " = " + 3 * i);
    			if (i == 10) {
    				sink.complete();
    			}
    			return state;
    		});
    		flux.subscribe(System.out::println);
    	}
    
    	// 如果状态对象需要清理资源，可以使用 generate(Supplier<S>, BiFunction, Consumer<S>)
    	// 这个签名方法来清理状态对象（译者注：Comsumer 在序列终止才被调用）。
    	@Test
    	public void testGenerateStateWithMutableAndThen() {
    		Flux<String> flux = Flux.generate(AtomicLong::new, (state, sink) -> {
    			long i = state.getAndIncrement();
    			sink.next("3 x " + i + " = " + 3 * i);
    			if (i == 10) {
    				sink.complete();
    			}
    			return state;
    			// 如果 state 使用了数据库连接或者其他需要最终进行清理的资源，这个 Consumer lambda 可以用来在最后关闭连接或完成相关的其他清理任务。
    		}, (state) -> System.out.println("state: " + state));
    
    		flux.subscribe(System.out::println);
    	}
    
    	/**
    	 * 作为一个更高级的创建 Flux 的方式， create 方法的生成方式既可以是同步， 也可以是异步的，并且还可以每次发出多个元素。
    	 * 
    	 * 该方法用到了 FluxSink，后者同样提供 next，error 和 complete 等方法。 与 generate 不同的是，create
    	 * 不需要状态值，另一方面，它可以在回调中触发 多个事件（即使是在未来的某个时间）。
    	 * 
    	 * create 有个好处就是可以将现有的 API 转为响应式，比如监听器的异步方法。
    	 * 
    	 * create 可以是异步地，并且能够控制背压，你可以通过提供一个 OverflowStrategy 来定义背压行为。
    	 * 
    	 * IGNORE： 完全忽略下游背压请求，这可能会在下游队列积满的时候导致 IllegalStateException。
    	 * 
    	 * ERROR： 当下游跟不上节奏的时候发出一个 IllegalStateException 的错误信号。
    	 * 
    	 * DROP：当下游没有准备好接收新的元素的时候抛弃这个元素。
    	 * 
    	 * LATEST：让下游只得到上游最新的元素。
    	 * 
    	 * BUFFER：（默认的）缓存所有下游没有来得及处理的元素（这个不限大小的缓存可能导致 OutOfMemoryError）
    	 * 
    	 * Mono 也有一个用于 create 的生成器（generator）—— MonoSink，它不能生成多个元素， 因此会抛弃第一个元素之后的所有元素。
    	 * 
    	 * create 的一个变体是 push，适合生成事件流。与 create`类似，`push 也可以是异步地， 并且能够使用以上各种溢出策略（overflow
    	 * strategies）管理背压。每次只有一个生成线程可以调用 next，complete 或 error
    	 */
    
    	// 可以使用 create 方法将其转化为响应式类型 Flux<T>：
    	// @Test
    	// public void testFluxCreateMethod() {
    	// Flux<String> bridge = Flux.create(sink -> {
    	// myEventProcessor.register(new MyEventListener<String>() {
    	// public void onDataChunk(List<String> chunk) {
    	// for (String s : chunk) {
    	// sink.next(s);
    	// }
    	// }
    	// public void processComplete() {
    	// sink.complete();
    	// }
    	// });
    	// });
    	// }
    
    	// handle 方法有些不同，它在 Mono 和 Flux 中都有。然而，它是一个实例方法 （instance
    	// method），意思就是它要链接在一个现有的源后使用（与其他操作符一样）。
    	// 下边的方法可以用于 Integer 序列，映射为字母或 null
    	public String alphabet(int letterNumber) {
    		if (letterNumber < 1 || letterNumber > 26) {
    			return null;
    		}
    		int letterIndexAscii = 'A' + letterNumber - 1;
    		return "" + (char) letterIndexAscii;
    	}
    
    	// 响应式流规范允许 null 这样的值出现在序列中。将 handle 用于一个 "映射 + 过滤 null" 的场景,可以把它当做 map 与 filter
    	// 的组合
    	@Test
    	public void testFluxHandle() {
    		Flux<String> alphabet = Flux.just(-1, 30, 13, 9, 20).handle((i, sink) -> {
    			String letter = alphabet(i); // 等价map操作
    			if (letter != null) // filter操作
    				sink.next(letter); // 如果返回的是 null, 就不会调用 sink.next 从而过滤掉
    		});
    
    		alphabet.subscribe(System.out::println);
    	}
    
    	/**
    	 * 当前线程（Schedulers.immediate()）
    	 * 
    	 * 可重用的单线程（Schedulers.single()）。注意，这个方法对所有调用者都提供同一个线程来使用，
    	 * 直到该调度器（Scheduler）被废弃。如果你想使用专一的线程，就对每一个调用使用 Schedulers.newSingle()。
    	 * 
    	 * 弹性线程池（Schedulers.elastic()。它根据需要创建一个线程池，重用空闲线程。线程池如果空闲时间过长 （默认为 60s）就会被废弃。对于
    	 * I/O 阻塞的场景比较适用。 Schedulers.elastic() 能够方便地给一个阻塞 的任务分配它自己的线程，从而不会妨碍其他任务和资源，见
    	 * 如何包装一个同步阻塞的调用？。
    	 * 
    	 * 固定大小线程池（Schedulers.parallel()）。所创建线程池的大小与 CPU 个数等同。
    	 * 
    	 * 此外，你还可以使用 Schedulers.fromExecutorService(ExecutorService) 基于现有的
    	 * ExecutorService 创建 Scheduler。（虽然不太建议，不过你也可以使用 Executor 来创建）。你也可以使用 newXXX
    	 * 方法来创建不同的调度器。比如 Schedulers.newElastic(yourScheduleName) 创建一个新的名为
    	 * yourScheduleName 的弹性调度器。
    	 * 
    	 * 一些操作符默认会使用一个指定的调度器（通常也允许开发者调整为其他调度器）例如， 通过工厂方法
    	 * Flux.interval(Duration.ofMillis(300)) 生成的每 300ms 打点一次的 Flux<Long>， 默认情况下使用的是
    	 * Schedulers.parallel()，下边的代码演示了如何将其装换为 Schedulers.single()：Schedulers工具类
    	 * 
    	 * 记住，在你订阅（subscribe）前，你只是定义了处理流程，而没有启动发布者.基于此，Reactor
    	 * 可以使用这些规则来决定如何执行操作链。然后，一旦你订阅了，整个流程就开始工作了。
    	 */
    	@Test
    	public void switchDefaultScheduler() {
    		Flux.interval(Duration.ofMillis(300), Schedulers.newSingle("test"));
    
    		// 支持任务共享的多线程模型：创建一个有 10,000 个元素的 Flux,创建等同于 CPU 个数的线程（最小为4）,subscribe()
    		// 之前什么都不会发生. 当 publishOn 请求元素的时候，会从任一个正在发出元素的线程那里获取元素
    		Flux.range(1, 10000).publishOn(Schedulers.parallel()).subscribe(System.out::println);
    	}
    
    	/**
    	 * 在响应式流中，错误（error）是终止（terminal）事件。当有错误发生时，它会导致流序列停止， 并且错误信号会沿着操作链条向下传递，直至遇到你定义的
    	 * Subscriber 及其 onError 方法。
    	 * 
    	 * 如果没有定义，onError 会抛出 UnsupportedOperationException。你可以接下来再 检测错误，并通过
    	 * Exceptions.isErrorCallbackNotImplemented 方法捕获和处理它。
    	 * 
    	 * 牢记 响应式流中的任何错误都是一个终止事件。 即使用了错误处理操作符，也不会让源头流序列继续。而是将 onError 信号转化为一个 新的 序列
    	 * 的开始。换句话说，它代替了被终结的 上游 流序列
    	 * 
    	 * 你也许熟悉在 try-catch 代码块中处理异常的几种方法。常见的包括如下几种：
    	 * 
    	 * 捕获并返回一个静态的缺省值。
    	 * 
    	 * 捕获并执行一个异常处理方法。
    	 * 
    	 * 捕获并动态计算一个候补值来顶替。
    	 * 
    	 * 捕获，并再包装为某一个 业务相关的异常，然后再抛出业务异常。
    	 * 
    	 * 捕获，记录错误日志，然后继续抛出。
    	 * 
    	 * 使用 finally 来清理资源，或使用 Java 7 引入的 "try-with-resource"。 以上所有这些在 Reactor 都有相应的基于
    	 * error-handling 操作符处理方式
    	 */
    	/**
    	 * 在开始研究这些操作符之前，我们先准备好响应式链（reactive chain）方式和 try-catch 代码块方式（以便对比）。
    	 * 当订阅的时候，位于链结尾的 onError 回调方法和 catch 块类似，一旦有异常，执行过程会跳入到 catch：
    	 * 
    	 * @throws InterruptedException
    	 */
    	@Test
    	public void testExpInReactor() throws InterruptedException {
    		Flux<String> s = Flux.range(1, 10)
    				// 执行 map 转换，有可能抛出异常。
    				.map(v -> doSomethingDangerous(v))
    				// 如果没问题，执行第二个 map 转换操作。
    				.map(v -> doSecondTransform(v));
    		// 所有转换成功的值都打印出来。
    		s.subscribe(value -> System.out.println("RECEIVED " + value)
    		// 一旦有错误，序列（sequence）终止，并打印错误信息。
    				, error -> System.err.println("CAUGHT " + error));
    
    		// 这与 try/catch 代码块是类似的：
    		try {
    			for (int i = 1; i < 11; i++) {
    				String v1 = (String) doSomethingDangerous(i);
    				String v2 = doSecondTransform(v1);
    				System.out.println("RECEIVED " + v2);
    			}
    		} catch (Throwable t) {
    			System.err.println("CAUGHT " + t);
    		}
    
    		// 既然我们准备了两种方式做对比，我们就来看一下不同的错误处理场景，以及相应的操作符
    		// 与第 (1) 条（捕获并返回一个静态的缺省值）对应的是 onErrorReturn：
    		Flux.just(10).map(this::doSomethingDangerous).onErrorReturn("RECOVERED");
    
    		// 你还可以通过判断错误信息的内容，来筛选哪些要给出缺省值，哪些仍然让错误继续传递下去：
    		Flux.just(10).map(this::doSomethingDangerous).onErrorReturn(e -> e.getMessage().equals("boom10"),
    				"recovered10");
    
    		// 异常处理方法,如果你不只是想要在发生错误的时候给出缺省值，而是希望提供一种更安全的处理数据的方式， 可以使用 onErrorResume。这与第 (2)
    		// 条（捕获并执行一个异常处理方法）类似。
    		// 假设，你会尝试从一个外部的不稳定服务获取数据，但仍然会在本地缓存一份 可能 有些过期的数据， 因为缓存的读取更加可靠。可以这样来做：
    		Flux.just("key1", "key2").flatMap(
    				// 对于每一个 key， 异步地调用一个外部服务。
    				k -> callExternalService(k))
    				// 如果对外部服务的调用失败，则再去缓存中查找该 key。注意，这里无论 e 是什么，都会执行异常处理方法。
    				.onErrorResume(e -> {
    					return null;
    				});// e -> getFromCache(k));
    
    		// 就像 onErrorReturn，onErrorResume 也有可以用于预先过滤错误内容的方法变体，可以基于异常类或 Predicate
    		// 进行过滤。它实际上是用一个 Function 来作为参数，还可以返回一个新的流序列。
    		Flux.just("timeout1", "unknown", "key2").flatMap(k -> callExternalService(k))
    				// 这个函数式允许开发者自行决定如何处理。
    				.onErrorResume(error -> {
    					// 如果源超时，使用本地缓存。
    					if (error instanceof TimeoutException) {
    						// return getFromCache(k);
    						return null;
    					}
    					// 如果源找不到对应的 key，创建一个新的实体。
    					else if (error instanceof UnknownKeyException) {
    						// return registerNewEntry(k, "DEFAULT");
    						return null;
    					} else {
    						// 否则， 将问题“重新抛出”。
    						return Flux.error(error);
    					}
    				});
    
    		// 动态候补值,有时候并不想提供一个错误处理方法，而是想在接收到错误的时候计算一个候补的值。这类似于第 (3) 条（捕获并动态计算一个候补值）。
    		// 例如，如果你的返回类型本身就有可能包装有异常（比如 Future.complete(T success)
    		// vsFuture.completeExceptionally(Throwable error)），你有可能使用流中的错误包装起来实例化 返回值。
    		// 这也可以使用上一种错误处理方法的方式（使用 onErrorResume）解决，代码如下：
    		// Flux.range(1, 10).onErrorResume(
    		// 在 onErrorResume 中，使用 Mono.just 创建一个 Mono。
    		// error -> Mono.just(
    		// 将异常包装到另一个类中
    		// myWrapper.fromError(error)));
    
    		// 捕获并重新抛出,在“错误处理方法”的例子中，基于 flatMap 方法的最后一行，我们可以猜到如何做到第 (4)
    		// 条（捕获，包装到一个业务相关的异常，然后抛出业务异常）：
    		// Flux.just("timeout1")
    		// .flatMap(k -> callExternalService(k))
    		// .onErrorResume(original -> Flux.error(
    		// new BusinessException("oops, SLA exceeded", original);
    		// );
    
    		// 然而还有一个更加直接的方法—— onErrorMap：
    		// Flux.just("timeout1")
    		// .flatMap(k -> callExternalService(k))
    		// .onErrorMap(original -> new BusinessException("oops, SLA exceeded",
    		// original));
    
    		// 记录错误日志
    		LongAdder failureStat = new LongAdder();
    		Flux<String> flux = Flux.just("unknown").flatMap(k -> callExternalService(k)).doOnError(e -> {
    			failureStat.increment();
    			log("uh oh, falling back, service failed for key " + e);// k);
    		}).onErrorResume(e -> {
    			return null;
    		});// e -> getFromCache(k));
    
    		// 使用资源和 try-catch 代码块
    		AtomicBoolean isDisposed = new AtomicBoolean();
    		Disposable disposableInstance = new Disposable() {
    			@Override
    			public void dispose() {
    				isDisposed.set(true);
    			}
    
    			@Override
    			public String toString() {
    				return "DISPOSABLE";
    			}
    		};
    
    		Flux<String> flux2 = Flux.using(
    				// 第一个 lambda 生成资源，这里我们返回模拟的（mock） Disposable。
    				() -> disposableInstance,
    				// 第二个 lambda 处理资源，返回一个 Flux<T>。
    				disposable -> Flux.just(disposable.toString()),
    				// /第三个 lambda 在 2) 中的资源 Flux 终止或取消的时候，用于清理资源。
    				Disposable::dispose);// 在订阅或执行流序列之后， isDisposed 会置为 true。
    
    		LongAdder statsCancel = new LongAdder();
    
    		// doFinally 在序列终止（无论是 onComplete、`onError`还是取消）的时候被执行
    		// 并且能够判断是什么类型的终止事件（完成、错误还是取消？）
    		Flux<String> flux3 = Flux.just("foo", "bar").doFinally(type -> {
    			if (type == SignalType.CANCEL)
    				statsCancel.increment();
    		}).take(1);
    
    		// 演示终止方法 onError
    		Flux<String> flux4 = Flux.interval(Duration.ofMillis(250)).map(input -> {
    			if (input < 3)
    				return "tick " + input;
    			throw new RuntimeException("boom");
    		}).onErrorReturn("Uh oh");
    		flux.subscribe(System.out::println);
    		// 注意 interval 默认基于一个 timer Scheduler 来执行。 如果我们想在 main 方法中运行， 我们需要调用
    		// sleep，这样程序就可以在还没有产生任何值的时候就退出了
    		Thread.sleep(2100);
    
    		// 重试
    		Flux.interval(Duration.ofMillis(250)).map(input -> {
    			if (input < 3)
    				return "tick " + input;
    			throw new RuntimeException("boom");
    			// elapsed 会关联从当前值与上个值发出的时间间隔
    		}).elapsed()
    				// retry(1) 不过是再一次从新订阅了原始的 interval
    				.retry(1).subscribe(System.out::println, System.err::println);
    		Thread.sleep(2100);
    
    		// 还有一个“高配版”的 retry （retryWhen），它使用一个伴随（"companion"） Flux 来判断对某次错误是否要重试。这个伴随
    		// Flux 是由操作符创建的，但是由开发者包装它， 从而实现对重试操作的配置
    		// 这个伴随 Flux 是一个 Flux<Throwable>，它作为 retryWhen 的唯一参数被传递给一个 Function，你可以定义这个
    		// Function 并让它返回一个新的 Publisher<?>。重试的循环 会这样运行：
    		// 每次出现错误，错误信号会发送给伴随 Flux，后者已经被你用 Function 包装。
    		// 如果伴随 Flux 发出元素，就会触发重试。
    		// 如果伴随 Flux 完成（complete），重试循环也会停止，并且原始序列也会 完成（complete）。
    		// 如果伴随 Flux 产生一个错误，重试循环停止，原始序列也停止 或 完成，并且这个错误会导致 原始序列失败并终止。
    		// 了解前两个场景的区别是很重要的。如果让伴随 Flux 完成（complete）等于吞掉了错误。如下代码用 retryWhen 模仿了 retry(3)
    		// 的效果：
    		Flux<String> flux5 =
    				// 持续产生错误
    				Flux.<String>error(new IllegalArgumentException())
    						// 在 retry 之前 的 doOnError 可以让我们看到错误。
    						.doOnError(System.out::println)
    						// 这里，我们认为前 3 个错误是可以重试的（take(3)），再有错误就放弃。
    						.retryWhen(companion -> companion.take(3));
    
    		// 事实上，上边例子最终得到的是一个 空的 Flux，但是却 成功 完成了。反观对同一个 Flux 调用 retry(3) 的话，最终是以最后一个 error
    		// 终止 Flux，故而 retryWhen 与之不同。
    
    		// 实现同样的效果需要一些额外的技巧：
    		Flux<String> flux6 = Flux.<String>error(new IllegalArgumentException())
    				.retryWhen(companion -> companion.zipWith(Flux.range(1, 4), (error, index) -> {
    					if (index < 4)
    						return index;
    					else
    						// Exceptions.propagate 方法来包装异常
    						throw Exceptions.propagate(error);
    				}));
    
    		// 在操作符或函数式中处理异常,Lambda表达式不能用s ->[Lambda expression's parameter s cannot
    		// redeclare another local variable defined in an enclosing scope. ]
    		Flux.just("foo").map(str -> {
    			throw new IllegalArgumentException(str);
    		}).subscribe(v -> System.out.println("GOT VALUE"), e -> System.out.println("ERROR: " + e));
    
    	}
    
    	public String convert(int i) throws IOException {
    		if (i > 3) {
    			throw new IOException("boom " + i);
    		}
    		return "OK " + i;
    	}
    
    	@Test
    	public void testExpsUtil() {
    		Flux<String> converted = Flux.range(1, 10).map(i -> {
    			try {
    				return convert(i);
    			} catch (IOException e) {
    				// 它同样会首先调用 throwIfFatal， 并且不会包装 RuntimeException。
    				throw Exceptions.propagate(e);
    			}
    		});
    
    		// 当后边订阅上边的这个 Flux 并响应错误（比如在用户界面）的时候，如果你想处理 IOException， 你还可以再将其转换为原始的异常
    		converted.subscribe(v -> System.out.println("RECEIVED: " + v), e -> {
    			// 使用 Exceptions.unwrap 方法来得到原始的未包装的异常（追溯最初的异常）。
    			if (Exceptions.unwrap(e) instanceof IOException) {
    				System.out.println("Something bad happened with I/O");
    			} else {
    				System.out.println("Something bad happened");
    			}
    		});
    	}
    
    	// 使用 transform 操作符
    	// transform可以将一段操作链打包为一个函数式。这个函数式能在组装期将被封装的操作符还原并接入到调用transform的位置。这样做和直接将被封装的操作符加入到链上的效果是一样的
    	@Test
    	public void testTransform() {
    		// 通过名为filterAndMap的函数式将filter和map操作符进行了打包，然后交给transform拼装到操作链中
    		Function<Flux<String>, Flux<String>> filterAndMap = f -> f.filter(color -> !color.equals("orange"))
    				.map(String::toUpperCase);
    
    		Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple")).doOnNext(System.out::println)
    				.transform(filterAndMap)
    				.subscribe(d -> System.out.println("Subscriber to Transformed MapAndFilter: " + d));
    	}
    
    	// 使用 compose 操作符
    	// compose 操作符与 transform
    	// 类似，也能够将几个操作符封装到一个函数式中。主要的区别就是，这个函数式是针对每一个订阅者起作用的。这意味着它对每一个 subscription
    	// 可以生成不同的操作链
    	public void testCompose() {
    		AtomicInteger ai = new AtomicInteger();
    		// filterAndMap函数式有一个名为ai的会自增的状态值。每次调用subscribe方法进行订阅的时候，compose会导致ai自增，从而两次订阅的操作链是不同的
    		Function<Flux<String>, Flux<String>> filterAndMap = f -> {
    			if (ai.incrementAndGet() == 1) {
    				return f.filter(color -> !color.equals("orange")).map(String::toUpperCase);
    			}
    			return f.filter(color -> !color.equals("purple")).map(String::toUpperCase);
    		};
    
    		Flux<String> composedFlux = Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
    				.doOnNext(System.out::println).compose(filterAndMap);
    
    		composedFlux.subscribe(d -> System.out.println("Subscriber 1 to Composed MapAndFilter :" + d));
    		composedFlux.subscribe(d -> System.out.println("Subscriber 2 to Composed MapAndFilter: " + d));
    	}
    
    	// 到目前介绍的其实都是 cold 家族的发布者。它们为每一个订阅（subscription）
    	// 都生成数据。如果没有创建任何订阅（subscription），那么就不会生成数据
    
    	// 热 发布者，不依赖于订阅者的数量。即使没有订阅者它们也会发出数据， 如果有一个订阅者接入进来，那么它就会收到订阅之后发出的元素。对于热发布者，
    	// 在你订阅它之前，确实已经发生了什么
    
    	// just 是 Reactor 中少数几个“热”操作符的例子之一：它直接在组装期（assembly time）
    	// 就拿到数据，如果之后有谁订阅它，就重新发送数据给订阅者。再拿 HTTP 调用举例，如果给 just 传入的数据是一个 HTTP
    	// 调用的结果，那么之后在初始化 just 的时候才会进行唯一的一次网络调用。
    
    	// 如果想将 just 转化为一种 冷 的发布者，你可以使用 defer。它能够将刚才例子中对 HTTP
    	// 的请求延迟到订阅时（这样的话，对于每一个新的订阅来说，都会发生一次网络调用）
    
    	// Reactor 中多数其他的 热 发布者是扩展自 Processor 的
    	@Test
    	public void testHotAndCold() {
    		Flux<String> source = Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
    				.doOnNext(System.out::println).filter(s -> s.startsWith("o")).map(String::toUpperCase);
    
    		source.subscribe(d -> System.out.println("Subscriber 1: " + d));
    		source.subscribe(d -> System.out.println("Subscriber 2: " + d));
    	}
    
    	// 两个订阅者都触发了所有的颜色，因为每一个订阅者都会让构造 Flux 的操作符运行一次。
    
    	// 将下边的例子与第一个例子对比：
    	public void testWithHotAndCold() {
    		UnicastProcessor<String> hotSource = UnicastProcessor.create();
    
    		Flux<String> hotFlux = hotSource.publish().autoConnect().map(String::toUpperCase);
    
    		hotFlux.subscribe(d -> System.out.println("Subscriber 1 to Hot Source: " + d));
    
    		hotSource.onNext("blue");
    		hotSource.onNext("green");
    
    		hotFlux.subscribe(d -> System.out.println("Subscriber 2 to Hot Source: " + d));
    
    		hotSource.onNext("orange");
    		hotSource.onNext("purple");
    		hotSource.onComplete();
    		// 第一个订阅者收到了所有的四个颜色，第二个订阅者由于是在前两个颜色发出之后订阅的， 故而收到了之后的两个颜色，在输出中有两次 "ORANGE" 和
    		// "PURPLE"。从这个例子可见， 无论是否有订阅者接入进来，这个 Flux 都会运行
    	}
    
    	// 使用 ConnectableFlux 对多个订阅者进行广播
    	@Test
    	public void testConnectableFlux() throws Exception {
    		Flux<Integer> source = Flux.range(1, 3).doOnSubscribe(s -> System.out.println("subscribed to source"));
    
    		ConnectableFlux<Integer> co = source.publish();
    
    		co.subscribe(System.out::println, e -> {
    		}, () -> {
    		});
    		co.subscribe(System.out::println, e -> {
    		}, () -> {
    		});
    
    		System.out.println("done subscribing");
    		Thread.sleep(500);
    		System.out.println("will now connect");
    
    		co.connect();
    
    		// 使用 autoConnect：
    		Flux<Integer> source2 = Flux.range(1, 3).doOnSubscribe(s -> System.out.println("subscribed to source"));
    
    		Flux<Integer> autoCo = source2.publish().autoConnect(2);
    
    		autoCo.subscribe(System.out::println, e -> {
    		}, () -> {
    		});
    		System.out.println("subscribed first");
    		Thread.sleep(500);
    		System.out.println("subscribing second");
    		autoCo.subscribe(System.out::println, e -> {
    		}, () -> {
    		});
    	}
    
    	@Test
    	public void testGroupBy() {
    		// 分组操作适用于分组个数不多的场景。而且所有的组都必须被消费，这样 groupBy 才能持续从上游获取数据。有时候这两种要求在一起——比如元素数量超多，
    		// 但是并行的用来消费的 flatMap 又太少的时候——会导致程序卡死
    		StepVerifier.create(Flux.just(1, 3, 5, 2, 4, 6, 11, 12, 13).groupBy(i -> i % 2 == 0 ? "even" : "odd")
    				.concatMap(g -> g.defaultIfEmpty(-1) // 如果组为空，显示为 -1
    						.map(String::valueOf) // 转换为字符串
    						.startWith(g.key())) // 以该组的 key 开头
    		).expectNext("odd", "1", "3", "5", "11", "13").expectNext("even", "2", "4", "6", "12").verifyComplete();
    	}
    
    	@Test
    	public void testWindow() {
    		// 如果将两个参数的配置反过来（maxSize < skip），序列中的一些元素就会被丢弃掉， 而不属于任何 window
    		StepVerifier.create(Flux.range(1, 10).window(5, 3) // overlapping windows
    				.concatMap(g -> g.defaultIfEmpty(-1)) // 将 windows 显示为 -1
    		).expectNext(1, 2, 3, 4, 5).expectNext(4, 5, 6, 7, 8).expectNext(7, 8, 9, 10).expectNext(10).verifyComplete();
    	}
    
    	// 对基于判断条件的 windowUntil 和 windowWhile，如果序列中的元素不匹配判断条件， 那么可能导致 空 windows，如下例所示：
    	@Test
    	public void testWindowX() {
    		StepVerifier
    				.create(Flux.just(1, 3, 5, 2, 4, 6, 11, 12, 13).windowWhile(i -> i % 2 == 0)
    						.concatMap(g -> g.defaultIfEmpty(-1)))
    				.expectNext(-1, -1, -1) // 分别被奇数 1 3 5 触发
    				.expectNext(2, 4, 6) // 被 11 触发
    				.expectNext(12) // 被 13 触发
    				.expectNext(-1) // 空的 completion window，如果 onComplete 前的元素能够匹配上的话就没有这个了
    				.verifyComplete();
    	}
    
    	@Test
    	public void testBuffer() {
    		StepVerifier.create(Flux.range(1, 10).buffer(5, 3) // 缓存重叠
    		).expectNext(Arrays.asList(1, 2, 3, 4, 5)).expectNext(Arrays.asList(4, 5, 6, 7, 8))
    				.expectNext(Arrays.asList(7, 8, 9, 10)).expectNext(Collections.singletonList(10)).verifyComplete();
    	}
    
    	// 不像窗口方法，bufferUntil 和 bufferWhile 不会发出空的 buffer
    	@Test
    	public void testBufferX() {
    		StepVerifier.create(Flux.just(1, 3, 5, 2, 4, 6, 11, 12, 13).bufferWhile(i -> i % 2 == 0))
    				.expectNext(Arrays.asList(2, 4, 6)) // 被 11 触发
    				.expectNext(Collections.singletonList(12)) // 被 13 触发
    				.verifyComplete();
    	}
    
    	// 使用 ParallelFlux 进行并行处理
    	// 可以对任何 Flux 使用 parallel() 操作符来得到一个 ParallelFlux.
    	// 不过这个操作符本身并不会进行并行处理，而是将负载划分到多个“轨道（rails）”上 （默认情况下，轨道个数与 CPU 核数相等）
    	// 为了配置 ParallelFlux 如何并行地执行每一个轨道，你需要使用 runOn(Scheduler)。
    	// 注意，Schedulers.parallel() 是推荐的专门用于并行处理的调度器。
    	@Test
    	public void testParallelFlux() {
    		// 我们给定一个轨道数字，而不是依赖于 CPU 核数
    		Flux.range(1, 10).parallel(2).subscribe(i -> System.out.println(Thread.currentThread().getName() + " -> " + i));
    
    		Flux.range(1, 10).parallel(2).runOn(Schedulers.parallel())
    				.subscribe(i -> System.out.println(Thread.currentThread().getName() + " -> " + i));
    	}
    
    	// 对于一些能够在一个线程中顺序处理的任务，即使调度到ParallelScheduler上，通常也只由一个Worker来执行，比如：
    	@Test
    	public void testParallelFluxX() throws InterruptedException {
    		Flux.range(1, 10).publishOn(Schedulers.parallel()).log().subscribe();
    		TimeUnit.MILLISECONDS.sleep(10);
    	}
    
    	/**
    	 * 有时候，我们确实需要一些任务能够“均匀”分布在不同的工作线程上执行，这时候就需要用到ParallelFlux。
    	 * 
    	 * 你可以对任何Flux使用parallel()操作符来得到一个ParallelFlux。不过这个操作符本身并不会进行并行处理，而只是将负载划分到多个执行“轨道”上（默认情况下，轨道个数与CPU核数相等）。
    	 * 
    	 * 为了配置ParallelFlux如何并行地执行每一个轨道，需要使用runOn(Scheduler)，这里，Schedulers.parallel()
    	 * 是比较推荐的专门用于并行处理的调度器。
    	 * 
    	 */
    	@Test
    	public void testParallelFluxXX() throws InterruptedException {
    		Flux.range(1, 10).parallel(2).runOn(Schedulers.parallel())
    				// .publishOn(Schedulers.parallel())
    				.log().subscribe();
    
    		TimeUnit.MILLISECONDS.sleep(10);
    	}
    
    	// 自从版本 3.1.0，Reactor 引入了一个类似于 ThreadLocal 的高级功能：Context。它作用于一个 Flux 或一个 Mono
    	// 上，而不是应用于一个线程（Thread）,也就是其生命周期伴随整个数据流，而不是线程。
    	@Test
    	public void testContext() {
    		String key = "message";
    		// 操作链以调用 subscriberContext(Function) 结尾，将 "World" 作为 "message" 这个 key 的 值添加到
    		// Context中
    		Mono<String> r = Mono.just("Hello")
    				// 对源调用 flatMap 用 Mono.subscriberContext() 方法拿到 Context
    				.flatMap(s -> Mono.subscriberContext()
    						// 然后使用 map 读取关联到 "message" 的值，然后与原来的值连接
    						.map(ctx -> s + " " + ctx.get(key)))
    				.subscriberContext(ctx -> ctx.put(key, "World"));
    		// 最后 Mono<String> 确实发出了 "Hello World"
    		StepVerifier.create(r).expectNext("Hello World").verifyComplete();
    
    		// 注意在你的操作链中，写入 与 读取 Context 的 相对位置 很重要：因为 Context 是不可变的，它的内容只能被上游的操作符看到
    		Mono<String> r2 = Mono.just("Hello").subscriberContext(ctx -> ctx.put(key, "World"))
    				.flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.getOrDefault(key, "Stranger")));
    
    		StepVerifier.create(r2).expectNext("Hello Stranger").verifyComplete();
    
    		// 下面的例子同样说明了 Context 的不可变性（Mono.subscriberContext() 总是返回由 subscriberContext 配置的
    		// Context）：
    
    		Mono<String> r3 = Mono.subscriberContext().map(ctx -> ctx.put(key, "Hello"))
    				.flatMap(ctx -> Mono.subscriberContext()).map(ctx -> ctx.getOrDefault(key, "Default"));
    
    		StepVerifier.create(r3).expectNext("Default").verifyComplete();
    
    		// 类似的，如果多次对 Context 中的同一个 key 赋值的话，要看 写入的相对顺序 ： 读取 Context 的操作符只能拿到下游最近的一次写入的值
    		Mono<String> r4 = Mono.just("Hello").flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key)))
    				.subscriberContext(ctx -> ctx.put(key, "Reactor")).subscriberContext(ctx -> ctx.put(key, "World"));
    
    		StepVerifier.create(r4).expectNext("Hello Reactor").verifyComplete();
    
    		// 你可能会觉得 Context 是与数据信号一块传播的。如果是那样的话，在两次写入操作中间加入的一个 flatMap 会使用最上游的这个
    		// Context。但并不是这样的，如下：原因在于 Context 是与 Subscriber 关联的，而每一个操作符访问的 Context 来自其下游的
    		// Subscriber
    		Mono<String> r5 = Mono.just("Hello").flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key)))
    				.subscriberContext(ctx -> ctx.put(key, "Reactor"))
    				.flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key)))
    				.subscriberContext(ctx -> ctx.put(key, "World"));
    
    		StepVerifier.create(r5).expectNext("Hello Reactor World").verifyComplete();
    
    		// 最后一个有意思的传播方式是，对 Context 的赋值也可以在一个 flatMap 内部,因为赋值 "Reactor" 的
    		// subscriberContext 是作用于第二个 flatMap 的内部序列的。所以不会在主序列可见/ 传播，第一个 flatMap
    		// 也看不到它。传播（Propagation） + 不可变性（immutability）将类似 flatMap 这样的操作符中的创建的内部序列中的
    		// Context 与外部隔离开来。
    		Mono<String> r6 = Mono.just("Hello").flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key)))
    				.flatMap(s -> Mono.subscriberContext().map(ctx -> s + " " + ctx.get(key))
    						.subscriberContext(ctx -> ctx.put(key, "Reactor")))
    				.subscriberContext(ctx -> ctx.put(key, "World"));
    
    		StepVerifier.create(r6).expectNext("Hello World Reactor").verifyComplete();
    	}
    
    	@Test
    	public void testAppendBoomError() {
    		StepVerifier.withVirtualTime(() -> Mono.delay(Duration.ofDays(1))).expectSubscription()
    				.expectNoEvent(Duration.ofDays(1)).expectNext(0L).verifyComplete();
    
    		StepVerifier
    				.create(Mono.just(1).map(i -> i + 10),
    						StepVerifierOptions.create().withInitialContext(Context.of("foo", "bar")))
    				.expectAccessibleContext().contains("foo", "bar").then().expectNext(11).verifyComplete();
    	}
    
    	/**
    	 * 当构建复杂的操作链时，可能会有多个子序列，从而导致多个执行路径。
    	 * 
    	 * 多数时候，这些子序列会生成一个足够明确的 onNext 信号，你可以通过检查最终结果 来判断它是否执行。
    	 * 
    	 * 考虑下边这个方法，它构建了一条操作链，并使用 switchIfEmpty 方法在源为空的情况下， 替换成另一个源。
    	 * 
    	 * @param source
    	 * @param fallback
    	 * @return
    	 * @date 2019-05-14
    	 */
    	public Flux<String> processOrFallback(Mono<String> source, Publisher<String> fallback) {
    		return source.flatMapMany(phrase -> Flux.fromArray(phrase.split("\\s+"))).switchIfEmpty(fallback);
    	}
    
    	// 很容易就可以测试出 switchIfEmpty 的哪一个逻辑分支被使用了
    	@Test
    	public void testSplitPathIsUsed() {
    		StepVerifier.create(processOrFallback(Mono.just("just a  phrase with    tabs!"), Mono.just("EMPTY_PHRASE")))
    				.expectNext("just", "a", "phrase", "with", "tabs!").verifyComplete();
    	}
    
    	@Test
    	public void testEmptyPathIsUsed() {
    		StepVerifier.create(processOrFallback(Mono.empty(), Mono.just("EMPTY_PHRASE"))).expectNext("EMPTY_PHRASE")
    				.verifyComplete();
    	}
    
    	private Mono<String> executeCommand(String command) {
    		return Mono.just(command + " DONE");
    	}
    
    	public Mono<Void> processOrFallback(Mono<String> commandSource, Mono<Void> doWhenEmpty) {
    		return commandSource.flatMap(command -> executeCommand(command).then()).switchIfEmpty(doWhenEmpty);
    	}
    
    	@Test
    	public void testFlatMap() {
    		StepVerifier.create(Flux.just("flux", "mono")
    				// 对于每一个字符串s，将其拆分为包含一个字符的字符串流；
    				// flatMap操作可以将【每个数据，一个大坨】，元素转换/映射为一个流，然后将这些流合并为一个大的数据流。
    				// 这个和Java8的类似,Stream<String[]> -> flatMap -> Stream<String>
    				.flatMap(s -> Flux.fromArray(s.split("\\s*")) // 1
    						.delayElements(Duration.ofMillis(100))) // 2对每个元素延迟100ms；
    				// 对每个元素进行打印（注doOnNext方法是“偷窥式”的方法，不会消费数据流）；
    				.doOnNext(System.out::print)) // 3
    				// 验证是否发出了8个元素。
    				.expectNextCount(8) // 4
    				.verifyComplete();
    	}
    
    	private Flux<String> getZipDescFlux() {
    		String desc = "Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.";
    		return Flux.fromArray(desc.split("\\s+")); // 1将英文说明用空格拆分为字符串流；
    	}
    
    	@Test
    	public void testSimpleOperators() throws InterruptedException {
    		// 定义一个CountDownLatch，初始为1，则会等待执行1次countDown方法后结束，不使用它的话，测试方法所在的线程会直接返回而不会等待数据流发出完毕；
    		CountDownLatch countDownLatch = new CountDownLatch(1); // 2
    		// 使用Flux.interval声明一个每200ms发出一个元素的long数据流；因为zip操作是一对一的，故而将其与字符串流zip之后，字符串流也将具有同样的速度；
    		Flux.zip(getZipDescFlux(), Flux.interval(Duration.ofMillis(200))) // 3
    				// zip之后的流中元素类型为Tuple2，使用getT1方法拿到字符串流的元素；定义完成信号的处理为countDown;
    				.subscribe(t -> System.out.println(t.getT1()), null, countDownLatch::countDown); // 4
    		// countDownLatch.await(10, TimeUnit.SECONDS)会等待countDown倒数至0，最多等待10秒钟。
    		countDownLatch.await(10, TimeUnit.SECONDS); // 5
    		// 除了zip静态方法之外，还有zipWith等非静态方法，效果与之类似
    	}
    
    	/**
    	 * 在 3.1 版本以前，你需要为每一种状态维护一个 AtomicBoolean 变量，然后在相应的 doOn*
    	 * 回调中观察它的值。这需要添加不少的额外代码。好在，版本 3.1.0 之后可以使用 `PublisherProbe`来做
    	 * 
    	 * @param string
    	 * @date 2019-05-14
    	 */
    	@Test
    	public void testCommandEmptyPathIsUsed() {
    		PublisherProbe<Void> probe = PublisherProbe.empty();
    
    		StepVerifier.create(processOrFallback(Mono.empty(), probe.mono())).verifyComplete();
    
    		probe.assertWasSubscribed();
    		probe.assertWasRequested();
    		probe.assertWasNotCancelled();
    	}
    
    	// 假设我们有一个同步阻塞的调用方法
    	private String getStringSync() {
    		try {
    			TimeUnit.SECONDS.sleep(2);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		return "Hello, Reactor!";
    	}
    
    	// 正常情况下，调用这个方法会被阻塞2秒钟，然后同步地返回结果。我们借助elastic调度器将其变为异步，由于是异步的，为了保证测试方法所在的线程能够等待结果的返回，我们使用CountDownLatch：
    	@Test
    	public void testSyncToAsync() throws InterruptedException {
    		CountDownLatch countDownLatch = new CountDownLatch(1);
    		Mono.fromCallable(() -> getStringSync()) // 1使用fromCallable声明一个基于Callable的Mono
    				// 使用subscribeOn将任务调度到Schedulers内置的弹性线程池执行，弹性线程池会为Callable的执行任务分配一个单独的线程。
    				.subscribeOn(Schedulers.elastic()) // 2
    				.subscribe(System.out::println, null, countDownLatch::countDown);
    		countDownLatch.await(10, TimeUnit.SECONDS);
    	}
    
    	// 假设，我们现在有一个非常快的Publisher——Flux.range(1,
    	// 6)，然后自定义一个每秒处理一个数据元素的慢的Subscriber，Subscriber就需要通过request(n)的方法来告知上游它的需求速度
    	@Test
    	public void testBackpressure() {
    		Flux.range(1, 6) // 1Flux.range是一个快的Publisher；
    				.doOnRequest(n -> System.out.println("Request " + n + " values...")) // 2在每次request的时候打印request个数；
    				.subscribe(new BaseSubscriber<Integer>() { // 3通过重写BaseSubscriber的方法来自定义Subscriber；
    					@Override
    					protected void hookOnSubscribe(Subscription subscription) { // 4hookOnSubscribe定义在订阅的时候执行的操作；
    						System.out.println("Subscribed and make a request...");
    						request(1); // 5订阅时首先向上游请求1个元素；
    					}
    
    					@Override
    					protected void hookOnNext(Integer value) { // 6hookOnNext定义每次在收到一个元素的时候的操作；
    						try {
    							TimeUnit.SECONDS.sleep(1); // 7sleep 1秒钟来模拟慢的Subscriber；
    						} catch (InterruptedException e) {
    							e.printStackTrace();
    						}
    						System.out.println("Get value [" + value + "]"); // 8打印收到的元素；
    						request(1); // 9每次处理完1个元素后再请求1个。s
    					}
    				});
    	}
    
    	// 用于生成数据流的方法create和push可以用于异步的场景，而且它们也支持回压，我们可以通过提供一个 OverflowStrategy
    	// 来定义背压行为。方法签名：public static <T> Flux<T> create(Consumer<? super FluxSink<T>>
    	// emitter, OverflowStrategy backpressure),默认（没有第二个参数的方法）是缓存策略的
    
    	/**
    	 * 调度器相当于Reactor中的ExecutorService，不同的调度器定义不同的线程执行环境。Schedulers工具类提供的静态方法可搭建不同的线程执行环境。
    	 * 
    	 * Schedulers类已经预先创建了几种常用的不同线程池模型的调度器：使用single()、elastic()和parallel()方法创建的调度器可以分别使用内置的单线程、弹性线程池和固定大小线程池。如果想创建新的调度器，可以使用newSingle()、newElastic()和newParallel()方法。这些方法都是返回一个Scheduler的具体实现。
    	 * 
    	 */
    	// 使用publishOn和subscribeOn切换不同的线程执行环境。
    	@Test
    	public void testScheduling() {
    		Flux.range(0, 10)
    				// .log() // 1只保留这个log()的话，可以看到，源头数据流是执行在myElastic-x线程上的；
    				// publishOn会影响链中其后的操作符，比如第一个publishOn调整调度器为elastic，则filter的处理操作是在弹性线程池中执行的；同理，flatMap是执行在固定大小的parallel线程池中的；
    				.publishOn(Schedulers.newParallel("myParallel"))
    				// .log() // 2只保留这个log()的话，可以看到，publishOn之后数据流是执行在myParallel-x线程上的；
    				// subscribeOn无论出现在什么位置，都只影响源头的执行环境，也就是range方法是执行在单线程中的，直至被第一个publishOn切换调度器之前，所以range后的map也在单线程中执行。
    				.subscribeOn(Schedulers.newElastic("myElastic"))
    				// 3只保留这个log()的话，可以看到，subscribeOn之后数据流依然是执行在myParallel-x线程上的。
    				.log().blockLast();
    	}
    
    	/**
    	 * 此外，有些操作符本身会需要调度器来进行多线程的处理，当你不明确指定调度器的时候，那些操作符会自行使用内置的单例调度器来执行。例如，Flux.delayElements(Duration)
    	 * 使用的是 Schedulers.parallel()调度器对象：
    	 * 
    	 */
    	@Test
    	public void testDelayElements() {
    		Flux.range(0, 10).delayElements(Duration.ofMillis(10))
    				// 从输出可以看到onNext运行在不同的线程上
    				.log().blockLast();
    	}
    
    	private void log(String string) {
    	}
    
    	private Publisher<String> callExternalService(String k) {
    		return null;
    	}
    
    	private String doSecondTransform(Object v) {
    		return null;
    	}
    
    	private Object doSomethingDangerous(Integer v) {
    		return null;
    	}
    
    	// ---------------------Reactor的操作符-----------------------------
    	// 创建一个新序列
    	@Test
    	public void testAllCreate() {
    		// 发出一个 T，我已经有了：just
    
    		// …基于一个 Optional<T>：Mono#justOrEmpty(Optional<T>)
    		Mono.justOrEmpty(Optional.of("haha"));
    
    		// …基于一个可能为 null 的 T：Mono#justOrEmpty(T)
    		Mono.justOrEmpty(null);
    
    		// 发出一个 T，且还是由 just 方法返回
    
    		// …但是“懒”创建的：使用 Mono#fromSupplier 或用 defer 包装 just
    		LinkedList<String> linkedList = new LinkedList<>();
    		Mono.fromSupplier(ArrayList::new);
    		Mono.fromSupplier(() -> "heihei");
    		Mono.fromSupplier(() -> {
    			return linkedList;
    		});
    
    		// 发出许多 T，这些元素我可以明确列举出来：Flux#just(T...)
    		Flux.<String>just("hehe", "heihei", "huhu");
    		Flux.<String>just(new String[] { "x", "y", "z" });
    
    		// 基于迭代数据结构:
    		ArrayList<String> list = new ArrayList<>(Arrays.asList("hehe", "heihei", "huhu"));
    
    		// 一个数组：Flux#fromArray
    		Flux.fromArray(new String[] { "x", "y", "z" });
    		// 一个集合或 iterable：Flux#fromIterable
    		Flux.fromIterable(list);
    		// 一个 Integer 的 range：Flux#range
    		Flux.range(0, 10);
    		// 一个 Stream 提供给每一个订阅：Flux#fromStream(Supplier<Stream>)
    		Flux.fromStream(Stream.of("hehe", "heihei", "huhu"));
    
    		// 基于一个参数值给出的源：
    
    		// 一个 Supplier<T>：Mono#fromSupplier
    		Mono.fromSupplier(String::new);
    		// 一个任务：Mono#fromCallable，Mono#fromRunnable
    		Mono.fromRunnable(() -> {
    		});
    		Mono.fromCallable(() -> "heihei");
    		Mono.fromCallable(() -> {
    			return "heihei";
    		});
    		// 一个 CompletableFuture<T>：Mono#fromFuture----->不会
    
    		// 直接完成：empty
    		Flux.empty();
    		Mono.empty();
    
    		// 立即生成错误：error
    		Mono.error(new RuntimeException());
    		// …但是“懒”的方式生成 Throwable：error(Supplier<Throwable>)
    		Flux.error(RuntimeException::new);
    
    		// 什么都不做：never
    		Mono.never();
    
    		// 订阅时才决定：defer
    		Flux.defer(() -> Flux.empty());
    		Flux.defer(() -> {
    			return Flux.empty();
    		});
    
    		// 依赖一个可回收的资源：using
    
    		// 可编程地生成事件（可以使用状态）:
    		// 同步且逐个的：Flux#generate
    
    		// 异步（也可同步）的，每次尽可能多发出元素：Flux#create （Mono#create 也是异步的，只不过只能发一个）
    	}
    
    	// 对序列进行转化
    	@Test
    	public void testAllConversion() {
    		// 我想转化一个序列：
    		// 1对1地转化（比如字符串转化为它的长度）：map
    
    		// ...类型转化：cast
    
    		// ...为了获得每个元素的序号：Flux#index
    
    		// 1对n地转化（如字符串转化为一串字符）：flatMap + 使用一个工厂方法
    
    		// 1对n地转化可自定义转化方法和/或状态：handle
    
    		// 对每一个元素执行一个异步操作（如对 url 执行 http 请求）：flatMap + 一个异步的返回类型为 Publisher 的方法
    
    		// ...忽略一些数据：在 flatMap lambda 中根据条件返回一个 Mono.empty()
    
    		// ...保留原来的序列顺序：Flux#flatMapSequential（对每个元素的异步任务会立即执行，但会将结果按照原序列顺序排序）
    
    		// ...当 Mono 元素的异步任务会返回多个元素的序列时：Mono#flatMapMany
    
    		// 我想添加一些数据元素到一个现有的序列：
    		// 在开头添加：Flux#startWith(T...)
    
    		// 在最后添加：Flux#concatWith(T...)
    
    		// 我想将 Flux 转化为集合（一下都是针对 Flux 的）
    
    		// 转化为 List：collectList，collectSortedList
    
    		// 转化为 Map：collectMap，collectMultiMap
    
    		// 转化为自定义集合：collect
    
    		// 计数：count
    
    		// reduce 算法（将上个元素的reduce结果与当前元素值作为输入执行reduce方法，如sum） reduce
    
    		// ...将每次 reduce 的结果立即发出：scan
    
    		// 转化为一个 boolean 值：
    
    		// 对所有元素判断都为true：all
    
    		// 对至少一个元素判断为true：any
    
    		// 判断序列是否有元素（不为空）：hasElements
    
    		// 判断序列中是否有匹配的元素：hasElement
    
    		// 我想合并 publishers...
    		// 按序连接：Flux#concat 或 .concatWith(other)
    
    		// ...即使有错误，也会等所有的 publishers 连接完成：Flux#concatDelayError
    
    		// ...按订阅顺序连接（这里的合并仍然可以理解成序列的连接）：Flux#mergeSequential
    
    		// 按元素发出的顺序合并（无论哪个序列的，元素先到先合并）：Flux#merge / .mergeWith(other)
    
    		// ...元素类型会发生变化：Flux#zip / Flux#zipWith
    
    		// 将元素组合：
    		// 2个 Monos 组成1个 Tuple2：Mono#zipWith
    
    		// n个 Monos 的元素都发出来后组成一个 Tuple：Mono#zip
    
    		// 在终止信号出现时“采取行动”：
    		// 在 Mono 终止时转换为一个 Mono<Void>：Mono#and
    
    		// 当 n 个 Mono 都终止时返回 Mono<Void>：Mono#when
    
    		// 返回一个存放组合数据的类型，对于被合并的多个序列：
    		// 每个序列都发出一个元素时：Flux#zip
    
    		// 任何一个序列发出元素时：Flux#combineLatest
    
    		// 只取各个序列的第一个元素：Flux#first，Mono#first，mono.or<br/>(otherMono).or(thirdMono)，`flux.or(otherFlux).or(thirdFlux)
    
    		// 由一个序列触发（类似于 flatMap，不过“喜新厌旧”）：switchMap
    
    		// 由每个新序列开始时触发（也是“喜新厌旧”风格）：switchOnNext
    
    		// 我有一个空序列，但是...
    		// 我想要一个缺省值来代替：defaultIfEmpty
    
    		// 我想要一个缺省的序列来代替：switchIfEmpty
    
    		// 我有一个序列，但是我对序列的元素值不感兴趣：ignoreElements
    		// ...并且我希望用 Mono 来表示序列已经结束：then
    
    		// ...并且我想在序列结束后等待另一个任务完成：thenEmpty
    
    		// ...并且我想在序列结束之后返回一个 Mono：Mono#then(mono)
    
    		// ...并且我想在序列结束之后返回一个值：Mono#thenReturn(T)
    
    		// ...并且我想在序列结束之后返回一个 Flux：thenMany
    
    		// 我有一个 Mono 但我想延迟完成...
    		// ...当有1个或N个其他 publishers 都发出（或结束）时才完成：Mono#delayUntilOther
    
    		// ...使用一个函数式来定义如何获取“其他 publisher”：Mono#delayUntil(Function)
    
    		// 我想基于一个递归的生成序列的规则扩展每一个元素，然后合并为一个序列发出：
    		// ...广度优先：expand(Function)
    
    		// ...深度优先：expandDeep(Function)
    	}
    
    	// “窥视”（只读）序列
    	@Test
    	public void testPeepReadOnly() {
    		// 再不对序列造成改变的情况下，我想：
    		// 得到通知或执行一些操作：
    		// 发出元素：doOnNext
    
    		// 序列完成：Flux#doOnComplete，Mono#doOnSuccess
    
    		// 因错误终止：doOnError
    
    		// 取消：doOnCancel
    
    		// 订阅时：doOnSubscribe
    
    		// 请求时：doOnRequest
    
    		// 完成或错误终止：doOnTerminate（Mono的方法可能包含有结果）
    
    		// 但是在终止信号向下游传递 之后 ：doAfterTerminate
    
    		// 所有类型的信号（Signal）：Flux#doOnEach
    		// 所有结束的情况（完成complete、错误error、取消cancel）：doFinally
    
    		// 记录日志：log
    	}
    
    	// 过滤序列
    	@Test
    	public void testFilter() {
    		// 我想过滤一个序列
    		// 基于给定的判断条件：filter
    
    		// ...异步地进行判断：filterWhen
    
    		// 仅限于指定类型的对象：ofType
    
    		// 忽略所有元素：ignoreElements
    
    		// 去重:
    		// 对于整个序列：Flux#distinct
    
    		// 去掉连续重复的元素：Flux#distinctUntilChanged
    
    		// 我只想要一部分序列：
    
    		// 只要 N 个元素：
    		// 从序列的第一个元素开始算：Flux#take(long)
    
    		// ...取一段时间内发出的元素：Flux#take(Duration)
    
    		// ...只取第一个元素放到 Mono 中返回：Flux#next()
    
    		// ...使用 request(N) 而不是取消：Flux#limitRequest(long)
    
    		// 从序列的最后一个元素倒数：Flux#takeLast
    
    		// 直到满足某个条件（包含）：Flux#takeUntil（基于判断条件），Flux#takeUntilOther（基于对 publisher 的比较）
    
    		// 直到满足某个条件（不包含）：Flux#takeWhile
    
    		// 最多只取 1 个元素：
    		// 给定序号：Flux#elementAt
    
    		// 最后一个：.takeLast(1)
    		// ...如果为序列空则发出错误信号：Flux#last()
    
    		// ...如果序列为空则返回默认值：Flux#last(T)
    
    		// 跳过一些元素：
    		// 从序列的第一个元素开始跳过：Flux#skip(long)
    
    		// ...跳过一段时间内发出的元素：Flux#skip(Duration)
    
    		// 跳过最后的 n 个元素：Flux#skipLast
    
    		// 直到满足某个条件（包含）：Flux#skipUntil（基于判断条件），Flux#skipUntilOther （基于对 publisher 的比较）
    
    		// 直到满足某个条件（不包含）：Flux#skipWhile
    
    		// 采样：
    		// 给定采样周期：Flux#sample(Duration)
    
    		// 取采样周期里的第一个元素而不是最后一个：sampleFirst
    
    		// 基于另一个 publisher：Flux#sample(Publisher)
    
    		// 基于 publisher“超时”：Flux#sampleTimeout （每一个元素会触发一个 publisher，如果这个 publisher
    		// 不被下一个元素触发的 publisher 覆盖就发出这个元素）
    
    		// 我只想要一个元素（如果多于一个就返回错误）...
    		// 如果序列为空，发出错误信号：Flux#single()
    
    		// 如果序列为空，发出一个缺省值：Flux#single(T)
    
    		// 如果序列为空就返回一个空序列：Flux#singleOrEmpty
    	}
    
    }
    
    /**
     * SampleSubscriber 类继承自 BaseSubscriber，在 Reactor 中, 推荐用户扩展它来实现自定义的
     * Subscriber。这个类提供了一些 hook 方法，我们可以通过重写它们来调整 subscriber 的行为。
     * 默认情况下，它会触发一个无限个数的请求，但是当你想自定义请求元素的个数的时候，扩展 BaseSubscriber 就很方便了。
     * 
     * 扩展的时候通常至少要覆盖 hookOnSubscribe(Subscription subscription) 和 hookOnNext(T value)
     * 这两个方法。这个例子中， hookOnSubscribe 方法打印一段话到标准输出，然后进行第一次请求。 然后 hookOnNext
     * 同样进行了打印，同时逐个处理剩余请求。
     * 
     * 建议你同时重写 hookOnError、hookOnCancel，以及 hookOnComplete 方法。 你最好也重写 hookFinally
     * 方法。SampleSubscribe 确实是一个最简单的实现了 请求有限个数元素的 Subscriber。
     *
     * @param <T>
     */
    class SampleSubscriber<T> extends BaseSubscriber<T> {
    
    	public void hookOnSubscribe(Subscription subscription) {
    		System.out.println("Subscribed");
    		request(1);
    	}
    
    	public void hookOnNext(T value) {
    		System.out.println(value);
    		request(1);
    	}
    }
    
    // 假设你有一个监听器 API，它按 chunk 处理数据，有两种事件：（1）一个 chunk 数据准备好的事件；（2）处理结束的事件。如下：
    interface MyEventListener<T> {
    	void onDataChunk(List<T> chunk);
    
    	void processComplete();
    }
    
    class UnknownKeyException extends RuntimeException {
    	private static final long serialVersionUID = 1L;
    }