【RxJava】- 結(jié)合操作符源碼分析

目錄

【RxJava】- 創(chuàng)建操作符源碼分析
【RxJava】- 變換操作符源碼分析
【RxJava】- 過(guò)濾操作符源碼分析
【RxJava】- 連接操作符源碼分析

CombineLatest

作用于最近發(fā)射的數(shù)據(jù)項(xiàng):如果Observable1發(fā)射了A并且Observable2發(fā)射了B和C,combineLatest()將會(huì)分組處理AB和AC据块,處理方式是使用一個(gè)函數(shù)結(jié)合每個(gè)Observable發(fā)射的最近數(shù)據(jù)項(xiàng),并且基于這個(gè)函數(shù)的結(jié)果發(fā)射數(shù)據(jù)闪朱。

RxJava將這個(gè)操作符實(shí)現(xiàn)為combineLatest,它接受二到九個(gè)Observable作為參數(shù)谆趾,或者單個(gè)Observables列表作為參數(shù)庵芭。它默認(rèn)不在任何特定的調(diào)度器上執(zhí)行。

ObservableCombineLatest

實(shí)現(xiàn)
看一下ObservableCombineLatest$LatestCoordinator中的innerNext闹蒜,combineLatest操作符實(shí)現(xiàn)邏輯在這個(gè)內(nèi)部類中。

void innerNext(int index, T item) {
    boolean shouldDrain = false;
    synchronized (this) {
        Object[] latest = this.latest;
        if (latest == null) {
            return;
        }
        Object o = latest[index];
        int a = active;
        if (o == null) {
            active = ++a;
        }
        latest[index] = item;
        if (a == latest.length) {
            queue.offer(latest.clone());
            shouldDrain = true;
        }
    }
    if (shouldDrain) {
        drain();

this.latest數(shù)組是在創(chuàng)建LatestCoordinator對(duì)象時(shí)抑淫,根據(jù)combineLatest方法Observable參數(shù)個(gè)數(shù)為數(shù)組長(zhǎng)度創(chuàng)建的數(shù)組绷落。

每一個(gè)Observable(發(fā)射數(shù)據(jù)的Observable,即被觀察者)都有自己的index始苇,當(dāng)發(fā)射數(shù)據(jù)時(shí)砌烁,先從緩存數(shù)組取該索引對(duì)應(yīng)的值,如果存在者更新,如果不存在者計(jì)數(shù)active變量加1函喉,如果計(jì)數(shù)變量值等于數(shù)組長(zhǎng)度避归,表示每一個(gè)Observable都發(fā)射過(guò)數(shù)據(jù),那么這個(gè)時(shí)候?qū)?shù)組克隆管呵,然后調(diào)用 drain()方法梳毙。

void drain() {
    ...
    v = ObjectHelper.requireNonNull(combiner.apply(s), "The combiner returned a null value");
    ...
    a.onNext(v);
    ...
}

傳入的s就是上面數(shù)組保存的發(fā)射數(shù)據(jù),a是觀察者捐下,即訂閱了combineLatest消息的實(shí)例顿天。

Join

任何時(shí)候,只要在另一個(gè)Observable發(fā)射的數(shù)據(jù)定義的時(shí)間窗口內(nèi)蔑担,這個(gè)Observable發(fā)射了一條數(shù)據(jù),就結(jié)合兩個(gè)Observable發(fā)射的數(shù)據(jù)咽白。由

ObservableJoin

實(shí)現(xiàn)啤握。

  • source
    第一個(gè)被觀察者
  • other(ObservableSource)
    第二個(gè)參入合并的被觀察者。
  • leftEnd(Function)
    選擇當(dāng)前發(fā)射的每個(gè)項(xiàng)目的持續(xù)時(shí)間的函數(shù)晶框,用于判斷重疊的部分排抬。
  • rightEnd(Function)
    選擇當(dāng)前發(fā)射的每個(gè)項(xiàng)目的持續(xù)時(shí)間的函數(shù),用于判斷重疊的部分授段。
  • resultSelector(BiFunction)
    一個(gè)函數(shù)蹲蒲,該函數(shù)計(jì)算任意兩個(gè)對(duì)象的結(jié)果。
