Android 响应式编程 RxJava2 完全解析

Posted 2021-04-30 穿背心儿的程序猿

使用了 RxJava2 有一段时间了，深深感受到了其“牛逼”之处。下面，就从 RxJava2 的基础开始，一步步与大家分享一下这个强大的异步库的用法！RxJava 是 一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库，也就是用于实现异步操作的库。

一、RxJava2 基础

RxJava可以浓缩为异步两个字，其核心的东西不外乎两个， Observables（被观察者） 和 Observable（观察者）。Observables可以发出一系列的 事件（例如网络请求、复杂计算、数据库操作、文件读取等），事件执行结束后交给Observable 的回调处理。

1.RxJava2 的观察者模式

观察者模式是对象的行为模式，也叫做发布-订阅(Publish/Subscribe)模式、模型-视图(Model/View)模式、源-监听器(Source/Listener)模式或从属者(Dependents)模式。

什么是观察者模式？举个栗子，android中View的点击监听器的实现，View是被观察者，OnClickListener对象是观察者，Activity要如何知道View被点击了？那就是派一个OnClickListener对象，入驻View，与View达成一个订阅关系，一旦View被点击了，就通过OnClickListener对象的OnClick方法传达给Activity。采用观察者模式可以避免去轮询检查，节约有限的cpu资源。

RxJava 作为一个工具库，使用的便是通用形式的观察者模式：

普通事件：onNext()，相当于 onClick()、onEvent()；特殊事件：onCompleted() 和 onError()

如图所示，RxJava 的基本概念分别为：Observable(被观察者，事件源)，Observer(观察者，订阅者)，subscribe (订阅)、事件；不同的是，RxJava 把多个事件看做一个队列，并对每个事件单独处理。在一个队列中 onCompleted() 和 onError()，只有一个会被调用。如果调用了 onCompleted() 就说明队列执行完毕，没有出现异常，否则调用 onError() 方法并终止队列。

2.RxJava2 响应式编程结构

什么是响应式编程？举个栗子，a = b + c; 这句代码将b+c的值赋给a，而之后如果b和c的值改变了不会影响到a，然而，对于响应式编程，之后b和c的值的改变也动态影响着a，意味着a会随着b和c的变化而变化。

响应式编程的组成为Observable/Operator/Subscriber，RxJava在响应式编程中的基本流程如下：

这个流程，可以简单的理解为：Observable -> Operator1 -> Operator2 -> Operator3 -> Subscriber

1. Observable发出一系列事件，他是事件的产生者；

2. Subscriber负责处理事件，他是事件的消费者；

3. Operator是对Observable发出的事件进行修改和变换；

4. 若事件从产生到消费不需要其他处理，则可以省略掉中间的Operator，从而流程变为Obsevable -> Subscriber；

5. Subscriber通常在主线程执行，所以原则上不要去处理太多的事务，而这些复杂的处理则交给Operator；

3.创建一个完整的 RxJava2 调用

首先需要添加 RxJava2 在 Android 中的 Gradle 依赖：

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile "io.reactivex.rxjava2:rxjava:2.0.8"

RxJava2 可以通过下面这几种方法创建被观察者：

// 发送对应的方法
Observable.create(new ObservableOnSubscribe<String>() {
    // 默认在主线程里执行该方法
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
        e.onNext("Hello");
        e.onNext("World");      
        // 结束标识
        e.onComplete();
    }
});
// 发送多个数据
Observable.just("Hello", "World");
// 发送数组
Observable.fromArray("Hello", "World");
// 发送一个数据
Observable.fromCallable(new Callable<String>() {
    @Override
    public String call() throws Exception {
        return "Hello";
    }
});

RxJava2 支持链式编程，下来我们创建被观察者，然后创建观察者并订阅：

