前言
第一次接觸學(xué)習(xí)RxJava應(yīng)該是一兩個月前的事情了社裆，但其中也是斷斷續(xù)續(xù)翁授，最近又再次去學(xué)習(xí)RxJava残拐，和當(dāng)初剛接觸RxJava完全不是同樣的心情涎拉，輕松了很多瑞侮，也感受到了RxJava的魅力，真是不由衷感嘆太牛了鼓拧。目前關(guān)于RxJava的文章也很多半火，個人推薦兩篇扔物線的給 Android 開發(fā)者的 RxJava 詳解和大頭鬼Bruce的譯文 深入淺出RxJava系列。那么這篇文章通過代碼介紹RxJava中的操作符季俩，以及操作符的使用钮糖。當(dāng)然操作符較多，準(zhǔn)備分幾篇文章介紹酌住。如果你想提前學(xué)習(xí)其他操作符可以去GitHub,歡迎star項目店归。
RxJava操作符源碼傳送門
基礎(chǔ)介紹
為了力求有沒有RxJava基礎(chǔ)都能看懂此文，簡單介紹一下RxJava以及一些名詞酪我。在RxJava開源的Github上是這樣解釋的a library for composing asynchronous and event-based programs using observable sequences for the Java VM消痛。無論多么復(fù)雜的邏輯，都可以保持整潔的代碼格式都哭。
在RxJava中最重要的就是Observable（被觀察者）秩伞，subscribe（訂閱），Observer（觀察者）或者Subscriber（訂閱者）欺矫，Observable也就是數(shù)據(jù)（事件）源纱新，Subscriber負責(zé)接收以及處理數(shù)據(jù)（事件）。當(dāng)然要想實現(xiàn)兩者通信穆趴，需要有一種機制那就是訂閱脸爱。Observer 通過 subscribe() 方法實現(xiàn)訂閱關(guān)系，從而 Observable 可以在需要的時候發(fā)出事件來通知 Observer未妹。
例如張三（觀察者）想看某款新聞軟件的科技信息（被觀察者）阅羹，由于科技信息是每天推送或者不定時推送勺疼，如果張三一直盯著手機屏幕看并且刷新消息是不是又新的信息，顯然不現(xiàn)實捏鱼。這時候就可以通過張三 subscribe（訂閱）科技信息执庐，而實現(xiàn)當(dāng)有新的科技信息時自動給張三推送消息，在這期間导梆，張三并不需要一直盯著屏幕刷新聞轨淌。在我們平時的認知中實現(xiàn)訂閱應(yīng)該是張三.subscribe(科技新聞)，不過在RxJava代碼中實現(xiàn)訂閱應(yīng)該寫成科技新聞.subscribe(張三)看尼。
在RxJava中递鹉，有三個事件回調(diào)方法，分別是onNext(),OnError(),onCompleted(),onNext()是最終輸出及處理數(shù)據(jù)的回調(diào)藏斩，在發(fā)射數(shù)據(jù)過程中出現(xiàn)錯誤異常會回調(diào)OnError()方法躏结，當(dāng)不會再有新的 onNext() 發(fā)出時，需要觸發(fā) onCompleted() 方法作為標(biāo)志狰域。,OnError()和onCompleted()是互斥的媳拴。下面舉一個最簡單的例子
Observable observable2 = Observable.just("也許當(dāng)初忙著微笑和哭泣", "忙著追逐天空中的流星", "人理所當(dāng)然的忘記", "是誰風(fēng)里雨里一直默默守護在原地");
Subscriber subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: " )
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ")
}
@Override
public void onNext(String s) {
Log.e(TAG, "onNext: "+s )
}
};
observable.subscribe(subscriber);
運行后打印信息為
onNext: 也許當(dāng)初忙著微笑和哭泣
onNext: 忙著追逐天空中的流星
onNext: 人理所當(dāng)然的忘記
onNext: 是誰風(fēng)里雨里一直默默守護在原地
onCompleted:
Create
我們可以使用該操作符從零開始創(chuàng)建一個Observable，給這個操作符傳遞一個接受觀察者作為參數(shù)的函數(shù)兆览，并調(diào)用觀察者的onNext屈溉，onError和onCompleted方法。如下
//被觀察者
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
//可以多次調(diào)用subscriber.onNext("大家好")發(fā)射數(shù)據(jù)
subscriber.onNext("大家好");
subscriber.onNext("我開始學(xué)習(xí)RxJava");
subscriber.onCompleted();
}
});
發(fā)送數(shù)據(jù)需要在毀掉方法call中調(diào)用subscriber的onNext(),onNext(T)發(fā)送的參數(shù)需要和Observable.OnSubscribe<T>()中參數(shù)相同抬探，在上面我們傳入的是String類型子巾。創(chuàng)建后Observale后，我們需要創(chuàng)建Subscriber(觀察者)去處理observable發(fā)送的數(shù)據(jù)小压。如下
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, e.getMessage());
}
@Override
public void onNext(String s) {
Log.e(TAG, "onNext："+s);
}
};
數(shù)據(jù)成功發(fā)送后线梗，會回調(diào)Subscriber的onNext()的方法，其中的參數(shù)就是接收到的數(shù)據(jù)怠益。當(dāng)onNext（）接收數(shù)據(jù)完畢后會執(zhí)行onCompleted（）,如果中途有環(huán)節(jié)出現(xiàn)錯誤異常仪搔，會執(zhí)行onError()。現(xiàn)在觀察者和被觀察者都創(chuàng)建完畢了溉痢，他們執(zhí)行還需要一個前提就是訂閱，如果不訂閱憋他，observable并不會發(fā)射數(shù)據(jù)孩饼，subscribe也不會接收數(shù)據(jù)，訂閱代碼如下
observable.subscribe(subscriber);
執(zhí)行后輸出信息
onNext：大家好
onNext：我開始學(xué)習(xí)RxJava
onCompleted
from
該操作符是將其它種類的對象和數(shù)據(jù)類型轉(zhuǎn)換為Observable竹挡，如果當(dāng)你發(fā)射的的數(shù)據(jù)是同一種類型镀娶，而不是混合使用Observables和其它類型的數(shù)據(jù)，會非常方便揪罕。如下創(chuàng)建Observable
Integer[] integers = {1,2, 3, 4};
Observable<Integer> observable=Observable.from(integers);
Subscriber<String> subscriber = new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, e.getMessage());
}
@Override
public void onNext(Integer i) {
Log.e(TAG, "onNext："+i);
}
};
observable.subscribe(subscriber);
輸出信息為
onNext：1
onNext：2
onNext：3
onNext：4
onCompleted
from操作符可以轉(zhuǎn)換Future梯码、Iterable和數(shù)組宝泵。對于Iterable和數(shù)組，產(chǎn)生的Observable會發(fā)射Iterable或數(shù)組的每一項數(shù)據(jù)轩娶。對于Future，它會發(fā)射Future.get()方法返回的單個數(shù)據(jù)鳄抒，并且還可以增加通過: from(Future,timeout, timeUnit)指定超時時間许溅，如果執(zhí)行的時候Future超時會回調(diào)onError（）方法。
just
just將單個數(shù)據(jù)轉(zhuǎn)換為發(fā)射那個數(shù)據(jù)的Observable茬祷，Just類似于From祭犯，但是From會將數(shù)組或Iterable的數(shù)據(jù)取出然后逐個發(fā)射借卧，而Just只是簡單的原樣發(fā)射铐刘，將數(shù)組或Iterable當(dāng)做單個數(shù)據(jù)，如果你傳遞null給Just檩禾，它會返回一個發(fā)射null值的Observable盼产。不要誤認為它會返回一個空Observable（完全不發(fā)射任何數(shù)據(jù)的Observable）勺馆。對于just可以接收1到10個數(shù)據(jù)，返回一個按參數(shù)列表順序發(fā)射這些數(shù)據(jù)的Observable灌灾。
Observable.just(1 2, 3, 4)
.subscribe(new Subscriber<Integer>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: ");
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext: " + integer);
}
});
輸出
onNext:1
onNext:2
onNext:3
onNext:4
onCompleted:
對于just參數(shù)類型可以是多種锋喜，如下嘿般，傳入兩個類型數(shù)據(jù)
Observable.just(0, "one", 6, "two", 8, "three")
.subscribe(new Subscriber<Serializable>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: " );
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ");
}
@Override
public void onNext(Serializable serializable) {
Log.e(TAG, "onNext: "+serializable.toString());
}
});
則輸出信息
onNext:0
onNext:one
onNext:6
onNext:two
onNext:8
onNext:three
onCompleted:
Empty/Never/Error
Empty:創(chuàng)建一個不發(fā)射任何數(shù)據(jù)但是正常終止的Observable炉奴，此時會回調(diào)onCompleted（）
Never:創(chuàng)建一個不發(fā)射數(shù)據(jù)也不終止的Observable
Error:創(chuàng)建一個不發(fā)射數(shù)據(jù)以一個錯誤終止的Observable
error操作符需要一個Throwable參數(shù)盆佣，你的Observable會以此終止共耍。這些操作符默認不在任何特定的調(diào)度器上執(zhí)行，但是empty和error有一個可選參數(shù)是Scheduler穆咐，如果你傳遞了Scheduler參數(shù)对湃，它們會在你指定的調(diào)度器上發(fā)送通知拍柒。
Range
該操作符創(chuàng)建特定整數(shù)序列的Observable屈暗，它接受兩個參數(shù)，一個是范圍的起始值种呐，一個是范圍的數(shù)據(jù)的數(shù)目弃甥。如果你將第二個參數(shù)設(shè)為0淆攻，將導(dǎo)致Observable不發(fā)射任何數(shù)據(jù)（如果設(shè)置為負數(shù)，會拋異常）啸箫。
Observable.range(1,4)
.subscribe(new Subscriber<Integer>() {
public String TAG="RXJAVA";
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: " );
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ");
}
@Override
public void onNext(Integer integer) {
Log.e(TAG, "onNext: "+integer);
}
});
輸出信息
onNext: 1
onNext: 2
onNext: 3
onNext: 4
onCompleted:
你可以在代碼實戰(zhàn)中筐高，更改第二個參數(shù)為負數(shù)柑土，或者0绊汹，以及將第一個參數(shù)更改為你想測試的任意值西乖，去觀察執(zhí)行日志幫助理解。
Timer
Timer操作符創(chuàng)建一個在給定的時間段之后返回一個特殊值的Observable薄腻。它在延遲一段給定的時間后發(fā)射一個簡單的數(shù)字0 庵楷。
Observable.timer(1, TimeUnit.SECONDS)
.subscribe(new Subscriber<Long>() {
public String TAG="RXJAVA";
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: " );
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ");
}
@Override
public void onNext(Long integer) {
Log.e(TAG, "onNext:1111111 "+integer);
}
});
對于該操作符默認在computation調(diào)度器上執(zhí)行的尽纽，如果你想在onNext（）回調(diào)方法更新UI,需要通過observeOn(AndroidSchedulers.mainThread())設(shè)置童漩，否則會調(diào)用onError()方法矫膨。當(dāng)然Time人提供的有一個三個參數(shù)的方法timer(long,TimeUnit,Scheduler)可以指定 Scheduler 豆拨。
Interval
該操作符按固定的時間間隔發(fā)射一個無限遞增的整數(shù)序列，它接受一個表示時間間隔的參數(shù)和一個表示時間單位的參數(shù)脚线，當(dāng)然該操作符合Timer一樣弥搞，是在computation調(diào)度器上執(zhí)行的攀例，若想更新UI需要指定Scheduler 為AndroidSchedulers.mainThread()挖胃。
Subscription subscription = Observable.interval(1, TimeUnit.SECONDS)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Long>() {
@Override
public void call(Long aLong) {
tv.append(" " + aLong + " ");
}
});
通過上面代碼就會每隔1秒在tv上追加一個數(shù)字，并且會永遠執(zhí)行烁登。如果在某個時刻不想繼續(xù)輸出，就需要要解除訂閱蔚舀。
if (subscription != null && !subscription.isUnsubscribed()) {
subscription.unsubscribe();
}
Repeat
該操作符是重復(fù)的發(fā)射某個數(shù)據(jù)序列饵沧，并且可以自己設(shè)置重復(fù)的次數(shù)。當(dāng)接收到onComplete()會觸發(fā)重訂閱再次重復(fù)發(fā)射數(shù)據(jù),當(dāng)重復(fù)發(fā)射數(shù)據(jù)次數(shù)到達后執(zhí)行onCompleted()赌躺。
String[] strs = {"也許當(dāng)初忙著微笑和哭泣", "忙著追逐天空中的流星"};
Observable.from(strs).repeat(2)..subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: " );
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ");
}
@Override
public void onNext(String s) {
Log.e(TAG, "onNext: "+s );
tv1.append("\n" + s);
}
});
輸出
onNext: 也許當(dāng)初忙著微笑和哭泣
onNext: 忙著追逐天空中的流星
onNext: 也許當(dāng)初忙著微笑和哭泣
onNext: 忙著追逐天空中的流星
onCompleted:
Defer
直到有觀察者訂閱時才創(chuàng)建Observable狼牺，并且為每個觀察者創(chuàng)建一個新的Observable，該操作符能保證訂閱執(zhí)行時數(shù)據(jù)源是最新的數(shù)據(jù)礼患。如下正常代碼
String test="舊數(shù)據(jù)";
Observable observable=Observable.just(test);
Subscriber subscriber=new Subscriber() {
public String TAG="RXJAVA";
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: " );
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: " );
}
@Override
public void onNext(Object o) {
Log.e(TAG, "onNext: "+o );
}
};
test="新數(shù)據(jù)";
observable.subscribe(subscriber);
輸出
onNext: 舊數(shù)據(jù)
onCompleted:
通過上面代碼和輸出日志發(fā)現(xiàn)锁右，雖然在后面講數(shù)據(jù)test更新為新數(shù)據(jù)，但是并沒有生效讶泰，要想使用最新的數(shù)據(jù)就需要使用defer操作符咏瑟。此時更改使用defer
test="舊數(shù)據(jù)";
Observable<String> observable=Observable.defer(new Func0<Observable<String>>() {
@Override
public Observable<String> call() {
return Observable.just(test);
}
});
Subscriber subscriber=new Subscriber() {
public String TAG="RXJAVA";
@Override
public void onCompleted() {
Log.e(TAG, "onCompleted: " );
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: " );
}
@Override
public void onNext(Object o) {
Log.e(TAG, "onNext: "+o );
}
};
test="新數(shù)據(jù)";
observable.subscribe(subscriber);
輸出信息
onNext: 新數(shù)據(jù)
onCompleted:
通過新的打印信息，發(fā)現(xiàn)輸出值已經(jīng)是最新的數(shù)據(jù)痪署。
到這里码泞，這篇文章暫時就先結(jié)束了，若文章有不足或者錯誤的地方狼犯，歡迎指正余寥，以防止給其他讀者錯誤引導(dǎo)。更多的操作符悯森，將在接下來的幾篇文章介紹宋舷。
