Rxjava2.0概述

通過前面的文章介紹闻书，讀者對RxJava2.0應該有了初步的認識壮啊。RxJava2.0相對1.0沒有增加新的功能节榜，最大的變化是把原來與背壓有關的操作符抽取出來放在Flowable中處理扭仁；另外2.0的接口定義更符合ReactiveStream的規(guī)范讼油。操作符的功能和1.0相比沒有太大的變化，不過Flowable是背壓相關的功能抽離出來的惑朦，本篇我們來詳細分析下各種操作符的作用和用法兽泄。每種類型的操作符會選擇幾個重點介紹，提醒讀者在閱讀的過程中注意操作符背壓的處理漾月。另外操作符都是支持泛型的病梢，對泛型不了解的讀者，需要先熟悉一下泛型相關的知識梁肿。

創(chuàng)建操作符

create操作符

這里首先介紹create操作符蜓陌，它是使用最廣泛的創(chuàng)建操作符。

create操作符的使用場景

create是最基本的操作符吩蔑，用來創(chuàng)建一個Flowable钮热，事件流的生產(chǎn)和下發(fā)由用戶自己定義，是一個完全可定制的Flowable哥纫。

create操作符的基本使用方法

  Flowable.create(new FlowableOnSubscribe<Object>() {
             @Override
             public void subscribe(@NonNull FlowableEmitter<Object> e) throws Exception {
                 e.onNext("1");
                 e.onNext("2")
            }
            }, BackpressureStrategy.BUFFER)
        .subscribe(new FlowableSubscriber<Object>() {
            Subscription  st=null;
            @Override
            public void onSubscribe(@NonNull Subscription s) {
                s.request(1);
                st =s;
            }

            @Override
            public void onNext(Object o) {
                System.out.print(s);
                st.request(1);
                
            }

            @Override
            public void onComplete() {

            }

            @Override
            public void onError(Throwable throwable) {

            }
        });

以上輸出的結(jié)果為：1 2
create方法的

• 第一個參數(shù)是FlowableOnSubscribe霉旗，它只有一個方法subscribe，subscribe是事件生產(chǎn)的地方蛀骇，subscribe的參數(shù)FlowableEmitter是內(nèi)部類厌秒，其中一個屬性是FlowableSubscriber，也就是我們定義的觀察者擅憔。通過調(diào)用 e.onNext("1")鸵闪，F(xiàn)lowableEmitter會調(diào)用觀察者onNext的方法即是FlowableSubscriber的onNext方法，從而完成事件下發(fā)給觀察者暑诸。
• 第二個參數(shù)是背壓策略蚌讼，RxJava2.0定義了五種背壓策略，后續(xù)文章會重點講述背壓策略的詳細區(qū)別个榕。背壓策略會影響觀察者能否正確接收到事件篡石。

create操作符事件的處理流程

