关于reactor:GrowingIO-Reactor速成指南

简介

Reactor响应式编程（Reactive Programming）：

Reactor is a fully non-blocking reactive programming foundation for the JVM, with efficient demand management (in the form of managing “backpressure”). It integrates directly with the Java 8 functional APIs, notably CompletableFuture, Stream, and Duration. It offers composable asynchronous sequence APIs — Flux (for [N] elements) and Mono (for [0|1] elements) — and extensively implements the Reactive Streams specification.

翻译一下就是：Reactor是JVM的一个齐全无阻塞的响应式编程根底，具备高效的需要治理(以治理“背压”的模式)。它间接与Java 8性能api集成，尤其是CompletableFuture、Stream和Duration。它提供了可组合的异步序列api——Flux(用于[N]元素)和Mono(用于[0|1]元素)——并且宽泛地实现了反馈流标准。

背景

咱们服务端的我的项目大多采纳了Spring WebFlux，reactor是 Spring WebFlux 的首选反馈库，WebFlux 须要 Reactor 作为外围依赖项。Reactor 存在肯定的学习老本，在开发中咱们遇到了些bug，相当一部分是因为咱们不够理解 reactor ，踩了很多坑。所以在本文档中咱们次要针对的是一些学习过程容易让新人感到迷茫的知识点（map、flatMap、异步、并发），冀望能让新人更好上手 Spring WebFlux。

Mono

Mono<T>是非凡的Publisher<T>，它通过onNext信号最多收回一个我的项目，而后以一个onComplete信号（胜利Mono，有或没有值）终止，或者只收回一个onError信号。某些Operator（尤其是那些将Mono与其余Publisher联合在一起的Operator）能够把Mono切换到Flux。例如，Mono.concatWith(Publisher)返回Flux，而Mono.then(Mono)返回另一个Mono。

Flux

Flux<T>是规范的Publisher<T>，示意它是能够发送0到N个元素的异步序列，可选的终止操作有onComplete或onError。与Reactive Streams标准一样，这三种信号转换为对上游的onNext，onComplete和onError办法的调用。 Flux是通用的响应式类型。请留神，所有事件，甚至是终止事件，都是可选的，意思是：可能没有onNext事件，但只有onComplete事件，这就示意此Flux是一个空的无限序列。删除onComplete则能够失去一个有限的空序列（除了勾销测试外，它没有什么用途）。同样，有限序列不肯定为空。例如，Flux.interval(Duration)有限生产Flux<Long>。

map 、flatMap 以及 flatMapSequential 区别：

办法签名

//Map 的办法签名<V> Flux<V> map(Function<? super T, ? extends V> mapper)//FlatMap的办法签名<R> Flux<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper)

map：通过对每个元素利用同步函数来转换此Flux收回的元素，并且是一对一的转换流元素。

flatMap：将此Flux收回的元素异步转换为 Publisher，而后通过合并将这些外部发布者扁平化为单个Flux 。Flux可能蕴含N个元素，所以flatMap是一对多的转换。将各个 publisher 合并的过程不会放弃源Flux公布的程序，可能呈现交织。

flatMapSequential：将此Flux收回的元素异步转换为 Publisher，而后通过合并将这些外部发布者扁平化为单个Flux 。于 flatMap 不同的是 flatMapSequential 在合并 publisher 时会按源元素的程序合并它们。

map是一个同步运算符，它只是一种将一个值转换为另一个值的办法。

flatMap能够是同步的，也能够是异步的，这取决于flatMap中调用的办法是否应用。

示例1：

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

在咱们的示例代码中，该 flatMap 操作是同步的，因为咱们应用Flux.just()办法收回元素。上面咱们会介绍如何在 flatMap 中实现异步操作。

上边代码中讲到了 Publisher 调用 subscribe 后才会真正开始执行，然而 subscribe 中的代码并不一定会执行。

当 Mono 是空序列时：

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

会有同学喜爱在 subscribe 中解决响应（例如rpc），但这个场景中responseObserver.onCompleted() 不会被执行。

正确的做法应该是：

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

异步与多线程

Reactor被视为与并发无关的。也就是说，取得Flux或Mono并不一定意味着它在专用线程中运行。取而代之的是，大多数Operator会持续在执行前一个Operator的线程中工作。除非指定，否则最顶层的Operator（源）自身运行在进行subscribe()调用的线程上。

先分享一个案例：
在 Growing 查问服务 olap 中因为应用谬误的操作符带来性能问题

景象：在查看某些看板时，响应工夫会超时（一分钟）。

通过剖析发现这些看板中的单图查问时间段蕴含“明天”，因为须要保障实时数据，olap在解决蕴含“明天”的查问时不会生成缓存；通过日志发现olap在查问clickhouse时始终都是一个线程在提交。

