聲明：

一股耽、subscribe方法

注意：所有的操作只有在訂閱的那一刻才開始進(jìn)行８小！物蝙！
subscribe方法有兩種常用的形式：

• Disposable subscribe(@Nullable Consumer<? super T> consumer, @Nullable Consumer<? super Throwable> errorConsumer, @Nullable Runnable completeConsumer, @Nullable Consumer<? super Subscription> subscriptionConsumer)
  這四個(gè)參數(shù)的意義如下：
  1）consumer：收到一個(gè)數(shù)據(jù)時(shí)的回調(diào)
  2）errorConsumer：上游報(bào)告出現(xiàn)錯(cuò)誤信號(hào)時(shí)的回調(diào)
  3）completeConsumer：上游報(bào)告完成信號(hào)時(shí)的回調(diào)
  4）subscriptionConsumer：提供一個(gè)Subscription類型的對(duì)象炎滞，你可以使用他對(duì)上游流量進(jìn)行反饋控制
  Disposable返回類型：使用Disposable#dispose方法可以取消訂閱行為，在Flux和Mono中诬乞，取消意味著告知數(shù)據(jù)源不再生產(chǎn)數(shù)據(jù)厂榛，但這種取消行為并不一定是及時(shí)的，也許數(shù)據(jù)源產(chǎn)生數(shù)據(jù)非忱霾眩快，在接收到取消信號(hào)之前就完成所有數(shù)據(jù)的生成辈双。
  Disposable有一個(gè)工具類Disposables责掏，它主要提供了Disposable.Swap和Disposable.Composite兩種包裝類的工廠方法，前者允許你對(duì)一個(gè)Disposable進(jìn)行替換（不取消容器中當(dāng)前的Disposable湃望，僅作替換）以及更新（取消容器中當(dāng)前的Disposable换衬，并替換為新的Dsiposable）操作，后者允許你一次性控制多個(gè)Disposable
• void subscribe(Subscriber<? super T> actual)
Subscriber是一個(gè)訂閱器证芭，里面包含了onSubscribe瞳浦、onNext、onError废士、onError四種方法叫潦，使用它們可以更加方便的控制訂閱的操作。當(dāng)然實(shí)際中使用的還是BaseSubscriber官硝，這個(gè)類繼承了Subscriber矗蕊，但是它是抽象的短蜕，無(wú)法直接實(shí)例化。使用時(shí)直接使用匿名類進(jìn)行實(shí)例化就好傻咖，這樣可以很好避免一個(gè)BaseSubscriber實(shí)例同時(shí)作為兩個(gè)不同訂閱操作的訂閱器所帶來(lái)的異常朋魔，因?yàn)楦鶕?jù)Reactive Stream規(guī)則來(lái)說(shuō)，一個(gè)訂閱器中的onNext方法只能被時(shí)序調(diào)用卿操，而不同同時(shí)調(diào)用警检。
  你可以看下面這個(gè)例子：

public class CustomSubsriber<T> extends BaseSubscriber<T> {
    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        System.out.println("subscribed");
        request(1);
    }

    @Override
    protected void hookOnNext(T value) {
        System.out.println("get value:" + value);
        LockSupport.parkNanos(Duration.ofSeconds(1).toNanos());
        request(1);
    }

    @Override
    protected void hookOnComplete() {
        System.out.println("completed");
    }
}

二、背壓——流控

在Reactor中害淤，下游想要控制上游的流速扇雕，是通過(guò)request來(lái)實(shí)現(xiàn)的，request的總數(shù)代表著下游當(dāng)前的需求量筝家，比如subscriber.request(1)就代表下游需要一個(gè)數(shù)據(jù)洼裤，多了不要。如果你把request的值設(shè)為Long.MAX_VALUE溪王，則它意味著下游可以接收無(wú)限制的數(shù)據(jù)腮鞍，比如Mono#block、Flux#blockFirst莹菱、Flux#blockLast就默認(rèn)無(wú)限制接收數(shù)據(jù)移国。對(duì)了，這三個(gè)方法也是一種訂閱操作道伟，使用它們會(huì)激活整個(gè)訂閱過(guò)程迹缀，Mono#block和Flux#blockFirst均表示接收流中第一個(gè)數(shù)據(jù)并返回，如果在等待中接收到了完成信號(hào)則返回null蜜徽，同理Flux#blockLast表示只接收最后一個(gè)數(shù)據(jù)祝懂，其會(huì)一直等待完成信號(hào)的到來(lái)。
值得注意的是：request并不一定恒定的拘鞋，它可能會(huì)被整個(gè)上游中的某個(gè)操作修改砚蓬，比如stringFlux.buffer(3).subscriber(null,null,null,sub -> {sub.request(2)})中，buffer(3)會(huì)將request進(jìn)一步修改為6盆色，因?yàn)?code>sub.request(2)的2個(gè)請(qǐng)求是請(qǐng)求buffer(3)的輸出結(jié)果的灰蛙，而一個(gè)buffer(3)結(jié)果需要其上游的3個(gè)數(shù)據(jù)，故request被更改為了6