subscribeActual方法
@Override
protected void subscribeActual(Observer<? super R> observer) {
   JoinDisposable<TLeft, TRight, TLeftEnd, TRightEnd, R> parent =new JoinDisposable<>(observer, leftEnd, rightEnd, resultSelector);
  // 通知觀察者侵贵,并傳入parent參數(shù)
   observer.onSubscribe(parent);
   LeftRightObserver left = new LeftRightObserver(parent, true);
  // disposables到JoinDisposable集合中
   parent.disposables.add(left);
   LeftRightObserver right = new LeftRightObserver(parent, false);
   // disposables到JoinDisposable集合中
   parent.disposables.add(right);  
   // 調(diào)用第一個(gè)被觀察者的subscribe方法
   source.subscribe(left);
   // 第二個(gè)加入的被觀察者subscribe方法
   other.subscribe(right);
}

調(diào)用第兩個(gè)(left和right)被觀察者的subscribe

source.subscribe(left);

會(huì)執(zhí)行LeftRightObserver中的onNext方法

public void onNext(Object t) {
    parent.innerValue(isLeft, t);
}

插入數(shù)組隊(duì)列queue

queue.offer(isLeft ? LEFT_CLOSE : RIGHT_CLOSE, index);
  • 執(zhí)行drain()方法

    取出數(shù)據(jù)隊(duì)列第一個(gè)數(shù)據(jù)

    Integer mode = (Integer)q.poll();
    

    如果讀取到數(shù)據(jù)是null届搁,者進(jìn)入下一次循環(huán)。

    讀取第二個(gè)數(shù)據(jù)窍育,即我們發(fā)射的數(shù)據(jù)卡睦。

    Object val = q.poll();
    
    • mode == LEFT_VALUE
      說(shuō)明是左邊的數(shù)據(jù),即第一個(gè)被觀察者發(fā)射的數(shù)據(jù)漱抓。然后進(jìn)行類型轉(zhuǎn)換表锻。

      左邊的數(shù)據(jù)索引(leftIndex)自加,然后將該數(shù)據(jù)放進(jìn)該索引下的名為lefts的map集合中乞娄。

      調(diào)用leftEnd的apply對(duì)數(shù)據(jù)進(jìn)行包裝瞬逊,返回ObservableSource實(shí)例。

      創(chuàng)建LeftRightEndObserver實(shí)例仪或,將該實(shí)例加入JoinDisposable的disposables中确镊。

      LeftRightEndObserver end = new LeftRightEndObserver(this, true, idx);
      disposables.add(end);
      

      調(diào)用數(shù)據(jù)包裝實(shí)例的subscribe并傳入JoinDisposable實(shí)例。

      p.subscribe(end);
      

      接下來(lái)遍歷rights集合溶其,調(diào)用resultSelector并傳入兩個(gè)參與發(fā)射的數(shù)據(jù)骚腥。

      R w = Objects.requireNonNull(resultSelector.apply(left, right)
      

      通知觀察者

      a.onNext(w);
      
    • mode == RIGHT_VALUE與
      和mode == LEFT_VALUE順序相反。

    • mode == LEFT_CLOSE
      從lefts或者rights移除該索引下的數(shù)據(jù)瓶逃,disposables中移除該索引下的數(shù)據(jù)束铭,

LeftRightEndObserver

上面會(huì)調(diào)用數(shù)據(jù)包裝實(shí)例的subscribe方法廓块,那么在LeftRightEndObserver中做了什么操作呢?

public void onNext(Object t) {
    if (DisposableHelper.dispose(this)) {
         parent.innerClose(isLeft, this);
     }
}

首先結(jié)束訂閱關(guān)系契沫,調(diào)用innerClose方法带猴。

public void innerClose(boolean isLeft, LeftRightEndObserver index) {synchronized (this) {
   queue.offer(isLeft ? LEFT_CLOSE : RIGHT_CLOSE, index);}
   drain();
}

那么又回到上面的邏輯了。

使用舉例
private void join(){Observable.<Integer>create(emitter -> emitter.onNext(1))
      .join(observer -> observer.onNext(2), new Function<Integer, ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> apply(Integer integer) {
                return observer -> {
                      // 立即調(diào)用懈万,發(fā)射的數(shù)據(jù)會(huì)被移除拴清,最后觀察者得不到任何數(shù)據(jù)
                      // observer.onComplete();
                };
            }
        }, integer -> (ObservableSource<Integer>) observer -> {
            // 立即調(diào)用,發(fā)射的數(shù)據(jù)會(huì)被移除会通,最后觀察者得不到任何數(shù)據(jù)
            // observer.onComplete();
        }, (BiFunction<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) { }

            @Override
            public void onNext(@NonNull Integer integer) {
                // 輸出為:create operate--->join: 3
                System.out.println("create operate--->join: " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) { }

            @Override
            public void onComplete() { }
        });
    }