对应代码：

//queries sql ListFlux.fromArray(queries.values())    //须要应用flatMapSequential 保障Flux订阅时返回的程序    .flatMapSequential(query -> {        //执行sql        return executor.getFromCacheOrExecute(...);    });})Mono<DataSet> getFromCacheOrExecute() {    return existsCache(hashing)        .flatMap(exists-> {             if(exists) {              //查问缓存             }else{             //查问clickhouse             return writeCache(hashing, doExecute(...))             }        });}//查问缓存public Mono<Boolean> existsKey(String key) {    return stringRedisTemplate.hasKey(key);}//提交查问至clickhouseMono<DataSet> doExecute(...) {    return Mono.fromCallable(() -> execute(...));}

代码中没有应用subscriptOn或者publishOn操作符，stringRedisTemplate.hasKey(key) 会使后续的操作符都在 lettuce-1 线程中执行并且整个流的执行都是串行的，咱们须要将查问过程改为异步查问，多个sql 能够并发查问。上面就介绍实现异步并发的一些办法，及GrowingIO利用于这个场景的抉择。

Reactor提供了两种在流中切换执行上下文（或Scheduler）的形式：

咱们能够通过这两种形式达到异步执行的目标

publishOn : 此运算符影响线程上下文，它上面的链中的其余运算符将在其中执行，直到新呈现的publishOn。

void demo() {    final Flux<Long> 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();}

会在调用flux.subscribe()的线程上执行
援用
指定后续的运算符执行的上下文
援用
会在parallel线程上执行
援用
指定后续的运算符执行的上下文
援用
会在elastic线程上执行

subscribeOn ：整个流在指定的Scheduler的Scheduler.Worker上运行，直至呈现publishOn ，publishOn后续的操作符由 publishOn 决定执行上下文。

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

会在elastic线程执行
援用
会在parallel线程上执行，publishOn会“笼罩”subscribeOn的行为

这样咱们就能够异步的去执行办法了。这里有一个误区，刚接触reactor的小伙伴会误认为Flux中多个元素通过log()或者 logOfMono() 会别离打印在不同线程上

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

在调用一些阻塞办法时（rpc、redis、io），咱们冀望每个元素通过 flatMap 中时能够运行在不同线程上（串行-> 并发），应该怎么做？
将Flux转为ParallelFlux，应用runOn来指明须要的线程池

要取得ParallelFlux，能够在Flux 上应用parallel()运算符。

为了通知ParallelFlux在哪里运行每个元素，必须应用runOn(Scheduler)。如果并行处理后，您想复原到“失常”状态 Flux并以程序形式利用运算符链的其余部分，能够应用sequential()

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

fluxMap中调用的办法须要在外部通过publishOn表明执行上下文。

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

应用 Mono.fromFuture() 创立流

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

Mono.fromFuture() 创立一个Mono ，应用提供的 CompletableFuture 产生它的值。

CompletableFuture.supplyAsync() 返回一个 CompletableFuture，它将在ForkJoinPool.commonPool() 上运行工作，异步实现。

ForkJoinPool一个全局线程池，次要利用于计算密集型的场景。

应用Mono.create() 、Flux.create() 办法创立流

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

目前咱们举荐应用这种办法来创立Mono、Flux来实现异步

这样做的益处：

Reactor对老手来说有肯定了解老本，在调用一个返回值为 Publisher 的类型的办法时，不点进去无奈晓得这是同步办法还是异步办法，防止对于调用者造成的心智累赘。

更不便得应用自定义线程池

咱们针对 olap 查问业务场景采取适合的办法：

办法一： ParallelFlux 不能保障sql返回程序，不满足该业务场景。

办法二： Reactor对老手来说有肯定了解老本，在调用一个返回值为 Publisher 的类型的办法时，不点进去无奈晓得这是同步办法还是异步办法，对于调用者存在心智累赘。

办法三：无奈指定自定义线程池

咱们采纳了第4种形式来实现异步。

reactor提供的线程池

