1. 程式人生 > >(18)Hot vs Cold——響應式Spring的道法術器

(18)Hot vs Cold——響應式Spring的道法術器

Spring WebFlux 響應式編程

本系列文章索引《響應式Spring的道法術器》
前情提要 響應式流 | Reactor 3快速上手 | 響應式流規範
本文測試源碼

2.8 Hot vs Cold

到目前為止,我們討論的發布者,無論是Flux還是Mono,都有一個特點:訂閱前什麽都不會發生。當我們“創建”了一個Flux的時候,我們只是“聲明”/“組裝”了它,但是如果不調用.subscribe來訂閱它,它就不會開始發出元素。

但是我們對“數據流”(尤其是乍聽到這個詞的時候)會有種天然的感覺,就是無論有沒有訂閱者,它始終在按照自己的步伐發出數據。就像假設一個人沒有一個粉絲,他也可以發微博一樣。

以上這兩種數據流分別稱為“冷”序列和“熱”序列。所以我們一直在介紹的Reactor3的發布者就屬於“冷”的發布者。不過有少數的例外,比如just

生成的就是一個“熱”序列,它直接在組裝期就拿到數據,如果之後有誰訂閱它,就重新發送數據給訂閱者。Reactor 中多數其他的“熱”發布者是擴展自Processor 的(下節會介紹到)。

下面我們通過對比了解一下兩種不同的發布者的效果,首先是我們熟悉的“冷”發布者:

    @Test
    public void testCodeSequence() {
        Flux<String> source = Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
                .map(String::toUpperCase);

        source.subscribe(d -> System.out.println("Subscriber 1: "+d));
        System.out.println();
        source.subscribe(d -> System.out.println("Subscriber 2: "+d));
    }

我們對發布者source進行了兩次訂閱,每次訂閱都導致它把數據流從新發一遍:

Subscriber 1: BLUE
Subscriber 1: GREEN
Subscriber 1: ORANGE
Subscriber 1: PURPLE

Subscriber 2: BLUE
Subscriber 2: GREEN
Subscriber 2: ORANGE
Subscriber 2: PURPLE

然後再看一個“熱”發布者的例子:

    @Test
    public void testHotSequence() {
        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();
    }

這個熱發布者是一個UnicastProcessor,我們可以使用它的onNext等方法手動發出元素。上邊的例子中,hotSource發出兩個元素後第二個訂閱者才開始訂閱,所以第二個訂閱者只能收到之後的元素:

Subscriber 1 to Hot Source: BLUE
Subscriber 1 to Hot Source: GREEN
Subscriber 1 to Hot Source: ORANGE
Subscriber 2 to Hot Source: ORANGE
Subscriber 1 to Hot Source: PURPLE
Subscriber 2 to Hot Source: PURPLE

由此可見,UnicastProcessor是一個熱發布者。

有時候,你不僅想要在某一個訂閱者訂閱之後才開始發出數據,可能還希望在多個訂閱者“到齊”之後 才開始。ConnectableFlux的用意便在於此。Flux API 中有兩種常用的返回ConnectableFlux 的方式:publishreplay

  1. publish會嘗試滿足各個不同訂閱者的需求(也就是回壓),並綜合這些請求反饋給源。假設有某個訂閱者的需求為 0,發布者會暫停向所有訂閱者發出元素。
  2. replay將對第一個訂閱後產生的數據進行緩存,最多緩存數量取決於配置(時間/緩存大小)。 它會對後續接入的訂閱者重新發送數據。

ConnectableFlux提供了多種對訂閱的管理方式。包括:

  • connect,當有足夠的訂閱接入後,可以對 flux 手動執行一次。它會觸發對上遊源的訂閱。
  • autoConnect(n)connect類似,不過是在有 n 個訂閱的時候自動觸發。
  • refCount(n)不僅能夠在訂閱者接入的時候自動觸發,還會檢測訂閱者的取消動作。如果訂閱者全部取消訂閱,則會將源“斷開連接”,再有新的訂閱者接入的時候才會繼續“連上”發布者。refCount(int, Duration)增加了一個倒計時:一旦訂閱者數量太低了,它會等待 Duration 參數指定的時間,如果沒有新的訂閱者接入才會與源斷開連接。

1)connect的例子

    @Test
    public void testConnectableFlux1() throws InterruptedException {
        Flux<Integer> source = Flux.range(1, 3)
                .doOnSubscribe(s -> System.out.println("上遊收到訂閱"));

        ConnectableFlux<Integer> co = source.publish();

        co.subscribe(System.out::println, e -> {}, () -> {});
        co.subscribe(System.out::println, e -> {}, () -> {});

        System.out.println("訂閱者完成訂閱操作");
        Thread.sleep(500);
        System.out.println("還沒有連接上");

        co.connect();
    }

