//Map 的方法签名 Flux map(Function mapper)//FlatMap的方法签名 Flux flatMap(Function> mapper)

void demo() { //1. Flux.interval 按时生产Long的无限流 final Flux flux = Flux.interval(Duration.of(100, ChronoUnit.MILLIS)) .map(log()) //2. 调用方法log，订阅后log()会在当前线程执行 .flatMap(logOfFlux()); //3. 通过flatMap调用 log(), 会在当前线程执行 //完成flux的定义，调用subscribe后才会真正开始执行 flux.subscribe(); } ? ? Function log() { return aLong -> { log.info("num is {}", aLong); return aLong; }; } ? Function> logOfFlux() { return aLong -> { log.info("num is {}", aLong); return Flux.just(aLong); }; }

void monoOfEmpty() { Mono.empty() .map(m -> func()) .subscribe(message -> { responseObserver.onNext(message); responseObserver.onCompleted(); ); }

void monoOfEmpty() { Mono.empty() .map(m -> func()) .doOnSuccess(message -> responseObserver.onCompleted()) .subscribe(responseObserver::onNext); }

//queries sql List Flux.fromArray(queries.values()) //需要使用flatMapSequential 保证Flux订阅时返回的顺序 .flatMapSequential(query -> { //执行sql return executor.getFromCacheOrExecute(...); }); }) ? Mono getFromCacheOrExecute() { return existsCache(hashing) .flatMap(exists-> { if(exists) { //查询缓存 }else{ //查询clickhouse return writeCache(hashing, doExecute(...)) } }); } ? //查询缓存 public Mono existsKey(String key) { return stringRedisTemplate.hasKey(key); } ? //提交查询至clickhouse Mono doExecute(...) { return Mono.fromCallable(() -> execute(...)); }

我们可以通过这两种方式达到异步执行的目的

void demo() { final Flux flux = Flux.interval(Duration.of(100, ChronoUnit.MILLIS)) .map(log())//1 .publishOn(Schedulers.parallel())//2 .map(log())//3 .publishOn(Schedulers.elastic())//4 .flatMap(logOfMono()); //5 flux.subscribe(); }

1. 会在调用flux.subscribe()的线程上执行
   引用
2. 指定后续的运算符执行的上下文
   引用
3. 会在parallel线程上执行
   引用
4. 指定后续的运算符执行的上下文
   引用
5. 会在elastic线程上执行

void demo() { final Flux flux = Flux.interval(Duration.of(100, ChronoUnit.MILLIS)) .map(log())//1 .publishOn(Schedulers.parallel()) .map(log())//2 .subscribeOn(Schedulers.elastic()) .flatMap(logOfMono()); flux.subscribe(); }

1. 会在elastic线程执行
   引用
2. 会在parallel线程上执行，publishOn会“覆盖”subscribeOn的行为

Flux依次发出了1，2，3三个元素
无论是subscribeOn、还是publishOn，经过flatMap 都会在 parallel-1 线程上执行，也就是说Flux中的所有元素只是从主线程发出，在另一个线程中执行。

void demo() { final Flux flux = Flux.fromIterable(Lists.newArrayList(3L, 1L, 2L)) .parallel().runOn(Schedulers.elastic()) .flatMap(logOfMono()) .sequential(); flux.subscribe(); }

2. fluxMap中调用的方法需要在内部通过publishOn表明执行上下文。

void demo() { Flux.fromIterable(Lists.newArrayList(3L, 1L, 2L)) .flatMap(this::blockFunction) .subscribe(); } ? //通过publishOn 来表明 blockFunction() 的上下文 Mono blockFunctionAsync(Long i) { return Mono.just(i).publishOn(Schedulers.elastic()).flatMap(this::blockFunction); } ? Mono blockFunction(Long i) { Thread.sleep(i * 1000); return Mono.just(i); }

3. 使用 Mono.fromFuture() 创建流

Mono.fromFuture(CompletableFuture.supplyAsync(() -> blockFunction()));

4. 使用Mono.create() 、Flux.create() 方法创建流

//自定义线程池 final Executor executor = new ThreadPoolExecutor(...); //创建Mono、Flux时 指明上下文 Mono.create(sink -> executor.execute(() -> { sink.success(blockFunction()); }));

1. 当前线程执行(Schedulers.immediate()): 在处理时，将直接执行提交的Runnable，从而在当前线程上有效地运行它们（可以视为“空对象”或无操作调度程序）。
2. 单个可重用线程(Schedulers.single())。请注意，此方法对所有调用方都使用相同的线程，直到调度程序被释放为止。如果需要每次调用一个专用线程，请对每个调用使用Schedulers.newSingle()。
3. 无限制的弹性线程池(Schedulers.elastic())。它根据需要创建新的工作池并重用空闲的工作池，随着Flux中元素的增多，会无限制创建线程去执行，存在安全隐患，不推荐使用。
4. 有界弹性线程池(Schedulers.boundedElastic())。根据需要创建新的工作池并重用空闲的工作池。闲置时间过长（默认值为60s）的工作池也将被丢弃。与elastic()有所不同，它对可以创建的线程数进行限制（默认为CPU核心数x 10）。达到上限后，最多可再提交10万个任务，并在有线程可用时重新调度（当任务被设置延迟执行时，延迟计时是在线程可用时开始）。这是I/O阻塞任务的更好选择。Schedulers.boundedElastic()是一种为阻塞处理提供自己的线程的简便方法，这样它就不会占用其他资源。缺点是这个线程池是全局的，假设有两个阻塞方法 getFiles、getData ，期望分别使用不同线程池去方便管理，就不适合用 boundedElastic 了。
5. 为并行工作而调整的固定工作线程池(Schedulers.parallel())。它创建的工作线程数量与CPU内核数量一样多。

1. 捕获并返回一个静态的缺省值。
2. 捕获并执行一个异常处理方法。
3. 捕获并动态计算一个候补值来顶替。
4. 捕获，并再包装为某一个 业务相关的异常，然后再抛出业务异常。
5. 捕获，记录错误日志，然后继续抛出。
6. 使用 finally 来清理资源

Flux.just(10) .flatMap(this::function) .onErrorReturn("Error"); //返回一个静态的缺省值

Flux.just(10) .flatMap(this::function) .onErrorReturn(e -> e.getMessage().equals("boom-1"), "Error-1");

Flux.just(10) .flatMap(m -> function(k).onErrorResume(e -> handleErr(k))); ? Flux.just(10) .flatMap(m -> function(k).onErrorResume(e -> errFunc(k))); ? Flux.just(10) .flatMap(m -> function(k) .onErrorResume(e -> Flux.error(new Exception(k))) );

Flux.just(10) .flatMap(k -> function(k)) .doOnError(e -> log.error(e.getLocalizedMessage(), e); });

Flux.just(10) .doFinally(type -> { if (type == SignalType.CANCEL){ log.info("a log"); } })