Reactor 操作符

上篇文章我們將 Flux 和 Mono 的操作符分了 11 類，我們來繼續(xù)學(xué)習(xí)轉(zhuǎn)換類操作符的第 2 篇。

轉(zhuǎn)換類操作符

轉(zhuǎn)換類的操作符數(shù)量最多，平常過程中也是使用最頻繁的。

Flux#concatMap

將響應(yīng)式流中元素順序轉(zhuǎn)換為目標(biāo)類型的響應(yīng)式流，之后再將這些流連接起來。該方法提供了 2 個(gè)重載方法，傳遞的第 2 個(gè)參數(shù)為內(nèi)部生成響應(yīng)式流的預(yù)取數(shù)量。見圖知意：

07_operator_flux_concatMap.png

Flux.range(3, 8)
    .concatMap(n -> Flux.just(n - 10, n, n + 10), 3)
    .subscribe(System.out::println);

Flux#concatMapDelayError

concatMapDelayError 和 concatMap 區(qū)別在于，當(dāng)內(nèi)部生成響應(yīng)式流發(fā)出 error 時(shí)，是否延遲響應(yīng) error 。該方法提供了 3 個(gè)重載方法，支持傳遞參數(shù)：是否延遲發(fā)出錯(cuò)誤和預(yù)取數(shù)量。

Flux.range(3, 8)
    .concatMapDelayError(n -> {
        if (n == 4) {
            return Flux.error(new NullPointerException());
        }
        return Flux.just(n - 10, n, n + 10);
    })
    .subscribe(System.out::println, System.err::println);

Flux#concatIterable

concatIterable 和 concatMap 的區(qū)別在于 內(nèi)部返回的類型不同，一個(gè)為 Iterable， 一個(gè)為 響應(yīng)式流。見圖知意：

08_operator_flux_concatIterable.png

Flux.range(3, 8)
    .publishOn(Schedulers.single())
    .concatMapIterable(n -> {
        if (n == 4) {
            throw new NullPointerException();
        }
        return Arrays.asList(n - 10, n, n + 10);
    })
    .onErrorContinue((e, n) -> System.err.println("數(shù)據(jù)：" + n + ",發(fā)生錯(cuò)誤：" + e))
    .subscribe(System.out::println);

elapsed

收集響應(yīng)式流中元素的間隔發(fā)出時(shí)間，轉(zhuǎn)換為 時(shí)間間隔 和 舊元素 組成的 Tuple2 的響應(yīng)式流。見圖知意：

09_operator_flux_elapsed.png

Flux.interval(Duration.ofMillis(300))
    .take(20)
    .elapsed(Schedulers.parallel())
    .subscribe(System.out::println);
Thread.sleep(7000);

expand

從上層節(jié)點(diǎn)逐層展開方式遞歸展開樹形節(jié)點(diǎn)。

Flux.just(16, 18, 20)
    .expand(n -> {
        if (n % 2 == 0) {
            return Flux.just(n / 2);
        } else {
            return Flux.empty();
        }
    })
    .subscribe(System.out::println);

expandDeep

從上層節(jié)點(diǎn)逐個(gè)展開方式遞歸展開樹形節(jié)點(diǎn)。expand 和 expandDeep 的區(qū)別在于展開方式不同，另外它倆都提供了 capacityHint 指定遞歸時(shí)初始化容器的容量。

Flux.just(16, 18, 20)
    .expandDeep(n -> {
        if (n % 2 == 0) {
            return Flux.just(n / 2);
        } else {
            return Flux.empty();
        }
    })
    .subscribe(System.out::println);

總結(jié)

本篇我們介紹了 Reactor 部分的轉(zhuǎn)換類操作符，講解示例時(shí)都是單個(gè)操作符，相信大家都能理解。

由于最近學(xué)習(xí)時(shí)間不確定，內(nèi)容比較少。無論工作還是生活的困難，我們只要堅(jiān)持，終將會(huì)被克服解決。今天的內(nèi)容就學(xué)到這里，我們下篇繼續(xù)學(xué)習(xí) Reactor 的操作符。

源碼詳見：https://github.com/crystalxmumu/spring-web-flux-study-note 下 02-reactor-core-learning
模塊下 ReactorTransformOperator02Test 測(cè)試類。