谈谈对于响应式编程RxJava的理解 - 核心思想篇
谈谈对于响应式编程RxJava的理解 - 原理篇
对于RxJava，大家应该需要很好的理解其核心思想，或者说你应该知道为什么要使用RxJava？使用RxJava的好处是什么？RxJava的使用场景是什么？更简单的说，你不知道响应式编程是什么？本文将针对这几点，谈谈自己对于RxJava的理解。
对于Rx响应式编程的理解：
有一个起点（Observable）、一个终点（Observer），事件从起点开始传递，无中断的流向终点，在传递的过程中，可以对事件进行拦截（拦截可以改变事件的返回值），但终点只关心它的上一个拦截。
这也是我们平时说的所谓的Rx“响应式编程”的体现。而链式调用（流式调用）只是RxJava里面的一个点而已。使用RxJava我们可以更优雅的书写代码、书写代码的整个思路也会变的非常的"流畅".
响应式编程框架除了RxJava，还有很多，例如RxJs、Rx.net、RxSwift，具体可以看。ReactiveX官网
举个例子
比如，现在我们点击一个按钮，从网络上下载一张图片，显示到ImageView里，
- 不使用RxJava的情况
public void downloadImageAction(View view) {
progressDialog = new ProgressDialog(this);
progressDialog.setTitle("下载图片中...");
progressDialog.show();
new Thread(new Runnable() {
@Override
public void run() {
try {
URL url = new URL(PATH);
HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();
httpURLConnection.setConnectTimeout(5000);
int responseCode = httpURLConnection.getResponseCode(); // 才开始 request
if (responseCode == HttpURLConnection.HTTP_OK) {
InputStream inputStream = httpURLConnection.getInputStream();
Bitmap bitmap = BitmapFactory.decodeStream(inputStream);
Message message = handler.obtainMessage();
message.obj = bitmap;
handler.sendMessage(message);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
}
private final Handler handler = new Handler(new Handler.Callback() {
@Override
public boolean handleMessage(@NonNull Message msg) {
Bitmap bitmap = (Bitmap) msg.obj;
image.setImageBitmap(bitmap);
if (progressDialog != null) progressDialog.dismiss();
return false;
}
});
上面代码很简单，就是开一个线程去下载图片，下载开始的时候显示一个ProgressDialog，下载完成后通过handler发送一个消息到主线程，在主线程里隐藏ProgressDialog，并且显示图片。代码是不是很零散，这里一块，那里一块的，如果我们还有增加需求，比如在图片上加水印，在各个地方加打印日志，代码找起来是不是会比较难找。
- 使用RxJava的情况下：
public void rxJavaDownloadImageAction(View view) {
// 起点
Observable.just(PATH) // 内部会分发 PATH Stirng // TODO 第二步
// TODO 第三步
.map(new Function<String, Bitmap>() {
@Override
public Bitmap apply(String s) throws Exception {
URL url = new URL(PATH);
HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();
httpURLConnection.setConnectTimeout(5000);
int responseCode = httpURLConnection.getResponseCode(); // 才开始 request
if (responseCode == HttpURLConnection.HTTP_OK) {
InputStream inputStream = httpURLConnection.getInputStream();
Bitmap bitmap = BitmapFactory.decodeStream(inputStream);
return bitmap;
}
return null;
}
})
//增加图片水印
.map(new Function<Bitmap, Bitmap>() {
@Override
public Bitmap apply(Bitmap bitmap) throws Exception {
Paint paint = new Paint();
paint.setTextSize(88);
paint.setColor(Color.RED);
return drawTextToBitmap(bitmap, "同学们大家好",paint, 88 , 88);
}
})
// 日志记录
.map(new Function<Bitmap, Bitmap>() {
@Override
public Bitmap apply(Bitmap bitmap) throws Exception {
Log.d(TAG, "apply: 是这个时候下载了图片啊:" + System.currentTimeMillis());
return bitmap;
}
})
// .compose(rxud())
.subscribeOn(Schedulers.io()) // 上面 异步
.observeOn(AndroidSchedulers.mainThread()) // 给下面切换 主线程
// 订阅 起点 和 终点 订阅起来
.subscribe(
// 终点
new Observer<Bitmap>() {
// 订阅开始
@Override
public void onSubscribe(Disposable d) {
// 预备 开始 要分发
// TODO 第一步
progressDialog = new ProgressDialog(DownloadActivity.this);
progressDialog.setTitle("download run");
progressDialog.show();
}
// TODO 第四步
// 拿到事件
@Override
public void onNext(Bitmap bitmap) {
image.setImageBitmap(bitmap);
}
// 错误事件
@Override
public void onError(Throwable e) {
}
// TODO 第五步
// 完成事件
@Override
public void onComplete() {
if (progressDialog != null)
progressDialog.dismiss();
}
});
}
上面的代码充分展示了RxJava的链式不中断调用，一开始有一个Observable的起点，然后通过map操作符将path也就是网络请求地址传入，在apply请求网络返回一个Bitmap，因为网络是耗时操作，所以一开始通过subscribeOn(Schedulers.io())切换到异步io线程执行网络请求，之后再通过observeOn(AndroidSchedulers.mainThread())切换回主线程，在ImageView里显示Bitmap。如果这时我们需要增加需求，例如：中间还包括写日志、增加图片水印操作，我们只有在这个链式调用的地方增加一些操作符就可以了，而且代码看起来十分的优雅（如果使用lambda表达式的话，代码量也会减少很多）。。
现在结合流程图和文章开头所说的Rx思想，现在相信大家对于RxJava的核心思想已经有自己的直观理解了。
我们知道在一个App里使用最多的就是网络请求，RxJava和Retrofit的结合使用，也是现在最常见的网络请求方式。下面结合RxJava的操作符说说RxJava的使用场景。
网络嵌套调用
- 场景一：双层列表
第一个列表的网络请求是查询各个主项目的信息，第二个列表的网络请求是根据主项目的id查询其子item的信息。那么我们是不是可以这样写
.subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Exception {
api.getProject() // 查询主数据
.compose(DownloadActivity.rxud())
.subscribe(new Consumer<ProjectBean>() {
@Override
public void accept(ProjectBean projectBean) throws Exception {
for (ProjectBean.DataBean dataBean : projectBean.getData()) { // 10
// 查询item数据
api.getProjectItem(1, dataBean.getId())
.compose(DownloadActivity.rxud())
.subscribe(new Consumer<ProjectItem>() {
@Override
public void accept(ProjectItem projectItem) throws Exception {
Log.d(TAG, "accept: " + projectItem); // 可以UI操作
}
});
}
}
});
}
});
.compose(DownloadActivity.rxud())，其实就是封装了之前线程切换的两句代码，便于调用，可以不要管，其他的代码大家应该一目了然了。但这样写，我们现在就两个网络请求看不出什么问题，但例如我们现在的app是银行的app的话，我们知道银行app 的业务十分复杂，一个业务请求可能涉及到5、6个网络请求，如果像上面这样写的话，代码一直往里缩进，很不优雅，这时flatMap就派上用场了。
flatMap的使用
// 我只给下面 切换 异步
.observeOn(Schedulers.io())
.flatMap(new Function<Object, ObservableSource<ProjectBean>>() {
@Override
public ObservableSource<ProjectBean> apply(Object o) throws Exception {
return api.getProject(); // 主数据
}
})
.flatMap(new Function<ProjectBean, ObservableSource<ProjectBean.DataBean>>() {
@Override
public ObservableSource<ProjectBean.DataBean> apply(ProjectBean projectBean) throws Exception {
return Observable.fromIterable(projectBean.getData()); // 我自己搞一个发射器 发多次 10
}
})
.flatMap(new Function<ProjectBean.DataBean, ObservableSource<ProjectItem>>() {
@Override
public ObservableSource<ProjectItem> apply(ProjectBean.DataBean dataBean) throws Exception {
return api.getProjectItem(1, dataBean.getId());
}
})
.observeOn(AndroidSchedulers.mainThread()) // 给下面切换 主线程
.subscribe(new Consumer<ProjectItem>() {
@Override
public void accept(ProjectItem projectItem) throws Exception {
// 如果我要更新UI 会报错2 不会报错1
Log.d(TAG, "accept: " + projectItem);
}
});
对于flatMap的理解，网上很多解释特别拗口，其实简单的说，就是你传进去一个事件，这个事件会返回多个结果，这时候，你就可以用操作符flatMap，它会把这个事件返回的多个结果一个个按顺序返回回来，这句话Observable.fromIterable(projectBean.getData());其实就是帮助我们实现了事件返回结果的迭代。
- 场景二：用户注册完直接进行登陆操作
doOnNext的使用
MyRetrofit.createRetrofit().create(IReqeustNetwor.class)
.registerAction(new RegisterRequest()) // todo 1.请求服务器注册操作 // todo 2
.subscribeOn(Schedulers.io()) // 给上面 异步
.observeOn(AndroidSchedulers.mainThread()) // 给下面分配主线程
.doOnNext(new Consumer<RegisterResponse>() { // todo 3
@Override
public void accept(RegisterResponse registerResponse) throws Exception {
// todo 2.注册完成之后，更新注册UI
}
})
// todo 3.马上去登录服务器操作
.observeOn(Schedulers.io()) // 给下面分配了异步线程
.flatMap(new Function<RegisterResponse, ObservableSource<LoginResponse>>() { // todo 4
@Override
public ObservableSource<LoginResponse> apply(RegisterResponse registerResponse) throws Exception {
Observable<LoginResponse> loginResponseObservable = MyRetrofit.createRetrofit().create(IReqeustNetwor.class)
.loginAction(new LoginReqeust());
return loginResponseObservable;
}
})
.observeOn(AndroidSchedulers.mainThread()) // 给下面 执行主线程
.subscribe(new Observer<LoginResponse>() {
// 一定是主线程，为什么，因为 subscribe 马上调用onSubscribe
@Override
public void onSubscribe(Disposable d) {
// TODO 1
progressDialog = new ProgressDialog(RequestActivity.this);
progressDialog.show();
// UI 操作
disposable = d;
}
@Override
public void onNext(LoginResponse loginResponse) { // todo 5
// TODO 4.登录完成之后，更新登录的UI
}
@Override
public void onError(Throwable e) {
}
// todo 6
@Override
public void onComplete() {
// 杀青了
if (progressDialog != null) {
progressDialog.dismiss();
}
}
});
因为subscribe方法返回值是void，调用subscribe的话，整个事件流就结束了，
而observeOn方法的返回值是Observable，
我们就可以继续在被观察者Observable上执行事件流，从而继续执行flatMap等一系列的操作符。
至此关于RxJava的核心使用思想已经介绍完毕了，当然如果大家还想学习RxJava的其他操作符的话可以点击ReactiveX官网或者自行百度，RxJava的操作符非常的多，下图，我总结了开发中会用到的RxJava的所有的操作符，大家可以采用“渐进式”的学习方法，用到哪个再去看哪个，再去学，基本使用思想知道了，百度到用法自然就不会闷逼。
