Rxjava 過程分析二（操作符流程）

說明

• 主要分析操作符的設(shè)計流程， 只要關(guān)注主要環(huán)節(jié)， 大致流程。
• 由于每個操作符背后的設(shè)計基本一致， 所以隨便選出一個分析即可。 那我們肯定選最簡單的 map 啦哈。

基本使用

Flowable.create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> emitter) throws Exception {
    // emitter.onNext("");
    // emitter.onError();
    // emitter.onComplete();
}
}, BackpressureStrategy.LATEST)
    .map(new Function<String, String>() {
        @Override
        public String apply(String s) throws Exception {
            return s;
        }
    })
    .subscribe(new FlowableSubscriber<String>() {
        @Override
        public void onSubscribe(Subscription s) {
        }
        @Override
        public void onNext(String s) {
        }
        @Override
        public void onError(Throwable t) {
        }

        @Override
        public void onComplete() {
        }
    });

上面一堆和下面一堆我們在上一篇已經(jīng)見過， 我們現(xiàn)在主要關(guān)注與中間那一堆， 也就是 map 的實現(xiàn)。

引發(fā)的思考

• 操作符是怎么承上啟下連接上游和下游的呢？
• Flowable 經(jīng)過操作符轉(zhuǎn)換后再進(jìn)行點操作， 還是同一個 Flowable 嗎？

源碼分析

 public final <R> Flowable<R> map(Function<? super T, ? extends R> mapper) {
    return new FlowableMap<T, R>(this, mapper);
}

很簡答嘛， 新建了一個 FlowableMap 而已, 把當(dāng)前的 Flowable 和 怎么轉(zhuǎn)換的對象（也就是接口， 提供除去讓用戶怎么實現(xiàn)怎么轉(zhuǎn)換）傳遞進(jìn)去， 并對成員變量賦值。

我們在 map 后去訂閱了下游， 那么有我們上一篇分析的結(jié)果來開， 在訂閱的時候是調(diào)用了 FlowableMap 的 subscribeActual。 所以我們看看里面做了什么。

 protected void subscribeActual(Subscriber<? super U> s) {
    source.subscribe(new MapSubscriber<T, U>(s, mapper));
}

等等這個 source 是什么呢， 好像在這里又發(fā)生了一次訂閱了對吧？ 對的。 這個 source 就是我們在實例化 FlowableMap 傳進(jìn)來的 this, 也就是上一個 Flowable 實例， 在這里是 FlowableCreate。 什么鬼！ 怎么又發(fā)生了一次訂閱呢。 那么好， 我們試著去分析下， 我們發(fā)生訂閱后結(jié)合上片文章的分析， 發(fā)生訂閱后， 會調(diào)用到 FlowableMap 中的 subscribeActual 最后不管是回調(diào)還是返回， 肯定是調(diào)用的相應(yīng)的方法如 onNext 等都會調(diào)用到 這里的 MapSubscriber 里相應(yīng)的方法。 你發(fā)現(xiàn)沒， 這里 MapSubscriber 就變聲了一個臨時中間節(jié)點的一個下游。 那我們這個中間臨時下游的 onNext 做了什么事情。

public void onNext(T t) {
    U v;
    try {
        v = mapper.apply(t);
    } catch (Throwable ex) {
        return;
    }
    downstream.onNext(v);
}

mapper 就是我們傳入 map 中的接口， 在這里調(diào)用， 從上層 Flowable 中拿到的結(jié)果， 通過 map 里的這個 function 接口去回調(diào)給用戶去實現(xiàn)， 然后把轉(zhuǎn)換后的值拿到直接塞給了最后的下游的 onNext。 你明白了沒， 原理就是這么簡單。 到這里前前后后都連接通了吧！

前面的疑惑問題

• 操作符是怎么承上啟下的呢？

沒一個操作符返回的都是自己實現(xiàn)的一個 Flowable， 然后保留了上一層的 Flowable， 在發(fā)生訂閱時會一級一級的發(fā)生訂閱， 頂層的 FlowableCreate 調(diào)用的方法， 會間接的調(diào)用到了操作符相應(yīng)的方法， 然后操作的相應(yīng)方法除了實現(xiàn)自己的業(yè)務(wù)， 然后把結(jié)果傳給了底層的 Flowable 或者下游。

• Flowable 經(jīng)過操作符后還是同一個 Flowable 嗎？

很明顯不是