以后线程执行(Schedulers.immediate()): 在解决时，将间接执行提交的Runnable，从而在以后线程上无效地运行它们（能够视为“空对象”或无操作调度程序）。
单个可重用线程(Schedulers.single())。请留神，此办法对所有调用方都应用雷同的线程，直到调度程序被开释为止。如果须要每次调用一个专用线程，请对每个调用应用Schedulers.newSingle()。
无限度的弹性线程池(Schedulers.elastic())。它依据须要创立新的工作池并重用闲暇的工作池，随着Flux中元素的增多，会无限度创立线程去执行，存在安全隐患，不举荐应用。
有界弹性线程池(Schedulers.boundedElastic())。依据须要创立新的工作池并重用闲暇的工作池。闲置工夫过长（默认值为60s）的工作池也将被抛弃。与elastic()有所不同，它对能够创立的线程数进行限度（默认为CPU外围数x 10）。达到下限后，最多可再提交10万个工作，并在有线程可用时从新调度（当工作被设置提早执行时，提早计时是在线程可用时开始）。这是I/O阻塞工作的更好抉择。Schedulers.boundedElastic()是一种为阻塞解决提供本人的线程的简便办法，这样它就不会占用其余资源。毛病是这个线程池是全局的，假如有两个阻塞办法 getFiles、getData ，冀望别离应用不同线程池去方便管理，就不适宜用 boundedElastic 了。
为并行工作而调整的固定工作线程池(Schedulers.parallel())。它创立的工作线程数量与CPU内核数量一样多。

处理错误

在响应式流中，谬误（error）是终止（terminal）事件。当有谬误产生时，它会导致流序列进行，并且谬误信号会沿着操作链条向下传递，直至遇到定义的 Subscriber 及其 onError 办法。

在 try-catch 代码块中解决异样的几种办法。常见的包含如下几种：

捕捉并返回一个动态的缺省值。
捕捉并执行一个异样解决办法。
捕捉并动静计算一个候补值来顶替。
捕捉，并再包装为某一个业务相干的异样，而后再抛出业务异样。
捕捉，记录谬误日志，而后持续抛出。
应用 finally 来清理资源

以上所有这些在 Reactor 都有等效的操作符解决形式。

与第 (1) 条（捕捉并返回一个动态的缺省值）对应的是 onErrorReturn：

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

依据谬误类型返回对应值

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

与第 (2、3、4) 条（捕捉并执行一个异样解决办法）对应的是 onErrorResume

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)))    );

对应第 (5) 条（捕捉，记录谬误日志，并持续抛出）

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

对应（6）doFinally 在序列终止（无论是 onComplete、onError还是勾销）的时候被执行，并且可能判断是什么类型的终止事件

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

我须要哪个运算符

官网文档

A.1. 创立一个新序列
收回一个 T ,当曾经有值（不须要再去IO）： just

基于一个 Optional<T>：Mono.justOrEmpty(Optional<T>)

基于一个可能为 null 的 T：Mono.justOrEmpty(T)

收回一个 T ，且还是由 just 返回

然而须要 “懒”创立的：应用 Mono.fromSupplier 或用 defer 包装 just

//会起到提早执行的作用，在订阅时才去执行 func()
Mono.defer(()-> Mono.just(func()));
收回许多 T，这些元素我能够明确列举进去：Flux.just(T...)

基于迭代数据结构:

一个数组：Flux.fromArray

一个汇合或 iterable：Flux.fromIterable

一个 Integer 的 range：Flux.range

一个 Stream 提供给每一个订阅：Flux.fromStream(Supplier<Stream>)

基于一个参数值给出的源：

一个 Supplier<T>：Mono.fromSupplier

一个工作：Mono.fromCallable，Mono.fromRunnable

一个 CompletableFuture<T>：Mono#fromFuture

间接实现：empty

立刻生成谬误： error

然而“懒”创立的形式生成 Throwable.error(Supplier<Throwable>)

什么都不做：never

依赖一个可回收的资源：using

可编程地生成事件（能够应用状态）:

同步且一一的：Flux.generate

异步（也可同步）的，每次尽可能多收回元素：Flux.create （Mono#create 也是异步的，只不过只能发一个）

A.2. 对现有序列进行转化
我想转化一个序列：

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 或者 Flux.mergeWith(other)

元素类型会发生变化：Flux.zip / Flux.zipWith

将元素组合：

