RxJava的簡單使用

導(dǎo)入RxJava的相關(guān)依賴
創(chuàng)建觀察者與被觀察者
鏈?zhǔn)秸{(diào)用
next几迄、error蔚龙、complete
subscribe的重載方法
指定觀察者與被觀察者所在的線程以及線程切換
結(jié)合Retrofit，實(shí)現(xiàn)一個(gè)簡單的網(wǎng)絡(luò)請求

一映胁、引入依賴：

implementation "io.reactivex.rxjava2:rxjava:2.2.3"
implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'

二木羹、觀察者與被觀察者
RxJava中的Observable和Observer通過SubScribe建立連接：

 //創(chuàng)建一個(gè)上游 Observable：
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onError(new Throwable());
                emitter.onComplete();
            }
        });
        //創(chuàng)建一個(gè)下游 Observer
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "subscribe");
            }

            @Override
            public void onNext(Integer value) {
                Log.d(TAG, "" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "error");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "complete");
            }
        };
        //建立連接
        observable.subscribe(observer);

運(yùn)行結(jié)果：

11-21 16:49:58.443 21398-21398/com.example.qidong.testrxjava D/MainActivity: subscribe
11-21 16:49:58.443 21398-21398/com.example.qidong.testrxjava D/MainActivity: 1
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: 2
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: 3
11-21 16:49:58.444 21398-21398/com.example.qidong.testrxjava D/MainActivity: error

注意: 只有當(dāng)上游和下游建立連接之后, 上游才會開始發(fā)送事件. 也就是調(diào)用了subscribe() 方法之后才開始發(fā)送事件.

三、RxJava的鏈?zhǔn)秸{(diào)用
使用RxJava的鏈?zhǔn)讲僮魅缦拢?/p> 

Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                emitter.onNext(20);
                emitter.onNext(30);
                emitter.onError(new Throwable());
                emitter.onNext(40);
                emitter.onComplete();

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

                Log.e(TAG,"OnSubscribe");

            }

            @Override
            public void onNext(Integer integer) {
                Log.e(TAG,"onNext:"+integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG,"onError");

            }

            @Override
            public void onComplete() {
                Log.e(TAG,"OnComplete");

            }
        });

四解孙、next坑填、error抛人、complete事件
從上面的代碼中可以看到ObservableEmitter和Disposable。
ObservableEmitter：意思為發(fā)射器脐瑰，可以通過調(diào)用onNext()函匕、onError(Throwable)、onComplete()來發(fā)射next蚪黑、error盅惜、complete這三種事件。
這其中需要注意的地方有：
上游可以發(fā)送無限個(gè)onNext, 下游也可以接收無限個(gè)onNext.
當(dāng)上游發(fā)送了一個(gè)onComplete后, 上游onComplete之后的事件將會繼續(xù)發(fā)送, 而下游收到onComplete事件之后將不再繼續(xù)接收事件.
當(dāng)上游發(fā)送了一個(gè)onError后, 上游onError之后的事件將繼續(xù)發(fā)送, 而下游收到onError事件之后將不再繼續(xù)接收事件.
上游可以不發(fā)送onComplete或onError.
最為關(guān)鍵的是onComplete和onError必須唯一并且互斥, 即不能發(fā)多個(gè)onComplete, 也不能發(fā)多個(gè)onError, 也不能先發(fā)一個(gè)onComplete, 然后再發(fā)一個(gè)onError, 反之亦然

Disposable, 這個(gè)單詞的字面意思是一次性用品,用完即可丟棄的. 那么在RxJava中怎么去理解它呢, 對應(yīng)于上面的水管的例子, 我們可以把它理解成兩根管道之間的一個(gè)機(jī)關(guān), 當(dāng)調(diào)用它的dispose()方法時(shí), 它就會將兩根管道切斷, 從而導(dǎo)致下游收不到事件.

注意: 調(diào)用dispose()并不會導(dǎo)致上游不再繼續(xù)發(fā)送事件, 上游會繼續(xù)發(fā)送剩余的事件.

