作为一个android开发者，在开发应用的过程中避免不了异步这个问题。android系统为我们提供了Handler这个类帮助我们进行线程间的通信和切换，但是GitHub上也有很多其他非常优秀的开源框架来帮助我们进行异步处理，比如今天学习的RxAndroid。

简介

GitHub传送门：https://github.com/ReactiveX/RxAndroid
RxAndroid是一套观察者模式、链式调用的异步编程的API，使用RxAndroid编写的程序在逻辑上将会更加的简介、更加的便于理解。

引入框架

首先

在Androidstudio中的build.gradle（project）文件中的dependencies节点下添加代码
classpath ‘me.tatarka:gradle-retrolambda:3.2.5’

buildscript {repositories {google()jcenter()}dependencies {classpath 'com.android.tools.build:gradle:3.5.0'// NOTE: Do not place your application dependencies here; they belong// in the individual module build.gradle filesclasspath 'me.tatarka:gradle-retrolambda:3.2.5'}
}

第二步

在build.gradle（app）文件下的dependencies节点中添加
implementation ‘io.reactivex.rxjava2:rxandroid:2.1.1’
implementation ‘io.reactivex.rxjava2:rxjava:2.2.12’

dependencies {implementation fileTree(dir: 'libs', include: ['*.jar'])implementation 'androidx.appcompat:appcompat:1.0.2'implementation 'androidx.constraintlayout:constraintlayout:1.1.3'testImplementation 'junit:junit:4.12'androidTestImplementation 'androidx.test:runner:1.1.1'androidTestImplementation 'androidx.test.espresso:espresso-core:3.1.1'//rximplementation 'io.reactivex.rxjava2:rxandroid:2.1.1'implementation 'io.reactivex.rxjava2:rxjava:2.2.12'
}

第三步

点击sync now即可
在这里插入图片描述

HelloWord

首先写一个HelloWord，了解一下RxAndroid的简单使用

 public static void helloWord() {//创建被观察者Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {e.onNext("Hello RxAndroid");//发送消息e.onComplete();//完成消息的发送}});//创建观察者Observer observer = new Observer() {@Overridepublic void onSubscribe(@NonNull Disposable d) {Log.e("建立连接", "OnSubscribe");}@Overridepublic void onNext(@NonNull Object o) {//获取被观察者发送过来的消息Log.e("被观察者发送过来的消息", o.toString());}@Overridepublic void onError(@NonNull Throwable e) {Log.e("错误", e.getMessage());}@Overridepublic void onComplete() {Log.e("传输完成", "onComplete");}};//建立连接observable.subscribe(observer);}

在上面的那个方法中，我们可以看到有被观察者：Observable，观察者：Observer，最后通过Observable的subscribe方法订阅连接两者，然后在被观察者中发送一个Hello RxAndroid的字符串，在观察者中的onNext方法中接收到被观察者发送过来的消息。这样一次观察者和被观察者之间的通信就基本完成了。
其中Observer有三个回调方法：onSubscribe、onNext、onError、onComplete，其中onSubscribe是两者在订阅的时候回调，onNext是被观察者调用onNext方法发送消息时的回调，onError是发生异常时的回调，onComplete是被观察者调用onComplete方法时的回调。

-----------------------------------------------说正事专用分割线---------------------------------------------
由以上可以简单的总结一下，RxAndroid有以下三个基本的元素
          1、被观察者(Observable)
          2、观察者（Observer）
          3、订阅（subscribe）

简单使用及常用操作符

我们在HelloWord中在被观察者中向观察者中发送了一个字符串就需要写了这么多，这样是不是太复杂了？不要着急，RxAndroid还为我们提供了一些其他的操作符。

just

通过just操作符创建被观察者，当有一个参数，观察者就会收到一次，两个参数收到两次，以此类推。

 public static void just() {//创建被观察者Observable<String> observable = Observable.just("你好", "222");//创建观察者Consumer<String> consumer = new Consumer<String>() {@Overridepublic void accept(String s) throws Exception {Log.e("通过just操作符创建被观察者", s);}};//建立连接observable.subscribe(consumer);}

有没有发现，通过just创建的被观察者没有onError这个回调，那我应该怎么处理异常呢，这个时候就需要我们手动去抛出异常了，例如;

 public static void justEx() {//创建被观察者try {Observable<Integer> observable = Observable.just(Integer.parseInt("M"));//创建观察者Consumer<Integer> consumer = new Consumer<Integer>() {@Overridepublic void accept(Integer s) throws Exception {Log.e("just操作符", s + "");}};//建立连接observable.subscribe(consumer);} catch (Exception e) {Log.e("justEx", e.getMessage());}}

我在被观察者中发送了一个字符"M"，并且将它转换成int类型，因此就会报错，我们可以在catch中打印错误发现： E/justEx: Invalid int: “M”
在这里插入图片描述
我们也可以根据Observer中的三个状态创建回调，通过action代替onComplete

 public static void action() {//创建被观察者Observable<String> observable = Observable.just("你好");//创建观察者//接收onNext发送过来的正常的消息Consumer<String> nextConsumer = new Consumer<String>() {@Overridepublic void accept(String s) throws Exception {Log.e("onNext", s + "");}};//接收异常消息Consumer<Throwable> onErrorConsumer = new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {Log.e("异常", "" + throwable.getMessage());}};//创建actionAction completeAction = new Action() {@Overridepublic void run() throws Exception {Log.e("actiong", "任务完成");}};//建立连接observable.subscribe(nextConsumer, onErrorConsumer, completeAction);}

打印log可以看到，依次调用了接收消息、消息完成
在这里插入图片描述

map

变化操作符，对被观察者发送的每一个事件应用一个函数，通过这个函数产生变化

 public static void map() {Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {e.onNext("3");}}).map(new Function<String, Integer>() {@Overridepublic Integer apply(@NonNull String s) throws Exception {return Integer.parseInt(s);}}).subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Exception {Log.e("返回值", "" + integer);}});}

zip

将多个被观察者发送的事件结合到一起，然后再发送，并且发送事件的数量和上游事件最少的相同，比如：被观察者A发送了两个事件，被观察者B发送了一个事件，那么观察者只能接受一个事件，被观察者的组合也是严格按照顺序来的

public static void zip() {//创建被观察者Observable<String> obs1 = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {e.onNext("我是1");e.onNext("我是2");e.onComplete();}});Observable<String> obs2 = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {e.onNext("我是A");e.onComplete();}});//创建观察者Consumer<String> consumer = new Consumer<String>() {@Overridepublic void accept(String o) throws Exception {Log.e("返回值", o);}};//通过zip打包Observable.zip(obs1, obs2, new BiFunction<String, String, String>() {@Overridepublic String apply(@NonNull String s, @NonNull String s2) throws Exception {return s + s2;}}).subscribe(consumer);}

merge

将多个被观察者合并成一个，这里和zip不同，zip是和具有最少的发送事件的被观察者的数量相同

public static void merge() {//创建被观察者Observable<String> obs1 = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {e.onNext("我是1");e.onNext("我是2");e.onComplete();}});Observable<String> obs2 = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {e.onNext("我是A");e.onComplete();}});//创建观察者Consumer<String> consumer = new Consumer<String>() {@Overridepublic void accept(String o) throws Exception {Log.e("返回值", o);}};//通过merge打包发送Observable.merge(obs1, obs2).subscribe(consumer);}

filter

操作符用于去除不符合条件的事件。例如下面例子中去掉数组中的空字符串

 public static void filter() {Observable.fromArray(new String[]{"123456789", "", "123456789", "1111"}).filter(new Predicate<String>() {@Overridepublic boolean test(@NonNull String s) throws Exception {//返回false表示去除该事件return s.length() != 0;}}).subscribe(new Consumer<String>() {@Overridepublic void accept(String s) throws Exception {Log.e("接受值", "111" + s);}});}

线程调度

如果说rxandroid只有上面那些功能是远远不会这么火热的，在android中线程间的调度是少不了的。
看一个简单例子

  public static void threadTest() {//创建被观察者Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {Log.e("被观察者", "当前线程：" + Thread.currentThread().getName());e.onNext("测试");e.onError(new Exception("111"));e.onComplete();}});//创建观察者Consumer<String> consumer = new Consumer<String>() {@Overridepublic void accept(String s) throws Exception {Log.e("观察者", "当前线程：" + Thread.currentThread().getName());//  Log.e("消息",""+s);}};//异常Consumer<Throwable> error = new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {Log.e("异常", "当前线程：" + Thread.currentThread().getName());}};//任务完成Action action = new Action() {@Overridepublic void run() throws Exception {Log.e("Action", "当前线程：" + Thread.currentThread().getName());}};//建立连接observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(consumer, error, action);}

首先创建一个被观察者、观察者，然后在订阅的时候进行指定观察者和被观察者各自的运行线程即可，通过subscribeOn方法指定被观察者运行的线程，Schedulers.io()代表开启一个新线程，一般用于进行访问网络的操作，通过observeOn指定观察者的线程，AndroidSchedulers.mainThread()代表运行的线程在主线程及UI线程，在该线程中可以进行更新UI的操作。

异常处理

如果被观察者中运行的方法抛出异常，观察者则会回调onError方法，并且被观察者中的代码不会继续执行

 public static void onError1() {//创建被观察者Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {e.onNext(2);int i = Integer.parseInt("l");e.onNext(i);int w = Integer.parseInt("e");e.onNext(w);e.onNext(3);Log.e("Observable", "我走到这里啦");e.onComplete();}});//创建观察者Observer<Integer> observer = new Observer<Integer>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {Log.e("建立连接", "onSubscribe");}@Overridepublic void onNext(@NonNull Integer integer) {Log.e("获取返回值", integer + "");}@Overridepublic void onError(@NonNull Throwable e) {Log.e("异常", e.getMessage());}@Overridepublic void onComplete() {Log.e("传输完成", "onComplete");}};//建立连接observable.subscribe(observer);}

被观察中只能调用一次onError方法，如果多次调用，将在第二次调用时抛出异常，并且一旦调用了onError方法，将不能调用onComplete方法报完成

 public static void onError2() {//创建被观察者Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {e.onError(new Exception("错误1"));e.onNext("111");e.onError(new Exception("错误2"));e.onComplete();}});//创建观察者Observer<String> observer = new Observer<String>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {Log.e("建立连接", "onSubscribe");}@Overridepublic void onNext(@NonNull String integer) {Log.e("获取返回值", integer + "");}@Overridepublic void onError(@NonNull Throwable e) {Log.e("异常", e.getMessage());}@Overridepublic void onComplete() {Log.e("传输完成", "onComplete");}};//建立连接observable.subscribe(observer);}

参考链接：
https://www.jianshu.com/p/7eb5ccf5ab1e
https://juejin.im/post/5b17560e6fb9a01e2862246f
https://www.jianshu.com/p/c66951952c74