RxJava2.0使用(一)
RxJava剛出來的時候練習過一些，只是沒有用到專案中，如今RxJava 2.x都已經出來很久了，這麼好的一個框架不能瞭解它實在是不能安撫我這顆躁動的心。
Ok 下面先看一下基本用法：
//建立被觀察者
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe 輸出結果：
07-11 13:39:59.370 8816-8816/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: hello
07-11 13:39:59.370 8816-8816/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: Lily
07-11 13:39:59.370 8816-8816/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: onComplete很簡單三步曲 建立被觀察者、建立觀察者、建立聯絡（讓觀察者觀察被觀察者）
舉個小例子：
一句話：甲往傳送帶上放東西，乙觀察著傳送帶。 其中傳送帶為被觀察者（Observable），乙為觀察者（Observer），乙去觀察這個動作為建立聯絡（observable.subscribe(observer)）。
上面的例子中乙（觀察者）可以觀察傳送帶（被觀察者）上的東西。那甲是幹什麼的呢？
（1）傳送帶作為被觀察者在建立的時候其內部有一個ObservableEmitter，這個可以翻譯成發射器，負責傳送事件，所以我們可以把他當成甲，因為傳送帶上的東西都是甲放入進去的。甲有幾個方法：《1》放東西 void onNext(T value); 《2》放錯了東西 void onError(Throwable error); 《3》告訴乙放完東西了 void onComplete(); 這個時候乙就不看了，甲在放東西跟乙也沒關係了。
（2）乙作為觀察者其內部有一個Disposable ，這個可以翻譯成可任意處理的，它有個方法dispose()，當呼叫了之後乙就相當於不觀察傳送帶了，這個時候甲在往傳送帶上放東西跟乙就沒關係了，因為乙現在去幹別的事情了不看著傳送帶了。
onComplete() 和 dispose() 之後乙都不會再收到東西。第一個是甲告訴乙我放完了不放了所以乙不收了，第二個是乙自己不願意收了。
上面的例子中observer 建立的時候帶了好幾個方法，onSubscribe，onNext，onError，onComplete好多啊，其實很多時候我們只需要onNext就可以啦 所以有簡單點的只監聽onNext。
Observable<String> observable1 = Observable.just("哈哈我真簡單");
Consumer<String> consumer = new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.i(TAG,s);
}
};
observable1.subscribe(consumer);輸出結果：
07-11 14:43:07.191 23107-23107/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: 哈哈我真簡單- 傳送帶可以對甲放入的東西進行處理操作，這時候就可以用到RxJava提供的操作符了。
<1>map操作符。可以定義一個方法來改變傳送帶上的資料，比如下面的方法，在每個String後面加一個”—me”
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("hello");
e.onNext("LiLy");
}
}).map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
return s+"---me";
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.i(TAG,s);
}
});輸出結果：
07-11 15:13:54.804 18955-18955/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: hello---me
07-11 15:13:54.804 18955-18955/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: LiLy---me比喻：甲將一個蘋果放到傳送帶上 ， 傳送帶將蘋果削皮，乙得到了削皮後的蘋果。
<2>FlatMap 網上的解釋是 把一個發射器Observable 通過某種方法轉換為多個Observables，然後再把這些分散的Observables裝進一個單一的發射器Observable。很難懂！
我的理解是：對一個事件進行不同的加工之後在發出來。map只能進行一種加工 這個可以進行多種加工。
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("hello");
e.onNext("LiLy");
}
}).flatMap(new Function<String, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(String s) throws Exception {
List<String> list = new ArrayList<String>();
list.add(s+"----1");
list.add(s+"----2");
return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.i(TAG,s);
}
});輸出結果：
07-11 16:27:05.536 22804-22965/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: hello----1
07-11 16:27:05.537 22804-22965/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: hello----2
07-11 16:27:05.538 22804-22965/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: LiLy----1
07-11 16:27:05.538 22804-22965/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: LiLy----2這裡flatMap並不保證事件的順序, ，如果需要保證，需要用到ConcatMap。
比喻：甲將一個蘋果放在了傳送帶上，傳送帶把蘋果切成了兩份，一份雕刻成一個小人，一份直接弄成了蘋果醬。乙得到了一個蘋果小人和一份蘋果醬。
<3>concatMap 跟FlatMap 的用法是一樣的
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("hello");
e.onNext("LiLy");
}
}).concatMap(new Function<String, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(String s) throws Exception {
List<String> list = new ArrayList<String>();
list.add(s + "----1");
list.add(s + "----2");
list.add(s + "----3");
return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.i(TAG, s);
}
});輸出結果：
07-11 16:37:18.143 31879-32113/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: hello----1
07-11 16:37:18.143 31879-32113/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: hello----2
07-11 16:37:18.143 31879-32113/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: hello----3
07-11 16:37:18.161 31879-32114/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: LiLy----1
07-11 16:37:18.162 31879-32114/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: LiLy----2
07-11 16:37:18.162 31879-32114/com.chs.myrxjavaandretrofit I/com.chs.myrxjavaandretrofit: LiLy----3
<4>Concat 把兩個Observable連線成一個Observable
Observable.concat(Observable.just("hello"), Observable.just("LiLy"))
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.i(TAG, "concat : "+ s );
}
});輸出結果：
07-11 16:48:05.404 9721-9721/com.chs.myrxjavaandretrofit E/com.chs.myrxjavaandretrofit: concat : hello
07-11 16:48:05.404 9721-9721/com.chs.myrxjavaandretrofit E/com.chs.myrxjavaandretrofit: concat : LiLy