以上面的代碼為例，當subscribe(new FlowableSubscriber<Object>()執(zhí)行時西采，會首先執(zhí)行onSubscribe(@NonNull Subscription s)方法凰萨，它執(zhí)行在調(diào)用者的線程，對于背壓來說械馆，這里需要告知生產(chǎn)者事件的下流觀察者的事件處理能力胖眷；然后是subscribe(@NonNull FlowableEmitter<Object> e）執(zhí)行事件的的生產(chǎn)邏輯，這里會判斷觀察者的處理數(shù)據(jù)的能力決定是否執(zhí)行觀察者的onNext方法霹崎。這里判斷的依據(jù)是s.request()的參數(shù)是否大于零珊搀。

create操作符需要注意的地方

前面也提到過，觀察者能否接收到事件取決于s.request()的參數(shù)是否大于零或者s.request()是否被調(diào)用尾菇。如果FlowableSubscriber的方法onSubscribe沒有調(diào)用過s.request(n)或者n<=0 ,FlowableSubscriber不會接收到事件境析。上面的例子中觀察者只能就收到事件"1", 而不能接收"2",因為只調(diào)用對s.request(1)囚枪，如果希望接收到兩次事件可以在onSubscribe調(diào)用s.request(2)，或者是onSubscribe調(diào)用s.request(1)并且onNext中調(diào)用s.request(1)簿晓。

fromArray操作符

fromArray操作符的使用場景

fromArray用于快速創(chuàng)建一個Flowable眶拉，直接發(fā)送數(shù)組的數(shù)據(jù)。

fromArray操作符的基本使用方法

  Flowable.fromArray(new String[]{"1","2","3"})
          .subscribe(new FlowableSubscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(3);
                    }
                    @Override
                    public void onNext(String s) {
                        System.out.print(s);
                    }
                    @Override
                    public void onError(Throwable throwable) {

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

以上輸出的結(jié)果為：1 2 3
fromArray的參數(shù)是一個數(shù)組憔儿，數(shù)組的元素會被依次發(fā)送出去忆植。

fromArray操作符的事件 執(zhí)行流程

1 從subscribe(new FlowableSubscriber<String>())方法開始執(zhí)行；
2 通過內(nèi)部處理onSubscribe方法會被執(zhí)行谒臼，這里需要背壓處理朝刊，如果沒有調(diào)用s.request()或者s.request(n)的參數(shù)小于等于0，流程結(jié)束蜈缤；
3 如果s.request(n)的參數(shù)大于零拾氓，會執(zhí)行onNext；
4 重復步驟2,3底哥，每執(zhí)行一次onNext咙鞍，參數(shù)n就會減1。

fromArray操作符需要注意的地方

1 數(shù)組不能是list類型趾徽，list會被當做一個事件發(fā)送出去续滋。
2 背壓處理類似create操作符，具體參考上文孵奶。
3 fromArray疲酌，F(xiàn)lowableSubscriber支持泛型，但是數(shù)組的元素類型和觀察者的類型要一致了袁。

轉(zhuǎn)換操作符

該類型的操作符都作用于一個可觀測序列畦韭，然后通過function函數(shù)把它變換其他的值谋币，最后用一種新的形式返回它們敬扛。

map操作符

map操作符的基本用法

Flowable.fromArray(new String[]{"1","2","3"})
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        return s+ "map";
                    }
                })
                .subscribe(new FlowableSubscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(3);
                    }
                    @Override
                    public void onNext(String s) {
                        System.out.print(s);
                    }
                    @Override
                    public void onError(Throwable throwable) {

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

以上輸出的結(jié)果為：1map 2map 3map

這里map操作符的參數(shù)Function有兩個泛型辙培，第一個泛型是接收到的事件的類型，第二個泛型是發(fā)送事件的類型崭庸，因此臀脏，第一個泛型需要和fromArray的泛型是同一個類型，而觀察者的泛型和第二個泛型是同一個類型冀自。

map操作符的執(zhí)行流程

1 fromArray操作符首先創(chuàng)建一個Flowable1
2 Flowable1調(diào)用操作符map，產(chǎn)生一個新的Flowable2秒啦；
3 Flowable2 訂閱觀察者
4 Flowable1 調(diào)用觀察者的onSubscribe()方法熬粗，需要處理背壓；
5 觀察者在onSubscribe()方法中通知上游觀察者的能力是3余境；
6 Flowable1 開始發(fā)射事件"1"驻呐；
7 Flowable2 的Function對事件“1”執(zhí)行apply()變換灌诅，轉(zhuǎn)化成"1map"；
8 觀察者接收到事件"1map";
9 重復步驟7,8含末，直到事件發(fā)送完畢或者被取消或者背壓為0猜拾。

flatMap操作符

flatMap操作符的功能

flatMap將一個發(fā)射數(shù)據(jù)的Observable變換為多個Observables，然后將它們發(fā)射的數(shù)據(jù)合并后放進一個單獨的Observable佣盒。
這個方法是很有用的挎袜，例如，當你有一個這樣的Observable：它發(fā)射一個數(shù)據(jù)序列肥惭，這些數(shù)據(jù)本身包含Observable成員或者可以變換為Observable盯仪，因此你可以創(chuàng)建一個新的Observable發(fā)射這些次級Observable發(fā)射的數(shù)據(jù)的完整集合。

flatMap操作符的基本使用

Student student1 = new Student();
        Student student2 = new Student();
        Student student3 = new Student();
        Student[] array={student1,student2,student3};
        for(int i=0;i<2;i++){
            array[i].cursors = new Cursor[2];
            Cursor cursor1 = new Cursor();
            cursor1.name="n1";
            cursor1.score=10*i;
            Cursor cursor2 = new Cursor();
            cursor2.name="n2";
            cursor2.score=10*i+1;
            array[i].cursors[0]=cursor1;
            array[i].cursors[1] = cursor2;
        }
        Flowable.fromArray(array)
                .flatMap(new Function<Student, Flowable<Cursor>>() {
                    @Override
                    public Flowable<Cursor> apply(Student student) throws Exception {
                        return  Flowable.fromArray(student.cursors);
                    }
                }).subscribe(new FlowableSubscriber<Cursor>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(3);
            }
            @Override
            public void onNext(Cursor cursor) {
                System.out.print(cursor.name+","+cursor.score);
            }
            @Override
            public void onError(Throwable throwable) {
            }
            @Override
            public void onComplete() {
            }
        });