總結(jié)

LeftRightEndObserver作用就是將lefts口予,rights,disposables中對(duì)應(yīng)索引的數(shù)據(jù)移除涕侈,也就是數(shù)據(jù)有效的時(shí)間期限沪停,一旦被移除,就不能被通知給觀察者了裳涛。其實(shí)功能就是將兩個(gè)被觀察者(left和right)發(fā)射的數(shù)據(jù)由resultSelector(BiFunction)做變換后通知給觀察者木张。

比如你在p.subscribe(end)方法后,會(huì)執(zhí)行l(wèi)eftEnd或者rightEnd的subscribe方法端三,如果你這時(shí)立即調(diào)用LeftRightEndObserver的onNext或者onComplete方法舷礼,最終都是得不到resultSelector變換后的數(shù)據(jù)的。

所以為什么說(shuō)leftEnd和rightEnd是持續(xù)時(shí)間的函數(shù)郊闯。我們可以控制數(shù)據(jù)的有效時(shí)間妻献。

Merge

合并多個(gè)Observables的發(fā)射物。太多了团赁,自己查看源碼旋奢。如果想知道大概用法請(qǐng)參考:Merge

StartWith

在數(shù)據(jù)序列的開(kāi)頭插入一條指定的項(xiàng)。

Switch

將一個(gè)發(fā)射多個(gè)Observables的Observable轉(zhuǎn)換成另一個(gè)單獨(dú)的Observable然痊,后者發(fā)射那些Observables最近發(fā)射的數(shù)據(jù)項(xiàng)

Zip

通過(guò)一個(gè)函數(shù)將多個(gè)Observables的發(fā)射物結(jié)合到一起至朗,基于這個(gè)函數(shù)的結(jié)果為每個(gè)結(jié)合體發(fā)射單個(gè)數(shù)據(jù)項(xiàng)。下面講解一下ObservableZip實(shí)現(xiàn)剧浸,看一下里面的drain()方法實(shí)現(xiàn)

 public void onNext(T t) {
     queue.offer(t);
     parent.drain();
 }
public void drain() {
    ...
    for (;;) {
        for (;;) {
            int i = 0;
            int emptyCount = 0;
            for (ZipObserver<T, R> z : zs) {
                if (os[i] == null) {
                    boolean d = z.done;
                    T v = z.queue.poll();
                    boolean empty = v == null;
                    if (checkTerminated(d, empty, a, delayError, z)) {
                        return;
                    }
                    if (!empty) {
                        os[i] = v;
                    } else {
                        emptyCount++;
                    }
                } else {
                    if (z.done && !delayError) {
                        Throwable ex = z.error;
                        if (ex != null) {
                            cancelled = true;
                            cancel();
                            a.onError(ex);
                            return;
                        }
                    }
                }
                i++;
            }
            if (emptyCount != 0) {
                break;
            }
            R v;
            try {
                v = Objects.requireNonNull(zipper.apply(os.clone()), "The zipper returned a null value");
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                cancel();
                a.onError(ex);
                return;
            }
            a.onNext(v);
            Arrays.fill(os, null);
        }
        missing = addAndGet(-missing);
        if (missing == 0) {
            return;
        }
    }
}