// 创建被观察者
Observable.just("Hello", "World")
// 将被观察者切换到子线程
.subscribeOn(Schedulers.io())
// 将观察者切换到主线程
.observeOn(AndroidSchedulers.mainThread())
// 创建观察者并订阅
.subscribe(new Observer<String>() {
    // Disposable 相当于RxJava1.x中的 Subscription，用于解除订阅
    private Disposable disposable;
    @Override
    public void onSubscribe(Disposable d) {
        disposable = d;
    }
    @Override
    public void onNext(String s) {
        Log.i("JAVA", "被观察者向观察者发送的数据:" + s);
       if (s == "-1") {   // "-1" 时为异常数据，解除订阅
            disposable.dispose();
        }
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

一旦 Observer 订阅了 Observable，Observable 就会调用 Observer 的 onNext()、onCompleted()、onError() 等方法。至此一个完整的 RxJava 调用就完成了。看一下输出的Log：

I/JAVA: 被观察者向观察者发送的数据:Hello
I/JAVA: 被观察者向观察者发送的数据:World

若喜欢简洁、定制服务，那么可以实现的方法跟上面的实现方法是对应起来的，大家看参数就知道哪个对应哪个了，你可以通过new Consumer（不需要实现的方法你可以不写，看上去更简洁），Consumer就是消费者的意思，可以理解为消费了 onNext 等事件：

Observable.just("Hello", "World")
.subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.i("JAVA", "被观察者向观察者发送的数据:" + s);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(@NonNull Throwable throwable) throws Exception {
    }
}, new Action() {
    @Override
    public void run() throws Exception {
    }
}, new Consumer<Disposable>() {
    @Override
    public void accept(@NonNull Disposable disposable) throws Exception {
    }
});

4.RxJava2 的操作符

RxJava中提供了大量不同种类，不同场景的Operators(操作符)，RxJava的强大性就来自于它所定义的操作符。主要分类：

其中有一些高频使用的操作符如下：

5.RxJava2 线程调度器

调度器 Scheduler 用于控制操作符和被观察者事件所执行的线程，不同的调度器对应不同的线程。RxJava提供了5种调度器：

可以使用 subscribeOn() 和 ObserveOn() 操作符进行线程调度，让 Observable 在一个特定的调度器上执行。subscribeOn() 指定 subscribe() 所发生的线程，事件产生的线程。ObserveOn() 指定 Observer 所运行在的线程，事件消费的线程。

6.RxJava2 模拟发送验证码倒计时功能

public void onCodeClick() {
    final long count = 60; // 设置60秒
    Observable.interval(0, 1, TimeUnit.SECONDS)
            .take(count + 1)
            .map(new Function<Long, Long>() {
                @Override
                public Long apply(@NonNull Long aLong) throws Exception {
                    return count - aLong; // 由于是倒计时，需要将倒计时的数字反过来
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(@NonNull Disposable disposable) throws Exception {
                    button.setEnabled(false);
                    button.setTextColor(Color.GRAY);
                }
            })
            .subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(Disposable d) {
                }
                @Override
                public void onNext(Long aLong) {
                    button.setText(aLong + "秒后重发");
                }
                @Override
                public void onError(Throwable e) {
                }
                @Override
                public void onComplete() {
                    button.setEnabled(true);
                    button.setTextColor(Color.RED);
                    button.setText("发送验证码");
                }
            });
}

二、RxJava2 系列框架

三、RxJava2 与 Retrofit 的使用

RxJava 与 Retrofit 的使用，更像我们的 AsyncTask，通过网络获取数据然后通过 Handler 更新UI。首先需要导入依赖：

compile 'com.squareup.retrofit2:retrofit:2.2.0'
compile 'com.squareup.retrofit2:converter-gson:2.2.0'
compile 'com.squareup.retrofit2:adapter-rxjava2:2.2.0'

1.模拟用户登陆获取用户数据

模拟用户登陆获取用户数据

1.Bean对象：

public class UserParam {
    private String param1;
    private String param2;
    public UserParam(String param1, String param2) {
        this.param1 = param1;        
        this.param2 = param2;
    }    
    // 省略了 getter setter
}
public class NetBean {
    private FormBean form;    
    // 省略了 getter setter
    public static class FormBean {
        private String username;
        private String password;        
        // 省略了 getter setter
    }
}
public class UserBean {
    private String username;
    private String password;
    public UserBean(String username, String password) {
        this.username = username;        
        this.password = password;
    }    
    // 省略了 getter setter
}

2.ApiService，这里返回Observable对象，也就是我们RxJava的被观察者

public interface ApiService {
    @FormUrlEncoded
    @POST("/post")
    Observable<NetBean> getUserInfo(@Field("username")String username,
                                    @Field("password")String password);
}

3.RxJava + Retrofit 的实现

// 构建Retrofit
ApiService apiService = new Retrofit.Builder()
        .baseUrl("http://httpbin.org/")
        .addConverterFactory(GsonConverterFactory.create()) // RxJava2与Gson混用
        .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // RxJava2与Retrofit混用
        .build()
        .create(ApiService.class);

