來(lái)自于：CSDN?blog.csdn.net/caihongdao123/article/details/51897793
前言：
第一次接觸RxJava是在前不久，一個(gè)新Android項(xiàng)目的啟動(dòng)乙嘀，在評(píng)估時(shí)選擇了RxJava末购。RxJava是一個(gè)基于事件訂閱的異步執(zhí)行的一個(gè)類庫(kù)。聽(tīng)起來(lái)有點(diǎn)復(fù)雜虎谢，其實(shí)是要你使用過(guò)一次盟榴，就會(huì)大概明白它是怎么回事了！為是什么一個(gè)android項(xiàng)目啟動(dòng)會(huì)聯(lián)系到RxJava呢婴噩？因?yàn)樵赗xJava使用起來(lái)得到廣泛的認(rèn)可擎场，又是基于Java語(yǔ)言的。自然會(huì)有善于組織和總結(jié)的開(kāi)發(fā)者聯(lián)想到Android几莽！沒(méi)錯(cuò)迅办，RxAndroid就這樣在RxJava的基礎(chǔ)上，針對(duì)Android開(kāi)發(fā)的一個(gè)庫(kù)章蚣。今天我們主要是來(lái)講解一下RxJava站欺，在接下來(lái)的幾篇博客中我會(huì)陸續(xù)帶大家來(lái)認(rèn)識(shí)RxAndroid，Retrofit框架的使用纤垂，這些都是目前比較火的一些技術(shù)框架矾策！
這里是Github上RxJava的項(xiàng)目地址：https://github.com/ReactiveX/RxJava
技術(shù)文檔Api：http://reactivex.io/RxJava/javadoc/
官方的介紹
1.支持Java6+
2.android 2.3+
3.異步的
4.基于觀察者設(shè)計(jì)模式（Observer、Observable）不懂設(shè)計(jì)模式的可以移步到此：淺談Java設(shè)計(jì)模式（十五）觀察者模式（Observer）
5.Subscribe (訂閱)
正式使用RxJava
用框架或者庫(kù)都是為了簡(jiǎn)潔峭沦、方便贾虽，RxJava也不例外它能使你的代碼邏輯更加的簡(jiǎn)潔。舉個(gè)例子之前我們先來(lái)引入依賴的 gradle 代碼：
[java]view plaincopy
compile'io.reactivex:rxjava:1.0.14'
compile'io.reactivex:rxandroid:1.0.1'
既然是基于異步吼鱼，當(dāng)然要在處理比較耗時(shí)的操作上才能彰顯它的優(yōu)勢(shì)蓬豁！現(xiàn)在我們假設(shè)有這樣一個(gè)需求：
需要實(shí)現(xiàn)一個(gè)多個(gè)下載的圖片并且顯示的功能，它的作用可以添加多個(gè)下載操作蛉抓，由于下載這一過(guò)程較為耗時(shí)庆尘，需要放在后臺(tái)執(zhí)行，而圖片的顯示則必須在 UI 線程執(zhí)行巷送。常用的實(shí)現(xiàn)方式有多種，我這里貼出其中一種：
[java]view plaincopy
newThread()?{
@Override
publicvoidrun()?{
super.run();
for(File?folder?:?folders)?{
File[]?files?=?folder.listFiles();
for(File?file?:?files)?{
if(file.getName().endsWith(".png"))?{
finalBitmap?bitmap?=?getBitmapFromFile(file);
getActivity().runOnUiThread(newRunnable()?{
@Override
publicvoidrun()?{
imageCollectorView.addImage(bitmap);
}
});
}
}
}
}
}.start();
里面的判斷是不是看起來(lái)有點(diǎn)暈暈矛辕，當(dāng)然這是我自己寫(xiě)的笑跛，我一眼就能看清楚里面的邏輯，但是如果換做是別人來(lái)閱讀你的代碼聊品，這就比較的尷尬了飞蹂！
我們來(lái)看看使用RxJava的代碼：
[java]view plaincopy
Observable.from(folders)
.flatMap(newFunc1>()?{
@Override
publicObservable?call(File?file)?{
returnObservable.from(file.listFiles());
}
})
.filter(newFunc1()?{
@Override
publicBoolean?call(File?file)?{
returnfile.getName().endsWith(".png");
}
})
.map(newFunc1()?{
@Override
publicBitmap?call(File?file)?{
returngetBitmapFromFile(file);
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(newAction1()?{
@Override
publicvoidcall(Bitmap?bitmap)?{
imageCollectorView.addImage(bitmap);
}
});
是不是明了，雖然說(shuō)算不上簡(jiǎn)單翻屈，但是習(xí)慣了就一如既往了陈哑！
如果你使用的AndroidStudio的話，你打開(kāi)Java文件的時(shí)候，你會(huì)看到被自動(dòng) Lambda 化的預(yù)覽惊窖，這將讓你更加清晰地看到程序邏輯：
[java]view plaincopy
Observable.from(folders)
.flatMap((Func1)?(folder)?->?{?Observable.from(file.listFiles())?})
.filter((Func1)?(file)?->?{?file.getName().endsWith(".png")?})
.map((Func1)?(file)?->?{?getBitmapFromFile(file)?})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe((Action1)?(bitmap)?->?{?imageCollectorView.addImage(bitmap)?});
不過(guò)如果你對(duì)Java8還不是很了解的話呢這一段可以暫時(shí)忽略刽宪，但是你可以移步到這里了解一下Java8：Java8部分新特性介紹
看完代碼，是不是有種相見(jiàn)恨晚的沖動(dòng)界酒？別急圣拄，我們來(lái)慢慢了解RxJava！
前面已經(jīng)提到他是基于Java觀察者設(shè)計(jì)模式的毁欣，這個(gè)模式上面有給大家鏈接庇谆，可以去看看，這里不不坐過(guò)多的介紹凭疮，我們來(lái)介紹一下RxJava中的觀察者模式：
RxJava 的觀察者模式
一饭耳、說(shuō)明
1）RxJava 有四個(gè)基本概念：Observable (可觀察者，即被觀察者)执解、 Observer (觀察者)寞肖、 subscribe (訂閱)、事件材鹦。Observable 和 Observer 通過(guò) subscribe() 方法實(shí)現(xiàn)訂閱關(guān)系逝淹，從而 Observable 可以在需要的時(shí)候發(fā)出事件來(lái)通知 Observer。
2）與傳統(tǒng)觀察者模式不同桶唐， RxJava 的事件回調(diào)方法除了普通事件 onNext() （相當(dāng)于 onClick() / onEvent()）之外栅葡，還定義了兩個(gè)特殊的事件：onCompleted() 和 onError()。
3）onCompleted(): 事件隊(duì)列完結(jié)尤泽。RxJava 不僅把每個(gè)事件單獨(dú)處理欣簇，還會(huì)把它們看做一個(gè)隊(duì)列。RxJava 規(guī)定坯约，當(dāng)不會(huì)再有新的 onNext() 發(fā)出時(shí)熊咽，需要觸發(fā) onCompleted() 方法作為標(biāo)志。
4）onError(): 事件隊(duì)列異常闹丐。在事件處理過(guò)程中出異常時(shí)横殴，onError() 會(huì)被觸發(fā)，同時(shí)隊(duì)列自動(dòng)終止卿拴，不允許再有事件發(fā)出衫仑。
5）在一個(gè)正確運(yùn)行的事件序列中, onCompleted() 和 onError() 有且只有一個(gè)，并且是事件序列中的最后一個(gè)堕花。需要注意的是文狱，onCompleted() 和 onError() 二者也是互斥的，即在隊(duì)列中調(diào)用了其中一個(gè)缘挽，就不應(yīng)該再調(diào)用另一個(gè)瞄崇。
二呻粹、實(shí)現(xiàn)
1) 創(chuàng)建 Observer
Observer 即觀察者，它決定事件觸發(fā)的時(shí)候?qū)⒂性鯓拥男袨椤?RxJava 中的 Observer 接口的實(shí)現(xiàn)方式：
[java]view plaincopy
Observer?observer?=newObserver()?{
@Override
publicvoidonNext(String?s)?{
Log.d(tag,"Item:?"+?s);
}
@Override
publicvoidonCompleted()?{
Log.d(tag,"Completed!");
}
@Override
publicvoidonError(Throwable?e)?{
Log.d(tag,"Error!");
}
};
除了 Observer 接口之外苏研，RxJava 還內(nèi)置了一個(gè)實(shí)現(xiàn)了 Observer 的抽象類：Subscriber等浊。 Subscriber 對(duì) Observer 接口進(jìn)行了一些擴(kuò)展，但他們的基本使用方式是完全一樣的：
[java]view plaincopy
Subscriber?subscriber?=newSubscriber()?{
@Override
publicvoidonNext(String?s)?{
Log.d(tag,"Item:?"+?s);
}
@Override
publicvoidonCompleted()?{
Log.d(tag,"Completed!");
}
@Override
publicvoidonError(Throwable?e)?{
Log.d(tag,"Error!");
}
};
不僅基本使用方式一樣楣富，實(shí)質(zhì)上凿掂，在 RxJava 的 subscribe 過(guò)程中，Observer 也總是會(huì)先被轉(zhuǎn)換成一個(gè) Subscriber 再使用纹蝴。所以如果你只想使用基本功能庄萎，選擇 Observer 和 Subscriber 是完全一樣的。它們的區(qū)別對(duì)于使用者來(lái)說(shuō)主要有兩點(diǎn)：
onStart():這是 Subscriber 增加的方法塘安。它會(huì)在 subscribe 剛開(kāi)始糠涛，而事件還未發(fā)送之前被調(diào)用，可以用于做一些準(zhǔn)備工作兼犯，例如數(shù)據(jù)的清零或重置忍捡。這是一個(gè)可選方法，默認(rèn)情況下它的實(shí)現(xiàn)為空切黔。需要注意的是砸脊，如果對(duì)準(zhǔn)備工作的線程有要求（例如彈出一個(gè)顯示進(jìn)度的對(duì)話框，這必須在主線程執(zhí)行）纬霞， onStart() 就不適用了凌埂，因?yàn)樗偸窃?subscribe 所發(fā)生的線程被調(diào)用，而不能指定線程诗芜。要在指定的線程來(lái)做準(zhǔn)備工作瞳抓，可以使用 doOnSubscribe() 方法，具體可以在后面的文中看到伏恐。
unsubscribe():這是 Subscriber 所實(shí)現(xiàn)的另一個(gè)接口 Subscription 的方法孩哑，用于取消訂閱。在這個(gè)方法被調(diào)用后翠桦，Subscriber 將不再接收事件横蜒。一般在這個(gè)方法調(diào)用前，可以使用 isUnsubscribed() 先判斷一下?tīng)顟B(tài)销凑。 unsubscribe() 這個(gè)方法很重要愁铺，因?yàn)樵?subscribe() 之后， Observable 會(huì)持有 Subscriber 的引用闻鉴，這個(gè)引用如果不能及時(shí)被釋放，將有內(nèi)存泄露的風(fēng)險(xiǎn)茂洒。所以最好保持一個(gè)原則：要在不再使用的時(shí)候盡快在合適的地方（例如 onPause() onStop() 等方法中）調(diào)用 unsubscribe() 來(lái)解除引用關(guān)系孟岛，以避免內(nèi)存泄露的發(fā)生。
2) 創(chuàng)建 Observable
Observable 即被觀察者，它決定什么時(shí)候觸發(fā)事件以及觸發(fā)怎樣的事件渠羞。 RxJava 使用 create() 方法來(lái)創(chuàng)建一個(gè) Observable 斤贰，并為它定義事件觸發(fā)規(guī)則：
[java]view plaincopy
Observable?observable?=?Observable.create(newObservable.OnSubscribe()?{
@Override
publicvoidcall(Subscriber?subscriber)?{
subscriber.onNext("Hello");
subscriber.onNext("Hi");
subscriber.onNext("Aloha");
subscriber.onCompleted();
}
});
可以看到，這里傳入了一個(gè) OnSubscribe 對(duì)象作為參數(shù)次询。OnSubscribe 會(huì)被存儲(chǔ)在返回的 Observable 對(duì)象中荧恍，它的作用相當(dāng)于一個(gè)計(jì)劃表，當(dāng) Observable 被訂閱的時(shí)候屯吊，OnSubscribe 的 call() 方法會(huì)自動(dòng)被調(diào)用送巡，事件序列就會(huì)依照設(shè)定依次觸發(fā)（對(duì)于上面的代碼，就是觀察者Subscriber 將會(huì)被調(diào)用三次 onNext() 和一次 onCompleted()）盒卸。這樣骗爆，由被觀察者調(diào)用了觀察者的回調(diào)方法，就實(shí)現(xiàn)了由被觀察者向觀察者的事件傳遞，即觀察者模式。
create() 方法是 RxJava 最基本的創(chuàng)造事件序列的方法乱凿》医荆基于這個(gè)方法， RxJava 還提供了一些方法用來(lái)快捷創(chuàng)建事件隊(duì)列抡驼，例如：
just(T...): 將傳入的參數(shù)依次發(fā)送出來(lái)。
[java]view plaincopy
Observable?observable?=?Observable.just("Hello","Hi","Aloha");
//?將會(huì)依次調(diào)用：
//?onNext("Hello");
//?onNext("Hi");
//?onNext("Aloha");
//?onCompleted();
from(T[]) / from(Iterable) : 將傳入的數(shù)組或 Iterable 拆分成具體對(duì)象后，依次發(fā)送出來(lái)外臂。
[java]view plaincopy
String[]?words?=?{"Hello","Hi","Aloha"};
Observable?observable?=?Observable.from(words);
//?將會(huì)依次調(diào)用：
//?onNext("Hello");
//?onNext("Hi");
//?onNext("Aloha");
//?onCompleted();
上面 just(T...) 的例子和 from(T[]) 的例子，都和之前的 create(OnSubscribe) 的例子是等價(jià)的体箕。
3) Subscribe (訂閱)
創(chuàng)建了 Observable 和 Observer 之后专钉，再用 subscribe() 方法將它們聯(lián)結(jié)起來(lái)，整條鏈子就可以工作了累铅。代碼形式很簡(jiǎn)單：
[java]view plaincopy
observable.subscribe(observer);
//?或者：
observable.subscribe(subscriber);
Observable.subscribe(Subscriber) 的內(nèi)部實(shí)現(xiàn)是這樣的（僅核心代碼）：
[java]view plaincopy
//?注意：這不是?subscribe()?的源碼跃须，而是將源碼中與性能、兼容性娃兽、擴(kuò)展性有關(guān)的代碼剔除后的核心代碼菇民。
//?如果需要看源碼，可以去?RxJava?的?GitHub?倉(cāng)庫(kù)下載投储。
publicSubscription?subscribe(Subscriber?subscriber)?{
subscriber.onStart();
onSubscribe.call(subscriber);
returnsubscriber;
}
可以看到第练，subscriber() 做了3件事：
1.調(diào)用 Subscriber.onStart() 。這個(gè)方法在前面已經(jīng)介紹過(guò)玛荞，是一個(gè)可選的準(zhǔn)備方法娇掏。
2.調(diào)用 Observable 中的 OnSubscribe.call(Subscriber) 。在這里勋眯，事件發(fā)送的邏輯開(kāi)始運(yùn)行婴梧。從這也可以看出下梢，在 RxJava 中， Observable 并不是在創(chuàng)建的時(shí)候就立即開(kāi)始發(fā)送事件塞蹭，而是在它被訂閱的時(shí)候孽江，即當(dāng) subscribe() 方法執(zhí)行的時(shí)候。
3.將傳入的 Subscriber 作為 Subscription 返回番电。這是為了方便 unsubscribe().
除了 subscribe(Observer) 和 subscribe(Subscriber) 岗屏，subscribe() 還支持不完整定義的回調(diào)，RxJava 會(huì)自動(dòng)根據(jù)定義創(chuàng)建出 Subscriber 漱办。形式如下：
[java]view plaincopy
Action1?onNextAction?=newAction1()?{
//?onNext()
@Override
publicvoidcall(String?s)?{
Log.d(tag,?s);
}
};
Action1?onErrorAction?=newAction1()?{
//?onError()
@Override
publicvoidcall(Throwable?throwable)?{
//?Error?handling
}
};
Action0?onCompletedAction?=newAction0()?{
//?onCompleted()
@Override
publicvoidcall()?{
Log.d(tag,"completed");
}
};
//?自動(dòng)創(chuàng)建?Subscriber?这刷，并使用?onNextAction?來(lái)定義?onNext()
observable.subscribe(onNextAction);
//?自動(dòng)創(chuàng)建?Subscriber?，并使用?onNextAction?和?onErrorAction?來(lái)定義?onNext()?和?onError()
observable.subscribe(onNextAction,?onErrorAction);
//?自動(dòng)創(chuàng)建?Subscriber?洼冻，并使用?onNextAction崭歧、?onErrorAction?和?onCompletedAction?來(lái)定義?onNext()、?onError()?和?onCompleted()
observable.subscribe(onNextAction,?onErrorAction,?onCompletedAction);
簡(jiǎn)單解釋一下這段代碼中出現(xiàn)的 Action1 和 Action0撞牢。 Action0 是 RxJava 的一個(gè)接口率碾，它只有一個(gè)方法 call()，這個(gè)方法是無(wú)參無(wú)返回值的屋彪；由于 onCompleted() 方法也是無(wú)參無(wú)返回值的所宰，因此 Action0 可以被當(dāng)成一個(gè)包裝對(duì)象，將 onCompleted() 的內(nèi)容打包起來(lái)將自己作為一個(gè)參數(shù)傳入 subscribe() 以實(shí)現(xiàn)不完整定義的回調(diào)畜挥。這樣其實(shí)也可以看做將 onCompleted() 方法作為參數(shù)傳進(jìn)了 subscribe()仔粥，相當(dāng)于其他某些語(yǔ)言中的『閉包』。 Action1 也是一個(gè)接口蟹但，它同樣只有一個(gè)方法 call(T param)躯泰，這個(gè)方法也無(wú)返回值，但有一個(gè)參數(shù)华糖；與 Action0 同理麦向，由于 onNext(T obj) 和 onError(Throwable error) 也是單參數(shù)無(wú)返回值的，因此 Action1 可以將 onNext(obj) 和 onError(error) 打包起來(lái)傳入 subscribe() 以實(shí)現(xiàn)不完整定義的回調(diào)客叉。事實(shí)上诵竭，雖然 Action0 和 Action1 在 API 中使用最廣泛，但 RxJava 是提供了多個(gè) ActionX 形式的接口 (例如 Action2, Action3) 的兼搏，它們可以被用以包裝不同的無(wú)返回值的方法卵慰。
4) 場(chǎng)景示例
下面舉兩個(gè)例子：
a. 打印字符串?dāng)?shù)組
將字符串?dāng)?shù)組 names 中的所有字符串依次打印出來(lái)：
[java]view plaincopy
String[]?names?=?...;
Observable.from(names)
.subscribe(newAction1()?{
@Override
publicvoidcall(String?name)?{
Log.d(tag,?name);
}
});
b. 由 id 取得圖片并顯示
由指定的一個(gè) drawable 文件 id drawableRes 取得圖片，并顯示在 ImageView 中佛呻，并在出現(xiàn)異常的時(shí)候打印 Toast 報(bào)錯(cuò)：
[java]view plaincopy
intdrawableRes?=?...;
ImageView?imageView?=?...;
Observable.create(newOnSubscribe()?{
@Override
publicvoidcall(Subscriber?subscriber)?{
Drawable?drawable?=?getTheme().getDrawable(drawableRes));
subscriber.onNext(drawable);
subscriber.onCompleted();
}
}).subscribe(newObserver()?{
@Override
publicvoidonNext(Drawable?drawable)?{
imageView.setImageDrawable(drawable);
}
@Override
publicvoidonCompleted()?{
}
@Override
publicvoidonError(Throwable?e)?{
Toast.makeText(activity,"Error!",?Toast.LENGTH_SHORT).show();
}
});
正如上面兩個(gè)例子這樣裳朋，創(chuàng)建出 Observable 和 Subscriber ，再用 subscribe() 將它們串起來(lái)吓著，一次 RxJava 的基本使用就完成了再扭。非常簡(jiǎn)單氧苍。
注意：在 RxJava 的默認(rèn)規(guī)則中，事件的發(fā)出和消費(fèi)都是在同一個(gè)線程的泛范。也就是說(shuō)，如果只用上面的方法紊撕，實(shí)現(xiàn)出來(lái)的只是一個(gè)同步的觀察者模式罢荡。觀察者模式本身的目的就是『后臺(tái)處理，前臺(tái)回調(diào)』的異步機(jī)制对扶，因此異步對(duì)于 RxJava 是至關(guān)重要的区赵。而要實(shí)現(xiàn)異步，則需要用到 RxJava 的另一個(gè)概念： Scheduler 浪南。
線程控制 —— Scheduler (一)
前言：
在不指定線程的情況下笼才， RxJava 遵循的是線程不變的原則，即：在哪個(gè)線程調(diào)用 subscribe()络凿，就在哪個(gè)線程生產(chǎn)事件骡送；在哪個(gè)線程生產(chǎn)事件，就在哪個(gè)線程消費(fèi)事件絮记。如果需要切換線程摔踱，就需要用到 Scheduler （調(diào)度器）。
1) Scheduler 的 API (一)
在RxJava 中怨愤，Scheduler ——調(diào)度器派敷，相當(dāng)于線程控制器，RxJava 通過(guò)它來(lái)指定每一段代碼應(yīng)該運(yùn)行在什么樣的線程撰洗。RxJava 已經(jīng)內(nèi)置了幾個(gè) Scheduler 篮愉，它們已經(jīng)適合大多數(shù)的使用場(chǎng)景：
Schedulers.immediate():直接在當(dāng)前線程運(yùn)行，相當(dāng)于不指定線程差导。這是默認(rèn)的 Scheduler试躏。
Schedulers.newThread():總是啟用新線程，并在新線程執(zhí)行操作柿汛。
Schedulers.io():I/O 操作（讀寫(xiě)文件冗酿、讀寫(xiě)數(shù)據(jù)庫(kù)、網(wǎng)絡(luò)信息交互等）所使用的 Scheduler络断。行為模式和 newThread() 差不多裁替，區(qū)別在于 io() 的內(nèi)部實(shí)現(xiàn)是是用一個(gè)無(wú)數(shù)量上限的線程池，可以重用空閑的線程貌笨，因此多數(shù)情況下 io() 比 newThread() 更有效率弱判。不要把計(jì)算工作放在 io() 中，可以避免創(chuàng)建不必要的線程锥惋。
Schedulers.computation():計(jì)算所使用的 Scheduler昌腰。這個(gè)計(jì)算指的是 CPU 密集型計(jì)算开伏，即不會(huì)被 I/O 等操作限制性能的操作，例如圖形的計(jì)算遭商。這個(gè) Scheduler 使用的固定的線程池固灵，大小為 CPU 核數(shù)。不要把 I/O 操作放在 computation() 中劫流，否則 I/O 操作的等待時(shí)間會(huì)浪費(fèi) CPU巫玻。
另外， Android 還有一個(gè)專用的 AndroidSchedulers.mainThread()祠汇，它指定的操作將在 Android 主線程運(yùn)行仍秤。
有了這幾個(gè) Scheduler ，就可以使用 subscribeOn() 和 observeOn() 兩個(gè)方法來(lái)對(duì)線程進(jìn)行控制了可很。 * subscribeOn(): 指定 subscribe() 所發(fā)生的線程诗力，即 Observable.OnSubscribe 被激活時(shí)所處的線程∥铱伲或者叫做事件產(chǎn)生的線程苇本。 * observeOn(): 指定 Subscriber 所運(yùn)行在的線程∮炝迹或者叫做事件消費(fèi)的線程圈澈。
代碼來(lái)理解上面的文字?jǐn)⑹觯?/p>
[java]view plaincopy
Observable.just(1,2,3,4)
.subscribeOn(Schedulers.io())//?指定?subscribe()?發(fā)生在?IO?線程
.observeOn(AndroidSchedulers.mainThread())//?指定?Subscriber?的回調(diào)發(fā)生在主線程
.subscribe(newAction1()?{
@Override
publicvoidcall(Integer?number)?{
Log.d(tag,"number:"+?number);
}
});
上面這段代碼中，由于 subscribeOn(Schedulers.io()) 的指定尘惧，被創(chuàng)建的事件的內(nèi)容 1康栈、2、3喷橙、4 將會(huì)在 IO 線程發(fā)出啥么；而由于 observeOn(AndroidScheculers.mainThread()) 的指定，因此 subscriber 數(shù)字的打印將發(fā)生在主線程 贰逾。事實(shí)上悬荣，這種在 subscribe() 之前寫(xiě)上兩句 subscribeOn(Scheduler.io()) 和 observeOn(AndroidSchedulers.mainThread()) 的使用方式非常常見(jiàn)，它適用于多數(shù)的 『后臺(tái)線程取數(shù)據(jù)疙剑，主線程顯示』的程序策略氯迂。
而前面提到的由圖片 id 取得圖片并顯示的例子，如果也加上這兩句：
[java]view plaincopy
intdrawableRes?=?...;
ImageView?imageView?=?...;
Observable.create(newOnSubscribe()?{
@Override
publicvoidcall(Subscriber?subscriber)?{
Drawable?drawable?=?getTheme().getDrawable(drawableRes));
subscriber.onNext(drawable);
subscriber.onCompleted();
}
})
.subscribeOn(Schedulers.io())//?指定?subscribe()?發(fā)生在?IO?線程
.observeOn(AndroidSchedulers.mainThread())//?指定?Subscriber?的回調(diào)發(fā)生在主線程
.subscribe(newObserver()?{
@Override
publicvoidonNext(Drawable?drawable)?{
imageView.setImageDrawable(drawable);
}
@Override
publicvoidonCompleted()?{
}
@Override
publicvoidonError(Throwable?e)?{
Toast.makeText(activity,"Error!",?Toast.LENGTH_SHORT).show();
}
});
那么言缤，加載圖片將會(huì)發(fā)生在 IO 線程嚼蚀，而設(shè)置圖片則被設(shè)定在了主線程。這就意味著管挟，即使加載圖片耗費(fèi)了幾十甚至幾百毫秒的時(shí)間轿曙，也不會(huì)造成絲毫界面的卡頓。
2) Scheduler 的原理 (一)
RxJava 的 Scheduler API 很方便，也很神奇（加了一句話就把線程切換了导帝，怎么做到的守谓？而且 subscribe() 不是最外層直接調(diào)用的方法嗎，它竟然也能被指定線程您单？）斋荞。然而 Scheduler 的原理需要放在后面講，因?yàn)樗脑硎且韵乱还?jié)《變換》的原理作為基礎(chǔ)的睹限。
好吧這一節(jié)其實(shí)我屁也沒(méi)說(shuō)譬猫，只是為了讓你安心，讓你知道我不是忘了講原理羡疗，而是把它放在了更合適的地方。
變換
RxJava 提供了對(duì)事件序列進(jìn)行變換的支持别洪，這是它的核心功能之一叨恨，也是大多數(shù)人說(shuō)『RxJava 真是太好用了』的最大原因。所謂變換挖垛，就是將事件序列中的對(duì)象或整個(gè)序列進(jìn)行加工處理痒钝，轉(zhuǎn)換成不同的事件或事件序列。概念說(shuō)著總是模糊難懂的痢毒，來(lái)看 API送矩。
1) API
首先看一個(gè) map() 的例子：
[java]view plaincopy
Observable.just("images/logo.png")//?輸入類型?String
.map(newFunc1()?{
@Override
publicBitmap?call(String?filePath)?{//?參數(shù)類型?String
returngetBitmapFromPath(filePath);//?返回類型?Bitmap
}
})
.subscribe(newAction1()?{
@Override
publicvoidcall(Bitmap?bitmap)?{//?參數(shù)類型?Bitmap
showBitmap(bitmap);
}
});
這里出現(xiàn)了一個(gè)叫做 Func1 的類。它和 Action1 非常相似哪替，也是 RxJava 的一個(gè)接口栋荸，用于包裝含有一個(gè)參數(shù)的方法。 Func1 和 Action 的區(qū)別在于凭舶， Func1 包裝的是有返回值的方法晌块。另外，和 ActionX 一樣帅霜， FuncX 也有多個(gè)匆背，用于不同參數(shù)個(gè)數(shù)的方法。FuncX 和 ActionX 的區(qū)別在 FuncX 包裝的是有返回值的方法身冀。
可以看到钝尸，map() 方法將參數(shù)中的 String 對(duì)象轉(zhuǎn)換成一個(gè) Bitmap 對(duì)象后返回，而在經(jīng)過(guò) map() 方法后搂根，事件的參數(shù)類型也由 String 轉(zhuǎn)為了 Bitmap珍促。這種直接變換對(duì)象并返回的，是最常見(jiàn)的也最容易理解的變換兄墅。不過(guò) RxJava 的變換遠(yuǎn)不止這樣踢星，它不僅可以針對(duì)事件對(duì)象，還可以針對(duì)整個(gè)事件隊(duì)列，這使得 RxJava 變得非常靈活沐悦。我列舉幾個(gè)常用的變換：
map(): 事件對(duì)象的直接變換成洗，具體功能上面已經(jīng)介紹過(guò)。它是 RxJava 最常用的變換藏否。 map() 的示意圖：
flatMap(): 這是一個(gè)很有用但非常難理解的變換瓶殃，因此我決定花多些篇幅來(lái)介紹它。 首先假設(shè)這么一種需求：假設(shè)有一個(gè)數(shù)據(jù)結(jié)構(gòu)『學(xué)生』副签，現(xiàn)在需要打印出一組學(xué)生的名字遥椿。實(shí)現(xiàn)方式很簡(jiǎn)單：
[java]view plaincopy
Student[]?students?=?...;
Subscriber?subscriber?=newSubscriber()?{
@Override
publicvoidonNext(String?name)?{
Log.d(tag,?name);
}
...
};
Observable.from(students)
.map(newFunc1()?{
@Override
publicString?call(Student?student)?{
returnstudent.getName();
}
})
.subscribe(subscriber);
很簡(jiǎn)單。那么再假設(shè)：如果要打印出每個(gè)學(xué)生所需要修的所有課程的名稱呢淆储？（需求的區(qū)別在于冠场，每個(gè)學(xué)生只有一個(gè)名字，但卻有多個(gè)課程本砰。）首先可以這樣實(shí)現(xiàn)：
[java]view plaincopy
Student[]?students?=?...;
Subscriber?subscriber?=newSubscriber()?{
@Override
publicvoidonNext(Student?student)?{
List?courses?=?student.getCourses();
for(inti?=0;?i?<?courses.size();?i++)?{
Course?course?=?courses.get(i);
Log.d(tag,?course.getName());
}
}
...
};
Observable.from(students)
.subscribe(subscriber);
依然很簡(jiǎn)單碴裙。那么如果我不想在 Subscriber 中使用 for 循環(huán)，而是希望 Subscriber 中直接傳入單個(gè)的 Course 對(duì)象呢（這對(duì)于代碼復(fù)用很重要）点额？用 map() 顯然是不行的舔株，因?yàn)?map() 是一對(duì)一的轉(zhuǎn)化，而我現(xiàn)在的要求是一對(duì)多的轉(zhuǎn)化还棱。那怎么才能把一個(gè) Student 轉(zhuǎn)化成多個(gè) Course 呢载慈？
這個(gè)時(shí)候，就需要用 flatMap() 了：
[java]view plaincopy
Student[]?students?=?...;
Subscriber?subscriber?=newSubscriber()?{
@Override
publicvoidonNext(Course?course)?{
Log.d(tag,?course.getName());
}
...
};
Observable.from(students)
.flatMap(newFunc1>()?{
@Override
publicObservable?call(Student?student)?{
returnObservable.from(student.getCourses());
}
})
.subscribe(subscriber);
從上面的代碼可以看出珍手， flatMap() 和 map() 有一個(gè)相同點(diǎn)：它也是把傳入的參數(shù)轉(zhuǎn)化之后返回另一個(gè)對(duì)象办铡。但需要注意，和 map() 不同的是珠十， flatMap() 中返回的是個(gè) Observable 對(duì)象料扰，并且這個(gè) Observable 對(duì)象并不是被直接發(fā)送到了 Subscriber 的回調(diào)方法中。 flatMap() 的原理是這樣的：1. 使用傳入的事件對(duì)象創(chuàng)建一個(gè) Observable 對(duì)象焙蹭；2. 并不發(fā)送這個(gè) Observable, 而是將它激活晒杈，于是它開(kāi)始發(fā)送事件；3. 每一個(gè)創(chuàng)建出來(lái)的 Observable 發(fā)送的事件孔厉，都被匯入同一個(gè) Observable 拯钻，而這個(gè) Observable 負(fù)責(zé)將這些事件統(tǒng)一交給 Subscriber 的回調(diào)方法。這三個(gè)步驟撰豺，把事件拆成了兩級(jí)粪般，通過(guò)一組新創(chuàng)建的 Observable 將初始的對(duì)象『鋪平』之后通過(guò)統(tǒng)一路徑分發(fā)了下去。而這個(gè)『鋪平』就是 flatMap() 所謂的 flat污桦。
擴(kuò)展：由于可以在嵌套的 Observable 中添加異步代碼亩歹， flatMap() 也常用于嵌套的異步操作，例如嵌套的網(wǎng)絡(luò)請(qǐng)求。示例代碼（Retrofit + RxJava）：
[java]view plaincopy
networkClient.token()//?返回?Observable小作，在訂閱時(shí)請(qǐng)求?token亭姥，并在響應(yīng)后發(fā)送?token
.flatMap(newFunc1>()?{
@Override
publicObservable?call(String?token)?{
//?返回?Observable，在訂閱時(shí)請(qǐng)求消息列表顾稀，并在響應(yīng)后發(fā)送請(qǐng)求到的消息列表
returnnetworkClient.messages();
}
})
.subscribe(newAction1()?{
@Override
publicvoidcall(Messages?messages)?{
//?處理顯示消息列表
showMessages(messages);
}
});
傳統(tǒng)的嵌套請(qǐng)求需要使用嵌套的 Callback 來(lái)實(shí)現(xiàn)达罗。而通過(guò) flatMap() ，可以把嵌套的請(qǐng)求寫(xiě)在一條鏈中静秆，從而保持程序邏輯的清晰粮揉。
throttleFirst(): 在每次事件觸發(fā)后的一定時(shí)間間隔內(nèi)丟棄新的事件。常用作去抖動(dòng)過(guò)濾抚笔，例如按鈕的點(diǎn)擊監(jiān)聽(tīng)器： RxView.clickEvents(button) // RxBinding 代碼扶认，后面的文章有解釋 .throttleFirst(500, TimeUnit.MILLISECONDS) // 設(shè)置防抖間隔為 500ms .subscribe(subscriber); 媽媽再也不怕我的用戶手抖點(diǎn)開(kāi)兩個(gè)重復(fù)的界面啦。
此外殊橙， RxJava 還提供很多便捷的方法來(lái)實(shí)現(xiàn)事件序列的變換蝠引，這里就不一一舉例了。
2) 變換的原理：lift()
這些變換雖然功能各有不同蛀柴，但實(shí)質(zhì)上都是針對(duì)事件序列的處理和再發(fā)送。而在 RxJava 的內(nèi)部矫夯，它們是基于同一個(gè)基礎(chǔ)的變換方法： lift(Operator)鸽疾。首先看一下 lift() 的內(nèi)部實(shí)現(xiàn)（僅核心代碼）：
// 注意：這不是 lift() 的源碼，而是將源碼中與性能训貌、兼容性制肮、擴(kuò)展性有關(guān)的代碼剔除后的核心代碼。
// 如果需要看源碼递沪，可以去 RxJava 的 GitHub 倉(cāng)庫(kù)下載豺鼻。
[java]view plaincopy
public?Observable?lift(Operator?operator)?{
returnObservable.create(newOnSubscribe()?{
@Override
publicvoidcall(Subscriber?subscriber)?{
Subscriber?newSubscriber?=?operator.call(subscriber);
newSubscriber.onStart();
onSubscribe.call(newSubscriber);
}
});
}
這段代碼很有意思：它生成了一個(gè)新的 Observable 并返回，而且創(chuàng)建新 Observable 所用的參數(shù) OnSubscribe 的回調(diào)方法 call() 中的實(shí)現(xiàn)竟然看起來(lái)和前面講過(guò)的 Observable.subscribe() 一樣款慨！然而它們并不一樣喲~不一樣的地方關(guān)鍵就在于第二行 onSubscribe.call(subscriber) 中的 onSubscribe 所指代的對(duì)象不同（高能預(yù)警：接下來(lái)的幾句話可能會(huì)導(dǎo)致身體的嚴(yán)重不適）——
subscribe() 中這句話的 onSubscribe 指的是 Observable 中的 onSubscribe 對(duì)象儒飒，這個(gè)沒(méi)有問(wèn)題，但是 lift() 之后的情況就復(fù)雜了點(diǎn)檩奠。
當(dāng)含有 lift() 時(shí)：
1.lift() 創(chuàng)建了一個(gè) Observable 后桩了，加上之前的原始 Observable，已經(jīng)有兩個(gè) Observable 了埠戳；
2.而同樣地井誉，新 Observable 里的新 OnSubscribe 加上之前的原始 Observable 中的原始 OnSubscribe，也就有了兩個(gè) OnSubscribe整胃；
3.當(dāng)用戶調(diào)用經(jīng)過(guò) lift() 后的 Observable 的 subscribe() 的時(shí)候颗圣，使用的是 lift() 所返回的新的 Observable ，于是它所觸發(fā)的 onSubscribe.call(subscriber)，也是用的新 Observable 中的新 OnSubscribe在岂，即在 lift() 中生成的那個(gè) OnSubscribe奔则；
4.而這個(gè)新 OnSubscribe 的 call() 方法中的 onSubscribe ，就是指的原始 Observable 中的原始 OnSubscribe 洁段，在這個(gè) call() 方法里应狱，新 OnSubscribe 利用 operator.call(subscriber) 生成了一個(gè)新的 Subscriber（Operator 就是在這里，通過(guò)自己的 call() 方法將新 Subscriber 和原始 Subscriber 進(jìn)行關(guān)聯(lián)祠丝，并插入自己的『變換』代碼以實(shí)現(xiàn)變換）疾呻，然后利用這個(gè)新 Subscriber 向原始 Observable 進(jìn)行訂閱。
這樣就實(shí)現(xiàn)了 lift() 過(guò)程写半，有點(diǎn)像一種代理機(jī)制岸蜗，通過(guò)事件攔截和處理實(shí)現(xiàn)事件序列的變換。
精簡(jiǎn)掉細(xì)節(jié)的話叠蝇，也可以這么說(shuō)：在 Observable 執(zhí)行了 lift(Operator) 方法之后璃岳，會(huì)返回一個(gè)新的 Observable，這個(gè)新的 Observable 會(huì)像一個(gè)代理一樣悔捶，負(fù)責(zé)接收原始的 Observable 發(fā)出的事件铃慷，并在處理后發(fā)送給 Subscriber。
舉一個(gè)具體的 Operator 的實(shí)現(xiàn)蜕该。下面這是一個(gè)將事件中的 Integer 對(duì)象轉(zhuǎn)換成 String 的例子犁柜，僅供參考：
[java]view plaincopy
observable.lift(newObservable.Operator()?{
@Override
publicSubscriber?call(finalSubscriber?subscriber)?{
//?將事件序列中的?Integer?對(duì)象轉(zhuǎn)換為?String?對(duì)象
returnnewSubscriber()?{
@Override
publicvoidonNext(Integer?integer)?{
subscriber.onNext(""+?integer);
}
@Override
publicvoidonCompleted()?{
subscriber.onCompleted();
}
@Override
publicvoidonError(Throwable?e)?{
subscriber.onError(e);
}
};
}
});
3) compose: 對(duì) Observable 整體的變換
除了 lift() 之外， Observable 還有一個(gè)變換方法叫做 compose(Transformer)堂淡。它和 lift() 的區(qū)別在于馋缅， lift() 是針對(duì)事件項(xiàng)和事件序列的，而 compose() 是針對(duì) Observable 自身進(jìn)行變換绢淀。舉個(gè)例子萤悴，假設(shè)在程序中有多個(gè) Observable ，并且他們都需要應(yīng)用一組相同的 lift() 變換皆的。你可以這么寫(xiě)：
[java]view plaincopy
observable1
.lift1()
.lift2()
.lift3()
.lift4()
.subscribe(subscriber1);
observable2
.lift1()
.lift2()
.lift3()
.lift4()
.subscribe(subscriber2);
observable3
.lift1()
.lift2()
.lift3()
.lift4()
.subscribe(subscriber3);
observable4
.lift1()
.lift2()
.lift3()
.lift4()
.subscribe(subscriber1);
你覺(jué)得這樣太不軟件工程了覆履，于是你改成了這樣：
[java]view plaincopy
privateObservable?liftAll(Observable?observable)?{
returnobservable
.lift1()
.lift2()
.lift3()
.lift4();
}
...
liftAll(observable1).subscribe(subscriber1);
liftAll(observable2).subscribe(subscriber2);
liftAll(observable3).subscribe(subscriber3);
liftAll(observable4).subscribe(subscriber4);
可讀性、可維護(hù)性都提高了祭务∧诠罚可是 Observable 被一個(gè)方法包起來(lái)，這種方式對(duì)于 Observale 的靈活性似乎還是增添了那么點(diǎn)限制义锥。怎么辦柳沙？這個(gè)時(shí)候，就應(yīng)該用 compose() 來(lái)解決了：
[java]view plaincopy
publicclassLiftAllTransformerimplementsObservable.Transformer?{
@Override
publicObservable?call(Observable?observable)?{
returnobservable
.lift1()
.lift2()
.lift3()
.lift4();
}
}
...
Transformer?liftAll?=newLiftAllTransformer();
observable1.compose(liftAll).subscribe(subscriber1);
observable2.compose(liftAll).subscribe(subscriber2);
observable3.compose(liftAll).subscribe(subscriber3);
observable4.compose(liftAll).subscribe(subscriber4);
像上面這樣拌倍，使用 compose() 方法赂鲤，Observable 可以利用傳入的 Transformer 對(duì)象的 call 方法直接對(duì)自身進(jìn)行處理噪径，也就不必被包在方法的里面了。
compose() 的原理比較簡(jiǎn)單数初，不附圖嘍找爱。
線程控制：Scheduler (二)
除了靈活的變換，RxJava 另一個(gè)牛逼的地方泡孩，就是線程的自由控制车摄。
1) Scheduler 的 API (二)
前面講到了，可以利用 subscribeOn() 結(jié)合 observeOn() 來(lái)實(shí)現(xiàn)線程控制仑鸥，讓事件的產(chǎn)生和消費(fèi)發(fā)生在不同的線程吮播。可是在了解了 map() flatMap() 等變換方法后眼俊，有些好事的（其實(shí)就是當(dāng)初剛接觸 RxJava 時(shí)的我）就問(wèn)了：能不能多切換幾次線程意狠？
答案是：能。因?yàn)?observeOn() 指定的是 Subscriber 的線程疮胖，而這個(gè) Subscriber 并不是（嚴(yán)格說(shuō)應(yīng)該為『不一定是』环戈，但這里不妨理解為『不是』）subscribe() 參數(shù)中的 Subscriber ，而是 observeOn() 執(zhí)行時(shí)的當(dāng)前 Observable 所對(duì)應(yīng)的 Subscriber 澎灸，即它的直接下級(jí) Subscriber 院塞。換句話說(shuō)，observeOn() 指定的是它之后的操作所在的線程性昭。因此如果有多次切換線程的需求迫悠，只要在每個(gè)想要切換線程的位置調(diào)用一次 observeOn() 即可。上代碼：
[java]view plaincopy
Observable.just(1,2,3,4)//?IO?線程巩梢，由?subscribeOn()?指定
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.map(mapOperator)//?新線程，由?observeOn()?指定
.observeOn(Schedulers.io())
.map(mapOperator2)//?IO?線程艺玲，由?observeOn()?指定
.observeOn(AndroidSchedulers.mainThread)
.subscribe(subscriber);//?Android?主線程括蝠，由?observeOn()?指定
如上，通過(guò) observeOn() 的多次調(diào)用饭聚，程序?qū)崿F(xiàn)了線程的多次切換忌警。
不過(guò)，不同于 observeOn() 秒梳， subscribeOn() 的位置放在哪里都可以法绵，但它是只能調(diào)用一次的。
又有好事的（其實(shí)還是當(dāng)初的我）問(wèn)了：如果我非要調(diào)用多次 subscribeOn() 呢酪碘？會(huì)有什么效果朋譬？
這個(gè)問(wèn)題先放著，我們還是從 RxJava 線程控制的原理說(shuō)起吧兴垦。
2) Scheduler 的原理（二）
其實(shí)徙赢， subscribeOn() 和 observeOn() 的內(nèi)部實(shí)現(xiàn)字柠，也是用的 lift()。具體看圖（不同顏色的箭頭表示不同的線程）：
從圖中可以看出狡赐，subscribeOn() 和 observeOn() 都做了線程切換的工作（圖中的 "schedule..." 部位）窑业。不同的是， subscribeOn() 的線程切換發(fā)生在 OnSubscribe 中枕屉，即在它通知上一級(jí) OnSubscribe 時(shí)常柄，這時(shí)事件還沒(méi)有開(kāi)始發(fā)送，因此 subscribeOn() 的線程控制可以從事件發(fā)出的開(kāi)端就造成影響搀擂；而 observeOn() 的線程切換則發(fā)生在它內(nèi)建的 Subscriber 中西潘，即發(fā)生在它即將給下一級(jí) Subscriber 發(fā)送事件時(shí)，因此 observeOn() 控制的是它后面的線程哥倔。
3) 延伸：doOnSubscribe()
然而秸架，雖然超過(guò)一個(gè)的 subscribeOn() 對(duì)事件處理的流程沒(méi)有影響，但在流程之前卻是可以利用的咆蒿。
在前面講 Subscriber 的時(shí)候东抹，提到過(guò) Subscriber 的 onStart() 可以用作流程開(kāi)始前的初始化。然而 onStart() 由于在 subscribe() 發(fā)生時(shí)就被調(diào)用了沃测，因此不能指定線程缭黔，而是只能執(zhí)行在 subscribe() 被調(diào)用時(shí)的線程。這就導(dǎo)致如果 onStart() 中含有對(duì)線程有要求的代碼（例如在界面上顯示一個(gè) ProgressBar蒂破，這必須在主線程執(zhí)行）馏谨，將會(huì)有線程非法的風(fēng)險(xiǎn)，因?yàn)橛袝r(shí)你無(wú)法預(yù)測(cè) subscribe() 將會(huì)在什么線程執(zhí)行附迷。
而與 Subscriber.onStart() 相對(duì)應(yīng)的惧互，有一個(gè)方法 Observable.doOnSubscribe() 。它和 Subscriber.onStart() 同樣是在 subscribe() 調(diào)用后而且在事件發(fā)送前執(zhí)行喇伯，但區(qū)別在于它可以指定線程喊儡。默認(rèn)情況下， doOnSubscribe() 執(zhí)行在 subscribe() 發(fā)生的線程稻据；而如果在 doOnSubscribe() 之后有 subscribeOn() 的話艾猜，它將執(zhí)行在離它最近的 subscribeOn() 所指定的線程。
示例代碼：
[java]view plaincopy
Observable.create(onSubscribe)
.subscribeOn(Schedulers.io())
.doOnSubscribe(newAction0()?{
@Override
publicvoidcall()?{
progressBar.setVisibility(View.VISIBLE);//?需要在主線程執(zhí)行
}
})
.subscribeOn(AndroidSchedulers.mainThread())//?指定主線程
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriber);
如上捻悯，在 doOnSubscribe()的后面跟一個(gè) subscribeOn() 匆赃，就能指定準(zhǔn)備工作的線程了。