精華邏輯全在者幾層for循環(huán)里面锹引,由里向外,第一層for循環(huán):遍歷observers唆香,而observers的長(zhǎng)度也是Observable參數(shù)個(gè)數(shù)嫌变,即有多少個(gè)發(fā)射的數(shù)據(jù)的Observable實(shí)例,而row保存發(fā)射的值的數(shù)組躬它。

如果Observable發(fā)射過(guò)數(shù)據(jù)腾啥,那么對(duì)應(yīng)取出來(lái)的值就不等于null,執(zhí)行else,如果沒(méi)有發(fā)生錯(cuò)誤或者終止倘待,那么繼續(xù)循環(huán)疮跑。

如果Observable沒(méi)有發(fā)射過(guò)數(shù)據(jù),者將數(shù)據(jù)保存到對(duì)應(yīng)的數(shù)據(jù)索引下凸舵。

第二層循環(huán)祖娘,if (emptyCount != 0)為true,者跳出第二次循環(huán)啊奄,否則將row保存的數(shù)據(jù)應(yīng)用變換函數(shù)渐苏,然后把變換后的值發(fā)射給觀察者。

emptyCount是在row數(shù)組中還有Observable沒(méi)有發(fā)射過(guò)數(shù)據(jù)的時(shí)候菇夸,會(huì)加1琼富。

第三層循環(huán),很簡(jiǎn)單庄新,自己查看公黑。

