RxJava2.0使用(一)

RxJava刚出来的时候练习过一些，只是没有用到项目中，如今RxJava 2.x都已经出来很久了，这么好的一个框架不能了解它实在是不能安抚我这颗躁动的心。 Ok 下面先看一下基本用法：

//创建被观察者 Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> e) throws Exception { e.onNext("hello"); e.onNext("Lily"); e.onComplete(); } }); //创建观察者 Observer<String> observer = new Observer<String>() { @Override public void onSubscribe(Disposable d) { } @Override public void onNext(String value) { Log.i(TAG,value); } @Override public void onError(Throwable e) { } @Override public void onComplete() { Log.i(TAG,"onComplete"); } }; //建立联系 observable.subscribe(observer);

输出结果：

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

RxJava中内置了很多操作符，位置在http://reactivex.io/documentation/operators.html#alphabetical 可以看到非常的多。操作符的使用方法跟上面的几个使用差不多，可以参考文档选择自己所需要的。