來看個(gè)例子, 我們讓上游依次發(fā)送1,2,3,complete,4,在下游收到第二個(gè)事件之后, 切斷水管, 看看運(yùn)行結(jié)果:

Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "emit 1");
                emitter.onNext(1);
                Log.e(TAG, "emit 2");
                emitter.onNext(2);
                Log.e(TAG, "emit 3");
                emitter.onNext(3);
                Log.e(TAG, "onComplete");
                emitter.onComplete();
                Log.e(TAG, "emit 4");
                emitter.onNext(4);

            }
        }).subscribe(new Observer<Integer>() {

            private Disposable mDispoable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "onSubscribe: ");
                this.mDispoable=d;


            }

            @Override
            public void onNext(Integer integer) {

                Log.e(TAG, "onNext: "+integer);


                i++;

                if(i==2)
                {
                    Log.e(TAG, "dispose");
                    mDispoable.dispose();
                    Log.e(TAG, "isDisposed:"+mDispoable.isDisposed());
                }

            }

            @Override
            public void onError(Throwable e) {

                Log.e(TAG, "onError");

            }

            @Override
            public void onComplete() {

                Log.e(TAG, "onComplete");

            }
        });

運(yùn)行結(jié)果：

11-21 17:11:50.338 23074-23074/com.example.qidong.testrxjava E/MainActivity: onSubscribe: 
11-21 17:11:50.338 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 1
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onNext: 1
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 2
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onNext: 2
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: dispose
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: isDisposed:true
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 3
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: onComplete
11-21 17:11:50.339 23074-23074/com.example.qidong.testrxjava E/MainActivity: emit 4

可以看到忌穿，切斷觀察者和被觀察者之間的關(guān)聯(lián)之后，事件3掠剑、complete、4事件依然被發(fā)送了井佑。

五躬翁、subscribe的重載方法
SubScribe有多個(gè)重載的方法：

   public final Disposable subscribe() {}
    public final Disposable subscribe(Consumer<? super T> onNext) {}
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} 
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
    public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
    public final void subscribe(Observer<? super T> observer) {}

最后一個(gè)帶有Observer參數(shù)的我們已經(jīng)使用過了,這里對其他幾個(gè)方法進(jìn)行說明.
不帶任何參數(shù)的subscribe() 表示下游不關(guān)心任何事件,你上游盡管發(fā)你的數(shù)據(jù)去吧, 老子可不管你發(fā)什么.
帶有一個(gè)Consumer參數(shù)的方法表示下游只關(guān)心onNext事件, 其他的事件我假裝沒看見, 因此我們?nèi)绻恍枰猳nNext事件可以這么寫:

Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "onNext:20");
                emitter.onNext(20);
                Log.e(TAG, "OnComplete");
                emitter.onComplete();
                Log.e(TAG, "onNext:30");
                emitter.onNext(30);

            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);

            }
        });
    }

運(yùn)行結(jié)果：

11-21 17:29:00.671 23664-23664/? E/MainActivity: onNext:10
11-21 17:29:00.672 23664-23664/? E/MainActivity: accept:10
11-21 17:29:00.672 23664-23664/? E/MainActivity: onNext:20
11-21 17:29:00.672 23664-23664/? E/MainActivity: accept:20
11-21 17:29:00.672 23664-23664/? E/MainActivity: OnComplete
11-21 17:29:00.672 23664-23664/? E/MainActivity: onNext:30

六、指定觀察者與被觀察者所在的線程以及線程切換
正常情況下狡逢，觀察者和被觀察者是工作在一個(gè)線程中的，也就是被觀察者在哪個(gè)線程發(fā)射事件蛮艰，觀察者就在哪個(gè)線程接受事件：

Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());

            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);
                Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());

            }
        });

運(yùn)行結(jié)果：

11-21 17:43:25.366 25090-25090/? E/MainActivity: onNext:10
11-21 17:43:25.366 25090-25090/? E/MainActivity: accept:10
11-21 17:43:25.366 25090-25090/? E/MainActivity: Observer Thread Name:main
11-21 17:43:25.366 25090-25090/? E/MainActivity: Observable Thread Name:main

一般而言壤蚜，我們會將耗時(shí)操作放在子線程中去做仍律，然后在主線程中根據(jù)運(yùn)行的結(jié)果去更新UI。
RxJava中提供了線程調(diào)度器實(shí)現(xiàn)這樣的需求：我們在子線程中發(fā)射事件善涨，在主線程中接受事件:

Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());

            }
        }).subscribeOn(Schedulers.newThread())
          .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);
                Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());

            }
        });

此時(shí)再看一下運(yùn)行結(jié)果：

11-21 17:50:37.408 25766-25934/com.example.qidong.testrxjava E/MainActivity: onNext:10
11-21 17:50:37.409 25766-25934/com.example.qidong.testrxjava E/MainActivity: Observable Thread Name:RxNewThreadScheduler-1
11-21 17:50:37.453 25766-25766/com.example.qidong.testrxjava E/MainActivity: accept:10
11-21 17:50:37.453 25766-25766/com.example.qidong.testrxjava E/MainActivity: Observer Thread Name:main

