RxJava Android实际应用

这里的网络请求部分在我的另一篇文章Retrofit + OkHttp3 + RxJava2中，后面用到的Retrofits、Api都是来自该项目的一些Demo

这里记录一些最近开发的想法和思路，一步一步是怎么实现一个简单功能，并去考虑拓展，逐步完善的过程。下面的需求实际是我自己的不断考虑和改变的一个过程，用来展示RxJava的一些优势。实际项目开发中这些应该提前就考虑到，但是需求总是变化的，这些优势也或多或少的会有价值。希望看完这篇文章你会有收获。

开始

请求一个网络结果并展示在View

接Retrofit + OkHttp3 + RxJava2中的例子

Retrofit配置中默认网络请求在IO线程

TextView showView;
...
Retrofits.get(Api.class)
  .getServerInfo()
  //指定订阅者在UI线程响应
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(result -> {
     showView.setText(result.toString());
  });

异常抓取

这里用到了onErrorReturn关键字，其作用就是当错误发生的时候返回一个默认值给订阅者。

Return的数据必须为Observable的数据类型，在Retrofits返回数据中为Observable<BaseResult<T>>，这里需要返回一个BaseResult<T>，携带错误信息

需求：网络无法连接code1001，连接超时code为1002，解析错误code为1100

这里新建一个工具类Observables。

public class Observables {
    public static <T> Function<Throwable, BaseResult<T>> getErrorReturn() {
        return throwable -> {
            BaseResult<T> result = new BaseResult<>();
            if (throwable instanceof ConnectException ) {
                result.setCode(1001);
                result.setMessage("无法访问服务器");
            } else if (throwable instanceof SocketTimeoutException) {
                result.setCode(1002);
                result.setMessage("连接超时");
            } else if (throwable instanceof JsonParseException) {
                result.setCode(1100);
                result.setMessage("解析失败");
            }
            return result;
        };
    }
  .....
}

添加到业务处理

TextView showView;
...
Retrofits.get(Api.class)
  .getServerInfo()
  .onErrorReturn(Observables.getErrorReturn())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(result -> {
     showView.setText(result.toString());
  });

这里当产生这3个Exception后就能直接返回错误码，方便上层处理

绑定生命周期

在很多异步处理请求中，我们都需要判断当数据请求返回后，是否我当前的页面被销毁，如果不做判断直接使用会出现崩溃或者内存泄露。

思路：

• 对Acitivy或者Fragment在生命周期做一个Observable当生命周期改变的时候就发射当前的状态
• 筛选生命周期的数据只发射结束的那一个数据
• 使用takeUntil关键字，当takeUntil设置的Observable发射任何一个数据时候，就取消订阅

核心代码(已Fragment为例)

static final int INIT = 0;
static final int RELEASE = 1;
Observable<Integer> mLife = BehaviorSubject.create();
...
  
...
@Override
public void onDestroyView() {
  mLife.onNext(RELEASE);
  super.onDestroyView();
}

Retrofits.get(Api.class)
  .getServerInfo()
  .onErrorReturn(getErrorReturn())
  //绑定生命周期
  .takeUntil(mLife.filter(state -> state == RELEASE))
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(result -> {
     showView.setText(result.toString());
  });

我的另一个项目RxApp实现了该功能，这里就直接使用。

  Retrofits.get(Api.class)
  .getServerInfo()
  .onErrorReturn(getErrorReturn())
  //绑定生命周期
  .compose(RxApp.with(this).bindLife())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(result -> {
     showView.setText(result.toString());
  });

绑定一个Dialog

需求

• 在开始请求时显示Dialog
• Dialog不可取消
• 网络请求结束关闭

这里用到关键字doOnLifecycle(final Consumer<? super Disposable> onSubscribe, final Action onDispose)，用于在订阅和取消订阅的时候进行回调。

需要注意的是doOnLifecycle并没有默认的线程调度器，需要指定调度器在UI主线程

添加到工具类

public class Observables {
  ...
    public static <T> ObservableTransformer<T, T> bindDialog(Context context) {
        //具体样式就不详述了
        final Dialog dialog = new ProgressDialog(context);
        dialog.setCancelable(false);
        return upstream -> upstream.doOnLifecycle(disposable -> dialog.show(), dialog::dismiss)
                //设置在UI主线程执行
                .subscribeOn(AndroidSchedulers.mainThread());
    }
}

使用

Retrofits.get(Api.class)
  .getServerInfo()
  .onErrorReturn(getErrorReturn())
  //绑定Dialog
  .compose(Observables.bindDialog(getActivity()))
  .compose(RxApp.with(this).bindLife())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(result -> {
     showView.setText(result.toString());
  });

Dialog可取消同时取消请求

同样使用关键字takeUntil来处理该流程

public class Observables {
  ...
    public static <T> ObservableTransformer<T, T> bindCancelDialog(Context context) {
        final Dialog dialog = new ProgressDialog(context);
        dialog.setCancelable(true);
        dialog.setCanceledOnTouchOutside(false);
        //定义取消触发的Observable
        BehaviorSubject<Boolean> dialogCancelSubject = BehaviorSubject.create();
        dialog.setOnCancelListener(dialog1 -> dialogCancelSubject.onNext(true));
        return upstream -> upstream.doOnLifecycle(disposable ->dialog.show(),dialog::dismiss)
                .subscribeOn(AndroidSchedulers.mainThread())
                .takeUntil(dialogCancelSubject);
    }
}

使用

Retrofits.get(Api.class)
  .getServerInfo()
  .onErrorReturn(getErrorReturn())
  //绑定Dialog
  .compose(Observables.bindDialog(getActivity()))
  .compose(RxApp.with(this).bindLife())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(result -> {
     showView.setText(result.toString());
  });

Dialog 2秒后可取消

使用Observable.timer来实现

public class Observables {
  ...
    public static <T> ObservableTransformer<T, T> bindCancelDialog(Context context) {
        final Dialog dialog = new ProgressDialog(context);
        dialog.setCancelable(true);
        dialog.setCanceledOnTouchOutside(false);
        BehaviorSubject<Boolean> dialogCancelSubject = BehaviorSubject.create();
        dialog.setOnCancelListener(dialog1 -> dialogCancelSubject.onNext(true));
        return upstream -> upstream.doOnLifecycle(disposable -> {
                    dialog.show();
                    Observable.timer(2, TimeUnit.SECONDS)
                            .observeOn(AndroidSchedulers.mainThread())
                            .subscribe(time -> {
                                if(dialog.isShowing()){
                                    Toast.makeText(context,"点击返回可取消",Toast.LENGTH_SHORT).show();
                                    dialog.setCancelable(true);
                                }
                            });
                },
                dialog::dismiss)
                .subscribeOn(AndroidSchedulers.mainThread())
                .takeUntil(dialogCancelSubject);
    }
}

总结

上面介绍了如何从一个简单的网络，到添加一系列的业务处理。不难发现RxJava优势，

• 代码流程看起来聚合很高，而且很易读懂（前提是你了解Rxjava的关键字的作用）
• 在业务中间添加流程很简单，方便以后的维护
• 解耦很方便，很多功能性的处理我们可以直接提出成工具类，使用compose进行批量处理

好了就分享这么多，希望对大家有帮助。