輸出如下:

訂閱者完成訂閱操作
還沒有連接上
上遊收到訂閱
1
1
2
2
3
3

可見當connect的時候,上遊才真正收到訂閱請求。

2)autoConnect的例子

    @Test
    public void testConnectableFluxAutoConnect() throws InterruptedException {
        Flux<Integer> source = Flux.range(1, 3)
                .doOnSubscribe(s -> System.out.println("上遊收到訂閱"));

        // 需要兩個訂閱者才自動連接
        Flux<Integer> autoCo = source.publish().autoConnect(2);

        autoCo.subscribe(System.out::println, e -> {}, () -> {});
        System.out.println("第一個訂閱者完成訂閱操作");
        Thread.sleep(500);
        System.out.println("第二個訂閱者完成訂閱操作");
        autoCo.subscribe(System.out::println, e -> {}, () -> {});
    }

輸出如下:

第一個訂閱者完成訂閱操作
第二個訂閱者完成訂閱操作
上遊收到訂閱
1
1
2
2
3
3

可見,只有兩個訂閱者都完成訂閱之後,上遊才收到訂閱請求,並開始發出數據。

3)refCononect的例子

    @Test
    public void testConnectableFluxRefConnect() throws InterruptedException {

        Flux<Long> source = Flux.interval(Duration.ofMillis(500))
                .doOnSubscribe(s -> System.out.println("上遊收到訂閱"))
                .doOnCancel(() -> System.out.println("上遊發布者斷開連接"));

        Flux<Long> refCounted = source.publish().refCount(2, Duration.ofSeconds(2));

        System.out.println("第一個訂閱者訂閱");
        Disposable sub1 = refCounted.subscribe(l -> System.out.println("sub1: " + l));

        TimeUnit.SECONDS.sleep(1);
        System.out.println("第二個訂閱者訂閱");
        Disposable sub2 = refCounted.subscribe(l -> System.out.println("sub2: " + l));

        TimeUnit.SECONDS.sleep(1);
        System.out.println("第一個訂閱者取消訂閱");
        sub1.dispose();

        TimeUnit.SECONDS.sleep(1);
        System.out.println("第二個訂閱者取消訂閱");
        sub2.dispose();

        TimeUnit.SECONDS.sleep(1);
        System.out.println("第三個訂閱者訂閱");
        Disposable sub3 = refCounted.subscribe(l -> System.out.println("sub3: " + l));

        TimeUnit.SECONDS.sleep(1);
        System.out.println("第三個訂閱者取消訂閱");
        sub3.dispose();

        TimeUnit.SECONDS.sleep(3);
        System.out.println("第四個訂閱者訂閱");
        Disposable sub4 = refCounted.subscribe(l -> System.out.println("sub4: " + l));
        TimeUnit.SECONDS.sleep(1);
        System.out.println("第五個訂閱者訂閱");
        Disposable sub5 = refCounted.subscribe(l -> System.out.println("sub5: " + l));
        TimeUnit.SECONDS.sleep(2);
    }

輸出如下:

第一個訂閱者訂閱
第二個訂閱者訂閱
上遊收到訂閱
sub1: 0
sub2: 0
第一個訂閱者取消訂閱
sub1: 1
sub2: 1
sub2: 2
第二個訂閱者取消訂閱
sub2: 3
第三個訂閱者訂閱
sub3: 6
sub3: 7
第三個訂閱者取消訂閱
上遊發布者斷開連接
第四個訂閱者訂閱
第五個訂閱者訂閱
上遊收到訂閱
sub4: 0
sub5: 0
sub4: 1
sub5: 1
sub4: 2
sub5: 2
sub4: 3
sub5: 3

本例中,refCount設置為最少兩個訂閱者接入是才開始發出數據,當所有訂閱者都取消時,如果不能在兩秒內接入新的訂閱者,則上遊會斷開連接。

上邊的例子中,隨著前兩個訂閱者相繼取消訂閱,第三個訂閱者及時(在2秒內)開始訂閱,所以上遊會繼續發出數據,而且根據輸出可以看出是“熱序列”。

當第三個訂閱者取消後,第四個訂閱者沒能及時開始訂閱,所以上遊發布者斷開連接。當第五個訂閱者訂閱之後,第四和第五個訂閱者相當於開始了新一輪的訂閱。

(18)Hot vs Cold——響應式Spring的道法術器