以上面的例子為例來理解flatMap操作符：需求是打印所有學生的各科成績蜜葱。
所有的學生為一個數(shù)組全景，通過fromArray操作符依次發(fā)送事件，每個學生作為一個事件發(fā)送出去牵囤，flatMap操作符把每個學生變換為對應的Flowable爸黄，這個Flowable會把學生的成績依次發(fā)送出去，最后所有的成績匯總揭鳞，下發(fā)給觀察者炕贵。
可以看出，不需要循環(huán)操作汹桦，通過flatMap操作符就完成了所有學生的各科成績的打印鲁驶。

flatMap操作符需要注意的地方

經(jīng)過flatMap操作變換后，最后輸出的序列有可能是交錯的舞骆，因為flatMap最后合并結(jié)果采用的是merge操作符钥弯，后面會詳細介紹merge操作符。如果要想經(jīng)過變換后督禽，最終輸出的序列和原序列一致脆霎，那就會用到另外一個操作符，concatMap狈惫。

過濾操作符

顧名思義睛蛛，這類操作符按照過濾條件對可觀測的事件序列篩選，返回滿足我們條件的事件胧谈。過濾類操作符主要包含： Filter, Take, TakeLast, TakeUntilSkip, SkipLast, ElementAt, Debounce, Distinct, DistinctUntilChanged, First, Last等等忆肾。

fliter操作符

fliter操作符的基本用法

Flowable.fromArray(array)
                .filter(new Predicate<Student>() {
                    @Override
                    public boolean test(Student student) throws Exception {
                        return student.cursors[1].score<20;
                    }
                })
                .subscribe(new FlowableSubscriber<Student>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(3);
                    }
                    @Override
                    public void onNext(Student student) {
                        System.out.print(student.name);
                    }
                    @Override
                    public void onError(Throwable throwable) {
                    }
                    @Override
                    public void onComplete() {
                    }
                });

以上面flatMap的場景，如果需求變?yōu)椋阂蟠蛴〉诙T課程的成績<20的學生名單菱肖。同樣的fromArray會依次下發(fā)所有的學生客冈，fliter操作符判斷成績是否小于20，將小于20的學生下發(fā)給觀察者稳强。

filter操作符的流程

1 fromArray創(chuàng)建Flowable1场仲；
2 filter創(chuàng)建操作符Flowable2和悦；
3 Flowable2 訂閱觀察者；
4 執(zhí)行觀察者的onSubscribe()方法渠缕，需要處理背壓鸽素；
5 Flowable1 依次發(fā)送student事件；
6 Flowable2調(diào)用test()判斷student事件是否符合篩選條件亦鳞，符合返回true馍忽，否則返回false；
7 Flowable2 根據(jù)test()的結(jié)果下發(fā)事件蚜迅，返回值=true 事件下發(fā)給觀察者舵匾，返回值=false事件丟棄；
8 重復步驟5,6,7 直到事件發(fā)送完畢或者被取消或者背壓為0谁不。

distinct操作符

distinct操作符過濾規(guī)則是只允許還沒有發(fā)射過的數(shù)據(jù)通過坐梯，所有重復的數(shù)據(jù)項都只會發(fā)射一次。在數(shù)據(jù)去重場景中非常有用刹帕。

distinct操作符的基本用法

Flowable.fromArray(array)
                .distinct()
                .subscribe(new FlowableSubscriber<Student>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(3);
                    }
                    @Override
                    public void onNext(Student student) {
                        System.out.print(student.name);
                    }
                    @Override
                    public void onError(Throwable throwable) {
                    }
                    @Override
                    public void onComplete() {
                    }
                });

distinct操作符的執(zhí)行流程