簡單的來說, subscribeOn() 指定的是上游發(fā)送事件的線程, observeOn() 指定的是下游接收事件的線程.
多次指定被觀察者的線程只有第一次指定的有效, 也就是說多次調(diào)用subscribeOn() 只有第一次的有效, 其余的會被忽略.
多次指定觀察者的線程是可以的, 也就是說每調(diào)用一次observeOn() , 下游的線程就會切換一次.

 Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {

                Log.e(TAG, "onNext:10");
                emitter.onNext(10);
                Log.e(TAG, "Observable Thread Name:"+Thread.currentThread().getName());

            }
        }).subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

                Log.e(TAG, "accept:"+integer);
                Log.e(TAG, "Observer Thread Name:"+Thread.currentThread().getName());

            }
        });

再來看下運(yùn)行結(jié)果：

11-21 17:54:36.940 26399-26522/com.example.qidong.testrxjava E/MainActivity: onNext:10
11-21 17:54:36.941 26399-26522/com.example.qidong.testrxjava E/MainActivity: Observable Thread Name:RxNewThreadScheduler-1
11-21 17:54:36.973 26399-26524/com.example.qidong.testrxjava E/MainActivity: accept:10
11-21 17:54:36.974 26399-26524/com.example.qidong.testrxjava E/MainActivity: Observer Thread Name:RxCachedThreadScheduler-2

可以看出觀察者所在的線程被修改成功炕横。

為了看出觀察者所在線程的切換過程葡粒，我們可以添加Log去查看：

Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.e(TAG, "emit 10");
                emitter.onNext(10);
                Log.e(TAG, "emit 20");
                emitter.onNext(20);

            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.e(TAG, "After Change to MainThread,thread name:"+Thread.currentThread().getName());

                    }
                }).observeOn(Schedulers.io()).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "After Change to IO,thread name:"+Thread.currentThread().getName());
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.e(TAG, "accept: "+integer);
            }
        });

運(yùn)行結(jié)果：

11-21 18:08:37.913 28956-29018/com.example.qidong.testrxjava E/MainActivity: emit 10
11-21 18:08:37.914 28956-29018/com.example.qidong.testrxjava E/MainActivity: emit 20
11-21 18:08:37.945 28956-28956/com.example.qidong.testrxjava E/MainActivity: After Change to MainThread,thread name:main
11-21 18:08:37.946 28956-28956/com.example.qidong.testrxjava E/MainActivity: After Change to MainThread,thread name:main
11-21 18:08:37.946 28956-29020/com.example.qidong.testrxjava E/MainActivity: After Change to IO,thread name:RxCachedThreadScheduler-1
11-21 18:08:37.946 28956-29020/com.example.qidong.testrxjava E/MainActivity: accept: 10
11-21 18:08:37.947 28956-29020/com.example.qidong.testrxjava E/MainActivity: After Change to IO,thread name:RxCachedThreadScheduler-1
11-21 18:08:37.947 28956-29020/com.example.qidong.testrxjava E/MainActivity: accept: 20