• limitRequest也是一種流控操作隔躲，它接收一個(gè)參數(shù)作為“最大請(qǐng)求量”摩梧，如果下游總的請(qǐng)求量并沒有到達(dá)上限，那么一切照常宣旱，否則仅父，在到達(dá)那一刻時(shí)，改操作將會(huì)給上游發(fā)出取消信號(hào)，并給下游發(fā)出完成信號(hào)
• limitRate將請(qǐng)求分組驾霜，比如下游請(qǐng)求100個(gè)request(100)案训，那么limitRate(10)將會(huì)把請(qǐng)求分為10個(gè)request(10)，每完成一個(gè)request(10)則自動(dòng)發(fā)出下一個(gè)request(10)

三粪糙、異步

在Reactor中强霎，如果不特別指定異步操作的話，那么整個(gè)流的發(fā)生到訂閱過(guò)程全部會(huì)執(zhí)行在subscribe那個(gè)線程中蓉冈。所以最簡(jiǎn)單的異步使用Reactor的方法就是新建一個(gè)線程城舞，并在其中執(zhí)行subscribe，比如：

public static void main(String[] args) throws InterruptedException {
  final Mono<String> mono = Mono.just("hello "); 

  Thread t = new Thread(() -> mono
      .map(msg -> msg + "thread ")
      .subscribe(v -> 
          System.out.println(v + Thread.currentThread().getName()) 
      )
  )
  t.start();
  t.join();
}

其結(jié)果為：

hello thread Thread-0

當(dāng)然Reactor其實(shí)提供更加簡(jiǎn)便的異步操作方式寞酿，其中比較常用的就是publishOn和SubscribeOn兩個(gè)方法了家夺，這兩個(gè)方法都需要一個(gè)Scheduler類型的參數(shù)，它控制著操作的執(zhí)行模式以及執(zhí)行執(zhí)行位置伐弹，單從表現(xiàn)上來(lái)看拉馋，倒是有點(diǎn)像ExecutorService。創(chuàng)建Scheduler你需要使用到Schedulers工廠類惨好，里面定義了許多不同類型的Scheduler：
1）Schedulers.immediate():直接在當(dāng)前線程中立刻執(zhí)行
2）Schedulers.single()/newSingle():提供一個(gè)單線程線程池以供操作煌茴，前者是一個(gè)固定的定義好的單線程線程池，后者你可以使用它來(lái)創(chuàng)建新的單線程線程池
3）Schedulers.elastic():相當(dāng)于提供了一個(gè)CachedThreadPool
4）Schedulers.parallel():相當(dāng)于提供了一個(gè)和Cpu核心數(shù)一樣多的核心線程數(shù)的線程池
5）Schedulers.fromExecutorService():從現(xiàn)有的ExecutorService中引入

• subscribeOn
subscribeOn會(huì)將整個(gè)流（包括數(shù)據(jù)源生成）放置在指定線程上執(zhí)行（具體由哪個(gè)線程執(zhí)行由前面所說(shuō)的Scheduler來(lái)控制）日川，注意蔓腐，無(wú)論subscribeOn放在哪，它都將影響整個(gè)流龄句，可以有多個(gè)subscribeOn回论，但是只有第一個(gè)會(huì)生效
• publishOn
publishOn會(huì)將其后的操作搬移到指定線程上執(zhí)行，注意分歇，publishOn優(yōu)先級(jí)比subscribeOn高傀蓉，而且多個(gè)publishOn都各自對(duì)其后的操作會(huì)有影響。

看個(gè)例子：

final Flux<String> flux = Flux
        .range(1, 2)
        .map(i -> {
            System.out.println("map1:"+Thread.currentThread().getName());
            LockSupport.parkNanos(Duration.ofSeconds(1).toNanos());
            return 10 + i;
        })
        .subscribeOn(s)
        .map(i -> {
            System.out.println("map2:"+Thread.currentThread().getName());
            LockSupport.parkNanos(Duration.ofSeconds(2).toNanos());
            return "value " + i;
        }).publishOn(Schedulers.single())
        .map(i -> {
            System.out.println("map3:"+Thread.currentThread().getName());
            return "mtk:"+i;
        });

new Thread(() -> flux.subscribe(System.out::println)).start();

結(jié)果如下：

map1:parallel-scheduler-1
map2:parallel-scheduler-1
map1:parallel-scheduler-1
map3:single-1
mtk:value 11
map2:parallel-scheduler-1
map3:single-1
mtk:value 12

從結(jié)果中我們可以發(fā)現(xiàn)整個(gè)流在沒有運(yùn)行在subscribe方法調(diào)用時(shí)所在的線程中职抡，因?yàn)橛?code>subscribeOn的緣故僚害，整個(gè)流運(yùn)行在parallel-scheduler-1線程中，但是在map3操作前繁调，有個(gè)publishOn，其使得其后的操作運(yùn)行在了single-1線程中

參考文檔：
[1] Reactor api doc
[2] Reactor reference doc