如果在地一層循環(huán)里面 ,f (os[i] == null)始終不成立摄咆,也就是說(shuō)所以O(shè)bservable都發(fā)射過(guò)數(shù)據(jù)了,那么調(diào)用變換函數(shù)人断,然后發(fā)射變換后的值吭从。

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
  • 序言:七十年代末,一起剝皮案震驚了整個(gè)濱河市恶迈,隨后出現(xiàn)的幾起案子涩金,更是在濱河造成了極大的恐慌,老刑警劉巖暇仲,帶你破解...
    沈念sama閱讀 222,729評(píng)論 6 517
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件步做,死亡現(xiàn)場(chǎng)離奇詭異,居然都是意外死亡奈附,警方通過(guò)查閱死者的電腦和手機(jī)全度,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 95,226評(píng)論 3 399
  • 文/潘曉璐 我一進(jìn)店門(mén),熙熙樓的掌柜王于貴愁眉苦臉地迎上來(lái)斥滤,“玉大人将鸵,你說(shuō)我怎么就攤上這事∮悠模” “怎么了顶掉?”我有些...
    開(kāi)封第一講書(shū)人閱讀 169,461評(píng)論 0 362
  • 文/不壞的土叔 我叫張陵,是天一觀的道長(zhǎng)挑胸。 經(jīng)常有香客問(wèn)我痒筒,道長(zhǎng),這世上最難降的妖魔是什么? 我笑而不...
    開(kāi)封第一講書(shū)人閱讀 60,135評(píng)論 1 300
  • 正文 為了忘掉前任簿透,我火速辦了婚禮移袍,結(jié)果婚禮上,老公的妹妹穿的比我還像新娘萎战。我一直安慰自己咐容,他們只是感情好,可當(dāng)我...
    茶點(diǎn)故事閱讀 69,130評(píng)論 6 398
  • 文/花漫 我一把揭開(kāi)白布蚂维。 她就那樣靜靜地躺著戳粒,像睡著了一般。 火紅的嫁衣襯著肌膚如雪虫啥。 梳的紋絲不亂的頭發(fā)上蔚约,一...
    開(kāi)封第一講書(shū)人閱讀 52,736評(píng)論 1 312
  • 那天,我揣著相機(jī)與錄音涂籽,去河邊找鬼苹祟。 笑死,一個(gè)胖子當(dāng)著我的面吹牛评雌,可吹牛的內(nèi)容都是我干的树枫。 我是一名探鬼主播,決...
    沈念sama閱讀 41,179評(píng)論 3 422
  • 文/蒼蘭香墨 我猛地睜開(kāi)眼景东,長(zhǎng)吁一口氣:“原來(lái)是場(chǎng)噩夢(mèng)啊……” “哼砂轻!你這毒婦竟也來(lái)了?” 一聲冷哼從身側(cè)響起斤吐,我...
    開(kāi)封第一講書(shū)人閱讀 40,124評(píng)論 0 277
  • 序言:老撾萬(wàn)榮一對(duì)情侶失蹤搔涝,失蹤者是張志新(化名)和其女友劉穎,沒(méi)想到半個(gè)月后和措,有當(dāng)?shù)厝嗽跇?shù)林里發(fā)現(xiàn)了一具尸體庄呈,經(jīng)...
    沈念sama閱讀 46,657評(píng)論 1 320
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡,尸身上長(zhǎng)有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 38,723評(píng)論 3 342
  • 正文 我和宋清朗相戀三年派阱,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了诬留。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片。...
    茶點(diǎn)故事閱讀 40,872評(píng)論 1 353
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡贫母,死狀恐怖故响,靈堂內(nèi)的尸體忽然破棺而出,到底是詐尸還是另有隱情颁独,我是刑警寧澤彩届,帶...
    沈念sama閱讀 36,533評(píng)論 5 351
  • 正文 年R本政府宣布,位于F島的核電站誓酒,受9級(jí)特大地震影響樟蠕,放射性物質(zhì)發(fā)生泄漏贮聂。R本人自食惡果不足惜,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 42,213評(píng)論 3 336
  • 文/蒙蒙 一寨辩、第九天 我趴在偏房一處隱蔽的房頂上張望吓懈。 院中可真熱鬧,春花似錦靡狞、人聲如沸耻警。這莊子的主人今日做“春日...
    開(kāi)封第一講書(shū)人閱讀 32,700評(píng)論 0 25
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽(yáng)甘穿。三九已至,卻和暖如春梢杭,著一層夾襖步出監(jiān)牢的瞬間温兼,已是汗流浹背。 一陣腳步聲響...
    開(kāi)封第一講書(shū)人閱讀 33,819評(píng)論 1 274
  • 我被黑心中介騙來(lái)泰國(guó)打工武契, 沒(méi)想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留募判,地道東北人。 一個(gè)月前我還...
    沈念sama閱讀 49,304評(píng)論 3 379
  • 正文 我出身青樓咒唆,卻偏偏與公主長(zhǎng)得像届垫,于是被迫代替她去往敵國(guó)和親。 傳聞我的和親對(duì)象是個(gè)殘疾皇子全释,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 45,876評(píng)論 2 361

推薦閱讀更多精彩內(nèi)容

  • 注:只包含標(biāo)準(zhǔn)包中的操作符装处,用于個(gè)人學(xué)習(xí)及備忘參考博客:http://blog.csdn.net/maplejaw...
    小白要超神閱讀 2,199評(píng)論 2 8
  • 記錄RxJava操作符,方便查詢(2.2.2版本) 英文文檔地址:http://reactivex.io/docu...
    凌云飛魚(yú)閱讀 826評(píng)論 0 0
  • 一恨溜、RxJava操作符概述 RxJava中的操作符就是為了提供函數(shù)式的特性,函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡(jiǎn)潔易...
    測(cè)天測(cè)地測(cè)空氣閱讀 637評(píng)論 0 1
  • 創(chuàng)建操作 用于創(chuàng)建Observable的操作符Create通過(guò)調(diào)用觀察者的方法從頭創(chuàng)建一個(gè)ObservableEm...
    rkua閱讀 1,836評(píng)論 0 1
  • 一找前、RxJava操作符概述 RxJava中的操作符就是為了提供函數(shù)式的特性糟袁,函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡(jiǎn)潔易...
    無(wú)求_95dd閱讀 3,123評(píng)論 0 21