在RxJava中, 已經(jīng)內(nèi)置了很多線程選項(xiàng)供我們選擇, 例如有
Schedulers.io() 代表io操作的線程, 通常用于網(wǎng)絡(luò),讀寫文件等io密集型的操作
Schedulers.computation() 代表CPU計(jì)算密集型的操作, 例如需要大量計(jì)算的操作
Schedulers.newThread() 代表一個(gè)常規(guī)的新線程
AndroidSchedulers.mainThread() 代表Android的主線程
這些內(nèi)置的Scheduler已經(jīng)足夠滿足我們開發(fā)的需求, 因此我們應(yīng)該使用內(nèi)置的這些選項(xiàng), 在RxJava內(nèi)部使用的是線程池來維護(hù)這些線程, 所有效率也比較高.

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 人面猴
    序言：七十年代末，一起剝皮案震驚了整個(gè)濱河市拾枣，隨后出現(xiàn)的幾起案子盒让，更是在濱河造成了極大的恐慌，老刑警劉巖姨蝴，帶你破解...
    沈念sama閱讀 218,755評論 6贊 507
  • 死咒
    序言：濱河連續(xù)發(fā)生了三起死亡事件似扔，死亡現(xiàn)場離奇詭異搓谆，居然都是意外死亡，警方通過查閱死者的電腦和手機(jī)黔寇，發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 93,305評論 3贊 395
  • 救了他兩次的神仙讓他今天三更去死
    文/潘曉璐 我一進(jìn)店門缝裤，熙熙樓的掌柜王于貴愁眉苦臉地迎上來憋飞，“玉大人姆吭，你說我怎么就攤上這事〖烀校” “怎么了？”我有些...
    開封第一講書人閱讀 165,138評論 0贊 355
  • 道士緝兇錄：失蹤的賣姜人
    文/不壞的土叔 我叫張陵刽严，是天一觀的道長舞萄。 經(jīng)常有香客問我管削，道長，這世上最難降的妖魔是什么把还？ 我笑而不...
    開封第一講書人閱讀 58,791評論 1贊 295
  • ?港島之戀（遺憾婚禮）
    正文 為了忘掉前任吊履，我火速辦了婚禮调鬓，結(jié)果婚禮上，老公的妹妹穿的比我還像新娘缀踪。我一直安慰自己虹脯，他們只是感情好，可當(dāng)我...
    茶點(diǎn)故事閱讀 67,794評論 6贊 392
  • 惡毒庶女頂嫁案：這布局不是一般人想出來的
    文/花漫 我一把揭開白布唇敞。 她就那樣靜靜地躺著疆柔，像睡著了一般镶柱。 火紅的嫁衣襯著肌膚如雪。 梳的紋絲不亂的頭發(fā)上鞋屈，一...
    開封第一講書人閱讀 51,631評論 1贊 305
  • 城市分裂傳說
    那天，我揣著相機(jī)與錄音湖蜕，去河邊找鬼宋列。 笑死评也，一個(gè)胖子當(dāng)著我的面吹牛，可吹牛的內(nèi)容都是我干的坤邪。 我是一名探鬼主播罚缕，決...
    沈念sama閱讀 40,362評論 3贊 418
  • 雙鴛鴦連環(huán)套：你想象不到人心有多黑
    文/蒼蘭香墨 我猛地睜開眼邮弹，長吁一口氣：“原來是場噩夢啊……” “哼！你這毒婦竟也來了腌乡？” 一聲冷哼從身側(cè)響起与纽，我...
    開封第一講書人閱讀 39,264評論 0贊 276
  • 萬榮殺人案實(shí)錄
    序言：老撾萬榮一對情侶失蹤，失蹤者是張志新（化名）和其女友劉穎影所，沒想到半個(gè)月后僚碎，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體，經(jīng)...
    沈念sama閱讀 45,724評論 1贊 315
  • ?護(hù)林員之死
    正文 獨(dú)居荒郊野嶺守林人離奇死亡胀溺，尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 37,900評論 3贊 336
  • ?白月光啟示錄
    正文 我和宋清朗相戀三年仓坞，在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了无埃。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片。...
    茶點(diǎn)故事閱讀 40,040評論 1贊 350
  • 活死人
    序言：一個(gè)原本活蹦亂跳的男人離奇死亡侦镇，死狀恐怖织阅，靈堂內(nèi)的尸體忽然破棺而出，到底是詐尸還是另有隱情闹炉，我是刑警寧澤，帶...
    沈念sama閱讀 35,742評論 5贊 346
  • ?日本核電站爆炸內(nèi)幕
    正文 年R本政府宣布渣触，位于F島的核電站嗅钻，受9級特大地震影響店展，放射性物質(zhì)發(fā)生泄漏。R本人自食惡果不足惜觉至，卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 41,364評論 3贊 330
  • 男人毒藥：我在死后第九天來索命
    文/蒙蒙 一语御、第九天 我趴在偏房一處隱蔽的房頂上張望。 院中可真熱鬧席怪，春花似錦、人聲如沸碉纺。這莊子的主人今日做“春日...
    開封第一講書人閱讀 31,944評論 0贊 22
  • 一樁弒父案声怔，背后竟有這般陰謀
    文/蒼蘭香墨 我抬頭看了看天上的太陽。三九已至醋火，卻和暖如春箱吕，著一層夾襖步出監(jiān)牢的瞬間茬高，已是汗流浹背假抄。 一陣腳步聲響...
    開封第一講書人閱讀 33,060評論 1贊 270
  • 情欲美人皮
    我被黑心中介騙來泰國打工， 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留婚瓜，地道東北人。 一個(gè)月前我還...
    沈念sama閱讀 48,247評論 3贊 371
  • 代替公主和親
    正文 我出身青樓幔摸，卻偏偏與公主長得像，于是被迫代替她去往敵國和親碍舍。 傳聞我的和親對象是個(gè)殘疾皇子邑雅，可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 44,979評論 2贊 355