RxJava在Android的开发里，一般用于网络请求，配合retrofit和OKhttp。
OKhttp：真正发生网络请求的地方。
retrofit：一个网络请求的框架，底层实现是OKhttp，使用retrofit框架的目的就是使得我们的网络请求更加易用。
RxJava：也是一个框架，目的是解决任务的同步和异步执行，同时还可以通过多种多样的操作符，实现不同的需求。
题外话：现在协程的作用，跟RxJava相同，而且个人感觉，协程使用后，更加的直观，代码逻辑清晰。
简单的对比可以戳：Android协程使用
但这里是打算记录下RxJava的，所以还是需要说明白的。RxJava的核心思想就是响应式编程，就是有起点和重点，就是一个条完整的链，而我们可以在中间利用操作符进行各种拦截处理。
RxJava怎么用
private fun requestByRx() {
if (::retrofit.isInitialized && ::api.isInitialized) {
api.listReposRx("TonyDash")
.subscribeOn(Schedulers.io()) // 上面 异步
.observeOn(AndroidSchedulers.mainThread()) // 下面 主线程
.subscribe(object : SingleObserver<List<Repo>> {
override fun onSuccess(t: List<Repo>) {
textView.text = "Rx${t[0].name}"
}
override fun onSubscribe(d: Disposable) {
textView.text = "onSubscribe"
}
override fun onError(e: Throwable) {
textView.text = e.message?:"onError"
}
})
}
}
这就是最简单的使用例子，api.listReposRx("TonyDash")就是我们的请求起点，而subscribe里面的SingleObserver<List<Repo>>，就是我们最后得到的结果。而在这个起点和结果之间，我们可以在中途增加各种各样的操作符。
例如如果是一个加载图片的接口，得到图片后，我们需要写点东西上去。
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);
}
})
// 订阅 起点 和 终点 订阅起来
.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();
}
});
}
这里就用到了操作符map，把文字写在了bitmap上，然后再传给下一个，这里是交给结果去显示了。
RxJava的优点
当然是有好处，才这么多人用了。RxJava是很明显的链式调用，我们可以根据上一个的处理结果，进行下一步处理，然后交给下一步，直到输出结果的那一步。
这样的好处是：
①避免了嵌套调用，造成逻辑混乱不清晰
②可以动态添加或者修改部分逻辑，例如上面的水印逻辑。
肉眼可见的设计模式
观察者模式
Observable.just(PATH)就是被观察者，new Observer<Bitmap>()就是观察者，被观察者内部会有一个发射器ObservableEmitter，通过调用发射器的onNext方法，通知观察者接受消息，这时候观察者的onNext方法就会被调用。
image.png
装饰者模式
这模式可以理解为买奶茶，基础就是奶茶，但是中途你可以加珍珠、加椰果、加果冻，最后打包拿手上喝，过程中的各种加料就是装饰者。而在RxJava中，就是各种的操作符，例如map。
image.png
策略模式
切换线程的时候，可以选择io线程，主线程等，这里使用的就是策略模式。
RxJava的线程切换
.subscribeOn(Schedulers.io()) // 上面 异步
.observeOn(AndroidSchedulers.mainThread()) // 下面 主线程
先查看Schedulers.io()
@NonNull
public static Scheduler io() {
return RxJavaPlugins.onIoScheduler(IO);
}
@NonNull
public static Scheduler onIoScheduler(@NonNull Scheduler defaultScheduler) {
Function<? super Scheduler, ? extends Scheduler> f = onIoHandler;
return f == null ? defaultScheduler : (Scheduler)apply(f, defaultScheduler);
}
可以看到onIoScheduler里面，其实如果我们不设置，就是默认返回defaultScheduler，那就直接看IO常量是怎么初始化的。
static final Scheduler IO = RxJavaPlugins.initIoScheduler(new Schedulers.IOTask());
static final class IOTask implements Callable<Scheduler> {
IOTask() {
}
public Scheduler call() throws Exception {
return Schedulers.IoHolder.DEFAULT;
}
}
看到IOTask是实现Callable的，也就是说这是线程，返回的是Schedulers.IoHolder.DEFAULT。
到这里就不需要再看了，因为线程的执行，后面肯定是线程池，线程池这里不详细说了。
那怎么再切回主线程？
.observeOn(AndroidSchedulers.mainThread()) // 下面 主线程
其实跟IO一样，也是返回一个主线程的Schedulers。不同的是，主线程的Schedulers是一个HandlerSchedulers。看名字就知道是一个handler，而且是主线程的handler，所以可以更新UI界面。
flatMap多次分发事件和RxBinding防止抖动
RxView.clicks(bt_anti_shake)
.throttleFirst(2000, TimeUnit.MILLISECONDS) // 2秒钟之内 响应你一次
// 切换 异步
.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());
}
})
.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 {
Log.d(TAG, "accept: " + projectItem);
}
});
image.png
网友评论