原文地址：RxJava 入門
什么是 ReactiveX脑漫？
ReactiveX 是一個(gè)專注于異步編程與控制可觀察數(shù)據(jù)（或者事件）流的API番电。它組合了觀察者模式刘莹，迭代器模式和函數(shù)式編程的優(yōu)秀思想祝沸。
實(shí)時(shí)數(shù)據(jù)處理是一件普通的現(xiàn)象使套，有一個(gè)高效病往、干凈和可擴(kuò)展的方式來處理這些情景是重要的。使用 Observables 和 Operators 來熟練操作它們冤议。ReactiveX 提供一個(gè)可組合又靈活的 API 來創(chuàng)建和處理數(shù)據(jù)流斟薇，同時(shí)簡(jiǎn)化了異步編程帶來的一些擔(dān)憂，如：線程創(chuàng)建和并發(fā)問題求类。
RxJava 簡(jiǎn)介
RxJava 是 ReactiveX 在 Java 上的開源的實(shí)現(xiàn)奔垦。Observable（觀察者） 和 Subscriber（訂閱者）是兩個(gè)主要的類。在 RxJava 上尸疆，一個(gè) Observable 是一個(gè)發(fā)出數(shù)據(jù)流或者事件的類，Subscriber 是一個(gè)對(duì)這些發(fā)出的 items （數(shù)據(jù)流或者事件）進(jìn)行處理（采取行動(dòng)）的類惶岭。一個(gè) Observable 的標(biāo)準(zhǔn)流發(fā)出一個(gè)或多個(gè) item寿弱，然后成功完成或者出錯(cuò)。一個(gè) Observable 可以有多個(gè) Subscribers按灶，并且通過 Observable 發(fā)出的每一個(gè) item症革，該 item 將會(huì)被發(fā)送到 Subscriber.onNext() 方法來進(jìn)行處理。一旦 Observable 不再發(fā)出 items鸯旁，它將會(huì)調(diào)用 Subscriber.onCompleted() 方法噪矛，或如果有一個(gè)出錯(cuò)的話 Observable 會(huì)調(diào)用 Subscriber.onError() 方法。
現(xiàn)在铺罢，我們知道了很多關(guān)于 Observable 和 Subscriber 類艇挨，我們可以繼續(xù)去介紹有關(guān) Observables 的創(chuàng)建和訂閱。
Observable integerObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext(1);
subscriber.onNext(2);
subscriber.onNext(3);
subscriber.onCompleted();
}
});
這個(gè) Observable 發(fā)出了整數(shù) 1韭赘，2缩滨，3 然后結(jié)束了。現(xiàn)在我們需要?jiǎng)?chuàng)建一個(gè) Subscriber，那樣我們就能讓這些發(fā)出的流起作用脉漏。
Subscriber integerSubscriber = new Subscriber() {
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer value) {
System.out.println("onNext: " + value);
}
};
我們的 Subscriber 只是簡(jiǎn)單的把任何發(fā)出的 items 打印出來苞冯，完成之后通知我們。一旦你有一個(gè) Observable 和一個(gè) Subscriber侧巨，可以通過 Observable.subscribe() 方法將他們聯(lián)系起來舅锄。
integerObservable.subscribe(integerSubscriber);
// Outputs:
// onNext: 1
// onNext: 2
// onNext: 3
// Complete!
上面所有這些代碼可以簡(jiǎn)單的通過使用 Observable.just() 方法來創(chuàng)建一個(gè) Observable 去發(fā)出這些定義的值，并且我們的 Subscriber 可以改變成匿名的內(nèi)部類司忱，如下：
Observable.just(1, 2 ,3).subscribe(new Subscriber() {
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(Integer value) {
System.out.println("onNext: " + value);
}
});
操作符
創(chuàng)建和訂閱一個(gè) Observable 是足夠簡(jiǎn)單的巧娱，可能這并不是非常有用的，但這只是用 RxJava 的一個(gè)開始烘贴。通過調(diào)用操作符禁添，任何的 Observable 都能進(jìn)行輸出轉(zhuǎn)變，多個(gè) Operators 能鏈接到 Observable 上桨踪。例如老翘，在我們剛才的 Observable 中，我們只想要收到奇數(shù)的數(shù)字锻离。要做到這一點(diǎn)铺峭，我可以使用 filter() 操作符。
Observable.just(1, 2, 3, 4, 5, 6) // add more numbers
.filter(new Func1() {
@Override
public Boolean call(Integer value) {
return value % 2 == 1;
}
})
.subscribe(new Subscriber() {
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer value) {
System.out.println("onNext: " + value);
}
});
// Outputs:
// onNext: 1
// onNext: 3
// onNext: 5
// Complete!
我們的 filter() 操作符定義了一個(gè)方法汽纠，將取出我們發(fā)出的整數(shù)卫键，并對(duì)所有的奇數(shù)返回為 true，所有的偶數(shù)返回為 false虱朵。從我們的 filter() 返回為 false 的值是不會(huì)發(fā)出到 Subscriber 的莉炉，我們也不會(huì)在輸出中看到他們。注意：filter() 操作符返回的是一個(gè) Observable碴犬，這樣我們的訂閱方式就可以像之前的做法那樣了絮宁。
現(xiàn)在，我想找到發(fā)出的這些奇數(shù)的平方根服协，一種方法是在調(diào)用我們的 Subscriber 的每一個(gè) onNext() 去計(jì)算平方根绍昂。然而，如果我們?cè)谖覀兊?Subscriber 中做計(jì)算平方根的操作的話偿荷，這樣得到期望可能就不能進(jìn)一步實(shí)現(xiàn)的數(shù)據(jù)的流式轉(zhuǎn)換了窘游。要做到這一點(diǎn)，我們可以在 filter() 操作符上鏈上 map() 操作符跳纳。
Observable.just(1, 2, 3, 4, 5, 6) // add more numbers
.filter(new Func1() {
@Override
public Boolean call(Integer value) {
return value % 2 == 1;
}
})
.map(new Func1() {
@Override
public Double call(Integer value) {
return Math.sqrt(value);
}
})
.subscribe(new Subscriber() { // notice Subscriber type changed to
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Double value) {
System.out.println("onNext: " + value);
}
});
// Outputs:
// onNext: 1.0
// onNext: 1.7320508075688772
// onNext: 2.23606797749979
// Complete!
操作符的鏈?zhǔn)绞褂檬菢?gòu)成 RxJava 必不可少的一部分忍饰，讓你可以靈活的實(shí)現(xiàn)任何你想要的需求。隨著對(duì)于 Observables 和 Operators 相互作用的理解棒旗，我們可以進(jìn)入下一個(gè)話題：RxJava 和 Android 的整合喘批。
讓 Android 中的線程操作變得簡(jiǎn)單
在 Android 開發(fā)中有一個(gè)常見的場(chǎng)景是需要在后臺(tái)線程去分擔(dān)一定量的工作撩荣，一旦該任務(wù)完成，會(huì)將結(jié)果回調(diào)到主線程去顯示結(jié)果饶深。
在 Android 中餐曹，我們有多種方法來做這樣的事：用 AsyncTasks，Loaders敌厘，Services 等台猴。然而，這些解決方式通常不是最好的俱两。Asynctasks 很容易導(dǎo)致內(nèi)存泄露饱狂，CursorLoaders 與 ContentProvider 需要大量的配置和設(shè)置樣板代碼，還有 Services 的目的是為了長(zhǎng)時(shí)間在后臺(tái)運(yùn)營(yíng)的宪彩，而不是處理快速完成的操作休讳，如：做一個(gè)網(wǎng)絡(luò)請(qǐng)求或者從數(shù)據(jù)庫(kù)加載內(nèi)容。
讓我們看看 RxJava 是怎么幫我們解決這些問題的尿孔。下面這樣的布局有一個(gè)按鈕去開始一個(gè)長(zhǎng)時(shí)間運(yùn)行的操作俊柔，并且始終顯示進(jìn)度條，這樣我們可以確保我們的操作是運(yùn)行在后臺(tái)線程的而不是在主線程活合。
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
xmlns:app="http://schemas.android.com/apk/res-auto"
android:id="@+id/root_view"
android:layout_width="match_parent"
android:layout_height="match_parent"
android:fitsSystemWindows="true"
android:orientation="vertical">
<android.support.v7.widget.Toolbar
android:id="@+id/toolbar"
android:layout_width="match_parent"
android:layout_height="?attr/actionBarSize"
android:background="?attr/colorPrimary"
app:popupTheme="@style/AppTheme.PopupOverlay"
app:theme="@style/ThemeOverlay.AppCompat.Dark.ActionBar" />
<Button
android:id="@+id/start_btn"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_gravity="center_horizontal"
android:text="@string/start_operation_text" />
<ProgressBar
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_gravity="center_horizontal"
android:indeterminate="true" />
</LinearLayout>
一旦按鈕被點(diǎn)擊雏婶，它會(huì)禁用按鈕并開啟長(zhǎng)時(shí)間運(yùn)行的操作，并且一旦這個(gè)操作完成便會(huì)顯示一個(gè) Snackbar白指，然后按鈕會(huì)重新變得可點(diǎn)擊留晚。這里是一個(gè)用 AsyncTask 實(shí)現(xiàn)我們這個(gè)“長(zhǎng)期運(yùn)行的操作”的例子。這個(gè)按鈕只是 new 了一個(gè) SampleAsyncTask 并 executes 了它告嘲。
public String longRunningOperation() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
// error
}
return "Complete!";
}
private class SampleAsyncTask extends AsyncTask {
@Override
protected String doInBackground(Void... params) {
return longRunningOperation();
}
@Override
protected void onPostExecute(String result) {
Snackbar.make(rootView, result, Snackbar.LENGTH_LONG).show();
startAsyncTaskButton.setEnabled(true);
}
}
現(xiàn)在错维，我們?nèi)绾螌⑦@個(gè) AsyncTask 用 RxJava 來實(shí)現(xiàn)呢？首先状蜗，我們需要添加以下內(nèi)容到我們 app 的 gradle build 文件下: compile 'io.reactivex:rxjava:1.0.14' 需五。然后我們需要?jiǎng)?chuàng)建一個(gè) Observable 來調(diào)用我們這個(gè)長(zhǎng)時(shí)間運(yùn)行的操作。這可以使用 Observable.create() 方法來做到轧坎。
final Observable operationObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext(longRunningOperation());
subscriber.onCompleted();
}
});
我們創(chuàng)建了 Observable 將會(huì)調(diào)用 longRunningOperation() 方法，將返回的結(jié)果作為參數(shù)給 onNext() 方法泽示，然后調(diào)用 onCompleted() 來完成 Observable （注：在我們的 Observable 去訂閱之前缸血，我們的操作是不會(huì)被調(diào)用的）。接下來械筛，當(dāng) button 被點(diǎn)擊時(shí)捎泻，我們需要給我們的 Observable 做訂閱。我添加了一個(gè)新的 button 用 RxJava 版本來處理我們的任務(wù)埋哟。
startRxOperationButton.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(final View v) {
v.setEnabled(false);
operationObservable.subscribe(new Subscriber() {
@Override
public void onCompleted() {
v.setEnabled(true);
}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(String value) {
Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show();
}
});
}
});
現(xiàn)在當(dāng)我們建立應(yīng)用程序時(shí)笆豁，然后點(diǎn)擊新 button 時(shí)郎汪，會(huì)發(fā)生什么？我們的進(jìn)度顯示會(huì)凍結(jié)闯狱，然后我們 UI 變得反應(yīng)遲鈍煞赢。這是因?yàn)槲覀冞€沒有定義我們的 Observable 應(yīng)該在什么線程上，以及我們應(yīng)該在什么線程去訂閱它哄孤。這是 RxJava 的 Schedulers（調(diào)度器） 功能照筑。
對(duì)于任何 Observable 你可以定義在兩個(gè)不同的線程，Observable 會(huì)操作在它上面瘦陈。使用 Observable.observeOn() 可以定義在一個(gè)線程上凝危，可以用來監(jiān)聽和檢查從 Observable 最新發(fā)出的 items （Subscriber 的 onNext，onCompleted 和 onError 方法會(huì)執(zhí)行在 observeOn 所指定的線程上）晨逝，并使用 Observable.subscribeOn() 來定義一個(gè)線程蛾默，將其運(yùn)行我們 Observable 的代碼（長(zhǎng)時(shí)間運(yùn)行的操作）。
RxJava 默認(rèn)情況下是單線程的捉貌，你會(huì)需要利用 observeOn() 和 subscribeOn() 方法為你的應(yīng)用帶來多線程操作支鸡。RxJava 附帶了幾個(gè)現(xiàn)成的 Schedulers 給 Observables 使用，如：Schedulers.io() （用于 I/O 操作）昏翰，Schedulers.computation()（計(jì)算工作）苍匆，和 Schedulers.newThread()（為任務(wù)創(chuàng)建的新線程）。然而棚菊，從 Android 的角度來看浸踩，你可能想知道如何把訂閱代碼執(zhí)行到主線程。我們可以用 RxAndroid 庫(kù)來實(shí)現(xiàn)這一目標(biāo)统求。
RxAndroid 是一個(gè)對(duì) RxJava 的輕量級(jí)擴(kuò)展為了 Android 的主線程提供 Scheduler检碗，也能去創(chuàng)建一個(gè) Scheduler 用于運(yùn)行在任何給定的 Android Handler 類上。用新的 Schedulers码邻，Observable 創(chuàng)建之前能讓我們將其修改為在后臺(tái)線程執(zhí)行我們的任務(wù)折剃，并將我們的結(jié)果推到主線程上。
要在 APP 中用 RxAndroid像屋，只要在 gradle build 文件中添加這行代碼就行了：compile 'io.reactivex:rxandroid:1.0.1'怕犁。
final Observable operationObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext(longRunningOperation());
subscriber.onCompleted();
}
})
.subscribeOn(Schedulers.io()) // subscribeOn the I/O thread
.observeOn(AndroidSchedulers.mainThread()); // observeOn the UI Thread
我們修改 Observable 將用 Schedulers.io() 去訂閱，并用 AndroidSchedulers.mainThread() 方法將觀察的結(jié)果返回到 UI 線程上 〖狠海現(xiàn)在奏甫，當(dāng)我們建立我們的 APP 并點(diǎn)擊我們的 Rx 操作的按鈕，我們可以看到當(dāng)操作運(yùn)行時(shí)它將不再阻塞 UI 線程凌受。
所有上述的例子利用了 Observable 類來發(fā)出我們的結(jié)果阵子，當(dāng)一個(gè)操作僅僅只需要發(fā)出一個(gè)結(jié)果然后就完成的情況我們可以有另外一個(gè)選擇。RxJava 發(fā)布的 1.0.13 版本介紹了 Single 類胜蛉。Single 類可以用于創(chuàng)建像下面這樣的方法：
Subscription subscription = Single.create(new Single.OnSubscribe() {
@Override
public void call(SingleSubscriber singleSubscriber) {
String value = longRunningOperation();
singleSubscriber.onSuccess(value);
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1() {
@Override
public void call(String value) {
// onSuccess
Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show();
}
}, new Action1() {
@Override
public void call(Throwable throwable) {
// handle onError
}
});
當(dāng)給一個(gè) Single 類做訂閱時(shí)挠进，只有一個(gè) onSuccess 的 Action 和 onError 的 action色乾。Single 類有不同于 Observable 的操作符，有幾個(gè)操作符具有將 Single 轉(zhuǎn)換到 Observable 的機(jī)制领突。例如：用 Single.mergeWith() 操作符暖璧，兩個(gè)或更多同類型的 Singles 可以合并到一起去創(chuàng)建一個(gè) Observable，發(fā)出每個(gè) Single 的結(jié)果給一個(gè) Observable攘须。
防止內(nèi)存泄露
對(duì)于 AsyncTasks 所提到的缺點(diǎn)是漆撞，如果對(duì)于涉及了 Activity 或 Fragment 的處理不仔細(xì)的話，AsyncTasks 可能會(huì)造成內(nèi)存泄露于宙。不幸的是浮驳，使用 RxJava 不會(huì)魔術(shù)般的緩解內(nèi)存泄露危機(jī)，但是防止內(nèi)存泄露是很簡(jiǎn)單的捞魁。
如果你一直在關(guān)注代碼至会，你可能會(huì)注意到你調(diào)用的 Observable.subscribe() 的返回值是一個(gè) Subscription 對(duì)象。Subscription 類只有兩個(gè)方法谱俭，unsubscribe() 和 isUnsubscribed()奉件。為了防止可能的內(nèi)存泄露，在你的 Activity 或 Fragment 的 onDestroy 里昆著，用 Subscription.isUnsubscribed() 檢查你的 Subscription 是否是 unsubscribed县貌。如果調(diào)用了 Subscription.unsubscribe() ，Unsubscribing將會(huì)對(duì) items 停止通知給你的 Subscriber凑懂，并允許垃圾回收機(jī)制釋放對(duì)象煤痕，防止任何 RxJava 造成內(nèi)存泄露。如果你正在處理多個(gè) Observables 和 Subscribers接谨，所有的 Subscription 對(duì)象可以添加到 CompositeSubscription摆碉，然后可以使用 CompositeSubscription.unsubscribe() 方法在同一時(shí)間進(jìn)行退訂(unsubscribed)。
寫在最后
RxJava 在 Android 生態(tài)系統(tǒng)中提供非常棒的多線程選項(xiàng)脓豪。讓我們能輕松的去后臺(tái)線程做操作巷帝，然后將結(jié)果推到 UI 線程上。這對(duì)于任何 Android 應(yīng)用來說都是非常需要的功能扫夜，能夠運(yùn)用 RxJava 的眾多操作符來處理任何操作的結(jié)果僅僅是為了創(chuàng)造更多的附加值楞泼。然而 RxJava 要求我們對(duì)這個(gè)庫(kù)有更好的了解，充分利用其功能笤闯，所花費(fèi)在這個(gè)庫(kù)上的時(shí)間就能讓你帶來更大的回報(bào)现拒。
這篇博客并未涉及 RxJava 的更進(jìn)一步的主題：熱觀察 vs 冷觀察、處理 backpressure望侈、 Rx 的 Subject 類。用 RxJava 替代 AsyncTask 所涉及的示例代碼可以在 Github 上找到勋桶。
番外：Retrolambda
Java 8 引入了 Lambda 表達(dá)式脱衙，遺憾的是 Android 并不支持 Java 8侥猬，所以我們不能在 RxJava 中利用這一特性。幸運(yùn)的是捐韩，有一個(gè)名為 Retrolambda 的庫(kù)反向移植了 Lambda 表達(dá)式到 Java 的早期版本退唠。還有提供有一個(gè) Retrolambda 的 gradle 插件，能讓我們?cè)?Android 應(yīng)用中去使用 Lambda荤胁。
對(duì)于 Lambda瞧预，可以簡(jiǎn)化 Observable 和 Subscriber 的代碼，如：
final Observable operationObservable = Observable.create(
(Subscriber subscriber) -> {
subscriber.onNext(longRunningOperation());
subscriber.onCompleted();
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
startRxOperationButton = (Button) findViewById(R.id.start_rxjava_operation_btn);
startRxOperationButton.setOnClickListener(v -> {
v.setEnabled(false);
operationObservable.subscribe(
value -> Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show(),
error -> Log.e("TAG", "Error: " + error.getMessage()),
() -> v.setEnabled(true));
});
Lambda 表達(dá)式為 RxJava 減少了很多代碼仅政，我會(huì)強(qiáng)烈建議你用上 Retrolambda 的垢油。它甚至比 RxJava 還要好用（setOnClickListener方法同樣可以使用 Retrolambda）。