// 构建RxJava
UserParam param = new UserParam("zhangsan", "123");
// 发送param参数
Observable.just(param)
        // flatMap方法是用于数据格式转换的方法，参数一表示原数据，
        // 参数二表示转换的数据，那么就是通过发送网络参数，转换成网络返回的数据，调用Retrofit
        .flatMap(new Function<UserParam, ObservableSource<NetBean>>() {
            @Override
            public ObservableSource<NetBean> apply(@NonNull UserParam userParam)
                    throws Exception {       
                 // 1.发送网络请求，获取NetBean
                return apiService.getUserInfo(userParam.getParam1(), userParam.getParam2());
            }
        })
        .flatMap(new Function<NetBean, ObservableSource<UserBean>>() {
            @Override
            public ObservableSource<UserBean> apply(@NonNull NetBean netBean)
                    throws Exception {
                UserBean user = new UserBean(netBean.getForm().getUsername(),
                        netBean.getForm().getPassword());
                // 2.转换NetBean数据为我们需要的UserBean数据
                return Observable.just(user);
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<UserBean>() {
            @Override
            public void accept(@NonNull UserBean userBean) throws Exception {
                Log.i("JAVA", "" + "用户名:" + userBean.getUsername()
                        + ", 密码:" + userBean.getPassword());
            }
        });

2.模拟合并本地与服务器购物车列表

这个案例其实就是用户添加购物车的时候，首先会在本地存储一份，然后发现如果没有网络，那么没办法提交到服务器上，只能等下一次有网络的时候采用本地数据库和服务器数据的合并来实现上传到服务器。

模拟合并本地与服务器购物车列表

首先需要准备 Retrofit 对象和获取本地数据、网络数据的方法：

private ApiService apiService;
@Override
protected void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    // 省略
    // 构建Retrofit
    apiService = new Retrofit.Builder()
            .baseUrl("http://httpbin.org/")
            .addConverterFactory(GsonConverterFactory.create()) // RxJava2与Gson混用
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // RxJava2与Retrofit混用
            .build()
            .create(ApiService.class);
}
/** * 获取本地数据 */
private Observable<List<String>> getDataForLocal() {
    List<String> list = new ArrayList<>();
    list.add("购物车的商品1");
    list.add("购物车的商品2");
    return Observable.just(list);
}
/** * 获取网络数据 */
private Observable<List<String>> getDataForNet() {
    return Observable.just("shopName")     
        // flatMap方法是用于数据格式转换的方法，参数一表示原数据，
        // 参数二表示转换的数据，那么就是通过发送网络参数，转换成网络返回的数据，调用Retrofit
        .flatMap(new Function<String, ObservableSource<NetBean>>() {
            @Override
            public ObservableSource<NetBean> apply(@NonNull String s) throws Exception {
                // 1.发送网络请求，获取数据
                return apiService.getCartList(s);
            }
        }).flatMap(new Function<NetBean, ObservableSource<List<String>>>() {
            @Override
            public ObservableSource<List<String>> apply(@NonNull NetBean netBean) throws Exception {
                // String shop = netBean.get_$Args257().getShopName();
                String shop = "购物车的商品3";
                List<String> list = new ArrayList<>();
                list.add(shop);                
                // 2.转换NetBean数据为我们需要的List<String>数据
                return Observable.just(list);
            }
        }).subscribeOn(Schedulers.io());
}

然后我们就可以创建被观察者并订阅了，来完成合并本地与服务器购物车列表操作：

// merge操作符: 将两个ObservableSource合并为一个ObservableSource
Observable.merge(getDataForLocal(), getDataForNet())
        .subscribe(new Observer<List<String>>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(List<String> strings) {                for (String str: strings) { Log.i("JAVA", str); }
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
                Log.i("JAVA", "onComplete");
            }
        });

最后的打印结果是：

I/JAVA: 购物车的商品1
I/JAVA: 购物车的商品2
I/JAVA: 购物车的商品3
I/JAVA: onComplete

四、RxJava2 与 RxBinding 的使用

1.优化搜索请求

当我们在 EditText 打字时搜索的时候，可能用户会打字很会快，那么我们就没有必要一直发送网络请求，请求搜索结果，我们可以通过当用户打字停止后的延时500毫秒再发送搜索请求：

// RxTextView.textChanges(edittext): Rxbinding用法
RxTextView.textChanges(editText)
        // 表示延时多少秒后执行，当你敲完字之后停下来的半秒就会执行下面语句
        .debounce(500, TimeUnit.MILLISECONDS)        
        // 数据转换 flatMap: 当同时多个数据请求访问的时候，前面的网络数据会覆盖后面的网络数据
        // 数据转换 switchMap: 当同时多个网络请求访问的时候，会以最后一个发送请求为准，前面网络数据会被最后一个覆盖
        .switchMap(new Function<CharSequence, ObservableSource<List<String>>>() {
            @Override
            public ObservableSource<List<String>> apply(
                    @NonNull CharSequence charSequence) throws Exception {
                // 网络请求操作，获取我们需要的数据
                List<String> list = new ArrayList<String>();
                list.add("2017");
                list.add("2018");
                return Observable.just(list);
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<List<String>>() {
            @Override
            public void accept(@NonNull List<String> strings) throws Exception {
                // 更新UI
                Log.i("JAVA", strings.toString());
            }
        });

2.优化点击请求

当用户一直点击一个按钮的时候，我们不应该一直调用访问网络请求，而是 1秒内，只执行一次网络请求。

RxView.clicks(button).throttleFirst(1, TimeUnit.SECONDS)
        .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(Object o) {
                Log.i("JAVA", "onClick");
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });

五、RxJava2 踩过的一些坑

1.未解除订阅而引起的内存泄漏

举个例子，对于前面常用操作符 interval 做周期性操作的例子，并没有使之停下来的，没有去控制订阅的生命周期，这样，就有可能引发内存泄漏。所以，在 Activity 的 onDestroy() 方法执行的时候或者不需要继续执行的时候应该解除订阅。

更多 RxJava 示例代码请访问：