1 fromArray創(chuàng)建Flowable1吵血；
2 distinct創(chuàng)建操作符Flowable2；
3 Flowable2 訂閱觀察者偷溺；
4 執(zhí)行觀察者的onSubscribe()方法蹋辅，此處應該調(diào)用request()方法，否則后面的流程中事件無法下發(fā)給觀察者
5 Flowable1 依次發(fā)送student事件給Flowable2挫掏；
6 Flowable2判斷事件是否已經(jīng)下發(fā)過侦另，對事件去重；
7 如果已經(jīng)下發(fā)過該事件則調(diào)用request(1)尉共，糾正背壓褒傅；若果該事件沒有下發(fā)過且被壓不為零則下發(fā)給觀察者；
8 重復步驟5,6,7 直到事件結(jié)束或者取消或者背壓為零袄友。

組合操作符

merge操作符

Merge其實只是將多個Flowable的輸出序列變?yōu)橐粋€殿托，方便訂閱者統(tǒng)一處理，對于訂閱者來說就仿佛只訂閱了一個觀察者一樣

merge操作符的基本使用方法

Flowable f1 = Flowable.just(1, 2, 3);
Flowable f2 = Flowable.just(7, 8, 9, 10);
Flowable.merge(f1,f2)
                .subscribe(new FlowableSubscriber() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(7);
                    }
                    @Override
                    public void onNext(Object o) {
                        System.out.print("merge:"+o.toString());
                    }
                    @Override
                    public void onError(Throwable throwable) {
                    }
                    @Override
                    public void onComplete() {
                    }
                });

merge操作符的執(zhí)行流程

1 創(chuàng)建Flowable1剧蚣；
2 創(chuàng)建Flowable2
3 merge創(chuàng)建操作符Flowable3支竹；
4 Flowable3 訂閱觀察者；
5 Flowable3 內(nèi)部創(chuàng)建1個隊列鸠按，用來Flowable1和Flowable2的事件
6 執(zhí)行觀察者的onSubscribe()方法礼搁，此處應該調(diào)用request()方法，否則后面的流程中事件無法下發(fā)給觀察者
7 Flowable1 發(fā)送事件目尖，F(xiàn)lowable3將其放進隊列中叹坦；
8 Flowable3 判斷隊列是否有事件，背壓是否不為零
9 如果8的條件都滿足卑雁，將隊列下發(fā)給觀察者募书；
10 重復步驟7,8,9 直到事件結(jié)束或者取消或者背壓為零。

zip操作符

zip通過一個函數(shù)將多個Flowable發(fā)送的事件結(jié)合到一起测蹲，然后發(fā)送這些組合到一起的事件. 它按照嚴格的順序應用這個函數(shù)莹捡。它只發(fā)射與發(fā)射數(shù)據(jù)項最少的那個Flowable一樣多的數(shù)據(jù)

zip操作符的基本使用方法

 Flowable f1 = Flowable.just(1, 2, 3);
 Flowable f2 = Flowable.just(7, 8, 9, 10);
Flowable.zip(f1, f2, new BiFunction() {
                             @Override
                            public Object apply(Object o, Object o2) throws Exception {
                            return "f1: "+o.toString()+" and f2: "+o2.toString();
                     }})
                .subscribe(new FlowableSubscriber() {
                    @Override
                    public void onSubscribe(Subscription s) {
                      s.request(4);  
                    }
                    @Override
                    public void onNext(Object o) {
                        System.out.print("zip:"+o.toString());
                    }
                    @Override
                    public void onError(Throwable throwable) {
                    }
                    @Override
                    public void onComplete() {
                    }
                });

zip操作符的執(zhí)行流程

1 創(chuàng)建Flowable1；
2 創(chuàng)建Flowable2
3 zip創(chuàng)建操作符Flowable3扣甲；
4 Flowable3 訂閱觀察者篮赢；
5 Flowable3 內(nèi)部創(chuàng)建兩個隊列，分別用來暫存Flowable1和Flowable2的事件
6 執(zhí)行觀察者的onSubscribe()方法琉挖，此處應該調(diào)用request()方法启泣，否則后面的流程中事件無法下發(fā)給觀察者
7 Flowable1 發(fā)送事件，F(xiàn)lowable3將其放進隊列1中示辈；
8 Flowable3 判斷隊列1和隊列2是否有事件寥茫，背壓是否不為零
9 如果8的條件都滿足，將隊列1和隊列2的事件執(zhí)行BiFunction的apply()方法矾麻；
10 如果8的條件不滿足 Flowable2 發(fā)送事件纱耻，F(xiàn)lowa3將其放進隊列2中；
11 Flowable3 判斷隊列1和隊列2是否有事件险耀，背壓是否不為零弄喘；
12 如果11的條件都滿足，將隊列1和隊列2的事件執(zhí)行BiFunction的apply()方法甩牺；
13 重復步驟7,8,9,10,11,12 直到事件結(jié)束或者取消或者背壓為零蘑志。