2个 Mono 组成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 (otherMono).or(thirdMono)，`flux.or(otherFlux).or(thirdFlux)

由一个序列触发（相似于 flatMap，不过“喜新厌旧”）：switchMap

由每个新序列开始时触发（也是“喜新厌旧”格调）：switchOnNext

我想反复一个序列： repeat

以肯定的距离反复：Flux.interval(duration).flatMap(tick -> myExistingPublisher)

我有一个空序列，然而…

我想要一个缺省值来代替：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)

A.3. “窥视”（只读）序列
在不对序列造成扭转的状况下，我想：

失去告诉或执行一些操作：

收回元素：doOnNext

序列实现：Flux#doOnComplete，Mono#doOnSuccess

因谬误终止：doOnError

勾销：doOnCancel

订阅时：doOnSubscribe

申请时：doOnRequest

实现或谬误终止： doOnTerminate（Mono的办法可能蕴含有后果）

然而在终止信号向上游传递之后：doAfterTerminate

所有类型的信号（Signal）：Flux.doOnEach

所有完结的状况（实现complete、谬误error、勾销cancel）：doFinally

记录日志：log

我想晓得所有的事件:

每一个事件都体现为一个 single 对象：

执行 callback：doOnEach

每个元素转化为 single对象： materialize

…在转化回元素：dematerialize

转化为一行日志：log

A.4. 过滤序列
我想过滤一个序列

基于给定的判断条件： 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

A.5. 错误处理
我想创立一个谬误序列： error

…替换一个实现的 Flux：.concat(Flux.error(e))

…替换一个实现的 Mono：.then(Mono.error(e))

…如果元素超时未收回：timeout

…“懒”创立：error(Supplier<Throwable>)

我想要相似 try/catch 的表达方式：

抛出异样：error

捕捉异样：

而后返回缺省值：onErrorReturn

而后返回一个 Flux 或 Mono：onErrorResume

包装异样后再抛出：.onErrorMap(t -> new RuntimeException(t))

finally 代码块：doFinally

Java 7 之后的 try-with-resources 写法：using 工厂办法

我想从谬误中复原…

返回一个缺省的：

的值：onErrorReturn

Publisher：Flux.onErrorResume 和 Mono.onErrorResume

重试： retry

…由一个用于随同 Flux 触发：retryWhen

我想解决背压谬误（向上游收回“MAX”的 request，如果上游的 request 比拟少，则利用策略）…

抛出 IllegalStateException：Flux#onBackpressureError

抛弃策略： Flux.onBackpressureDrop

…然而不抛弃最初一个元素：Flux.onBackpressureLatest

缓存策略（无限或有限）： Flux.onBackpressureBuffer

…当无限的缓存空间用满则利用给定策略：Flux.onBackpressureBuffer 带有策略 BufferOverflowStrategy

A.6. 基于工夫的操作
我想将元素转换为带有工夫信息的 Tuple2<Long, T>

从订阅时开始：elapsed

记录时间戳：timestamp

如果元素间提早过长则停止序列：timeout

以固定的周期收回元素：Flux.interval

在肯定的提早后收回 0：static Mono.delay

我想引入提早：

对每一个元素：Mono.delayElement，Flux.delayElements

提早订阅：delaySubscription

A.7. 拆分 Flux

我想将一个 Flux<T> 拆分为一个 Flux<Flux<T>>：

以个数为界： window(int)

…会呈现重叠或抛弃的状况：window(int, int)

以工夫为界： window(Duration)

…会呈现重叠或抛弃的状况：window(Duration, Duration)

以个数或工夫为界：windowTimeout(int, Duration)

基于对元素的判断条件： windowUntil

…触发判断条件的元素会分到下一波（cutBefore 变量）：.windowUntil(predicate, true)

…满足条件的元素在一波，直到不满足条件的元素收回开始下一波：windowWhile （不满足条件的元素会被抛弃）

通过另一个 Publisher 的每一个 onNext 信号来拆分序列：window(Publisher)，windowWhen

我想将一个 Flux<T>的元素拆分到汇合…

拆分为一个一个的 List:

以个数为界： buffer(int)

…会呈现重叠或抛弃的状况：buffer(int, int)

以工夫为界： buffer(Duration)

…会呈现重叠或抛弃的状况：buffer(Duration, Duration)

以个数或工夫为界：bufferTimeout(int, Duration)

基于对元素的判断条件： bufferUntil(Predicate)

…触发判断条件的元素会分到下一个buffer：.bufferUntil(predicate, true)

…满足条件的元素在一个buffer，直到不满足条件的元素收回开始下一buffer：bufferWhile(Predicate)

通过另一个 Publisher 的每一个 onNext 信号来拆分序列：buffer(Publisher)，bufferWhen

拆分到指定类型的 "collection"：buffer(int, Supplier<C>)

我想将 Flux<T> 中具备独特特色的元素分组到子 Flux：groupBy(Function<T,K>) TIP：留神返回值是 Flux<GroupedFlux<K, T>>，每一个 GroupedFlux 具备雷同的 key 值 K，能够通过 key() 办法获取。

A.8. 回到同步的世界
我有一个 Flux<T>，我想：

在拿到第一个元素前阻塞： Flux.blockFirst

…并给出超时时限：Flux.blockFirst(Duration)

在拿到最初一个元素前阻塞（如果序列为空则返回 null）： Flux#blockLast

…并给出超时时限：Flux.blockLast(Duration)

同步地转换为 Iterable<T>：Flux.toIterable

同步地转换为 Java 8 Stream<T>：Flux.toStream

我有一个 Mono<T>，我想：

在拿到元素前阻塞： Mono.block

…并给出超时时限：Mono#block(Duration)

转换为 CompletableFuture<T>：Mono.toFuture