zip操作符需要注意的地方

1 如果多個Flowable的事件長度不一樣，觀察者接收到的是事件長度和最短的Flowable事件長度一樣贬派。
2 如果多個Flowable的事件長度不一樣急但，最短的是d，每個Flowable只取前面d個事件赠群。

切換線程操作符

subscribeOn

subscribeOn操作符的基本用法

 Flowable.fromArray(array)
                .subscribeOn(Schedulers.newThread())
                .subscribe(new FlowableSubscriber<Student>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(3);
                        System.out.print(Thread.currentThread().getName());
                    }
                    @Override
                    public void onNext(Student student) {
                        System.out.print(student.name);
                        System.out.print(Thread.currentThread().getName());
                    }
                    @Override
                    public void onError(Throwable t) {
                    }
                    @Override
                    public void onComplete() {
                    }
                });

subscribeOn操作符的流程

1 fromArray操作符創(chuàng)建Flowable1;
2 subscribeOn操作符創(chuàng)建Flowable2羊始；
3 Flowable2訂閱觀察者；
4 Flowable2 內(nèi)部調(diào)用觀察者的onSubscribe()方法查描；
5 Flowable2 切換到線程Thread2突委；
6 Flowable1 在線程Thread2中生產(chǎn)并下發(fā)事件；
7 觀察者在線程Thread2中接收事件冬三；
8 重復步驟6,7匀油，直到事件下發(fā)完畢或者被取消。

subscribeOn操作符需要注意的地方

1 在一個場景中可以多次使用subscribeOn操作符勾笆；
2 多次使用時敌蚜，F(xiàn)lowable的事件都是在第一個subscribeOn操作符的線程種執(zhí)行；
3 后面的 subscribeOn 只會改變前面的 subscribeOn 調(diào)度操作所在的線程窝爪，并不能改變最終被調(diào)度的事件執(zhí)行的線程弛车，但對于中途的代碼執(zhí)行的線程齐媒，還是會影響到的。

observerOn

observerOn操作符的基本用法

Flowable.fromArray(array)
                .observeOn(Schedulers.newThread())
                .subscribe(new FlowableSubscriber<Student>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        s.request(3);
                        System.out.print(Thread.currentThread().getName());
                    }
                    @Override
                    public void onNext(Student student) {
                        System.out.print(student.name);
                        System.out.print(Thread.currentThread().getName());
                    }
                    @Override
                    public void onError(Throwable t) {
                    }
                    @Override
                    public void onComplete() {
                    }
                });

observerOn操作符的流程

1 fromArray操作符創(chuàng)建Flowable1;
2 observerOn操作符創(chuàng)建Flowable2纷跛；
3 Flowable2訂閱觀察者喻括；
4 Flowable2 內(nèi)部調(diào)用觀察者的onSubscribe()方法；
5 Flowable1 生產(chǎn)并下發(fā)事件贫奠；
6 Flowable2 接收Flowable1的事件并切換到線程Thread2唬血；
7 觀察者在線程Thread2中接收事件；
8 重復步驟5唤崭，6拷恨，7直到事件下發(fā)完畢或者被取消。

observerOn操作符需要注意的地方

1 在一個場景中可以多次使用observerOn操作符谢肾；
2 每級的Observer的onXXX方法都在不同的線程中被調(diào)用腕侄。所以observeOn的調(diào)用會多次生效；
3 observeOn 影響它下面的調(diào)用執(zhí)行時所在的線程勒叠，每次調(diào)用都會改變數(shù)據(jù)向下傳遞時所在的線程兜挨。

參考

給初學者的RxJava2.0教程
探索專為 Android 而設計的 RxJava 2
RxJava系列6(從微觀角度解讀RxJava源碼)
淺談 RxJava 與 2.0 的新特性
RxJava 2.x 使用詳解