RxJava基础使用

Rx思想

事件从起点流向终点。

过程中可以对事件进行拦截，拦截时可以对事件进行处理，处理后将处理后的事件继续流向终点。

终点接收上一次处理后的事件。

获取网络图片并显示

基础实现

使用Observable.just(path) 创建图片路径的Observable。

使用map添加拦截，将path转换为Bitmap。

使用map添加拦截，给Bitmap添加水印。

。。。。。随意添加拦截

使用subscribe订阅事件，将Bitmap显示到View中。

private void rxjavaGetBitmap() {
    Log.e(TAG,"rxjavaGetBitmap currentThread = "+Thread.currentThread());
    //图片路径
    String path = "https://pic2.zhimg.com/v2-0dda71bc9ced142bf7bb2d6adbebe4f0_r.jpg";
    //1 起点 类型是 path  创建Observable 事件是path
    Observable.just(path)
            //2 拦截 将String->Bitmap 从网络中获取图片
            .map(new Function<String, Bitmap>() {
                @Override
                public Bitmap apply(String path) throws Exception {
                    //根据path获取Bitmap
                    Log.e(TAG,"2-使用path请求Bitmap（String->Bitmap） currentThread = "+Thread.currentThread());
                    return new HttpURLUtill().getBitmap(path);
                }
            })
            // 上面代码子线程中执行
            .subscribeOn(Schedulers.io())
            // 下面代码在主线程执行
            .observeOn(AndroidSchedulers.mainThread())
            // 3 拦截 Bitmap->Bitmap 给图片添加水印
            .map(new Function<Bitmap, Bitmap>() {
                @Override
                public Bitmap apply(Bitmap bitmap) throws Exception {
                    Log.e(TAG,"3-给图片添加水印（Bitmap->Bitmap） currentThread = "+Thread.currentThread());
                    return bitmap;
                }
            })
            //4 终点 类型是上一个拦截者的返回类型 添加水印后的Bitmap
            .subscribe(new Observer<Bitmap>() {
                @Override
                public void onSubscribe(Disposable d) {
                    //代码执行从此处开始
                    Log.e(TAG,"1-subscribe:onSubscribe currentThread = "+Thread.currentThread());
                    progressDialog = new ProgressDialog(MainActivity.this);
                    progressDialog.show();
                }

                @Override
                public void onNext(Bitmap value) {
                    //5 获取成功 显示图片
                    Log.e(TAG,"4-获取成功，显示图片 currentThread = "+Thread.currentThread());
                    //显示图片
                    ((ImageView)findViewById(R.id.imageview)).setImageBitmap(value);
                }

                @Override
                public void onError(Throwable e) {
                    //执行失败
                    Log.e(TAG,"4-失败 currentThread = "+Thread.currentThread());
                    if(progressDialog != null){
                        progressDialog.dismiss();
                    }
                }

                @Override
                public void onComplete() {
                    //6 执行结束
                    Log.e(TAG,"5-完成 currentThread = "+Thread.currentThread());
                    if(progressDialog != null){
                        progressDialog.dismiss();
                    }
                }
            });
}

中断

对于异步执行的耗时操作，需要使用Disposable实现中断。

在流程开始的时候，保存Disposable对象（在Observer中的onSubscribe时保存Disposable对象）

在Activity onDestory()时如果该流程没有执行完成，需要中断该流程，否则在执行流程结束更新UI的时候，会出现空指针异常。

Rx思想流程

代码执行流程

抽离公共代码 comose

ObservableTransformer+compose

抽离线程切换和添加水印部分功能为ObservableTransformer，使用compose操作符，使用ObservableTransformer。

/**
 * 将部分流程抽离出来共用，使用ObservableTransformer
 * @return
 * @param <UD>                    
 */
private <UD> ObservableTransformer<UD, UD>  rxud(){
    return new ObservableTransformer<UD, UD>(){
        @Override
        public ObservableSource<UD> apply(Observable<UD> upstream) {
            return upstream
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .map(new Function<UD, UD>() {
                        @Override
                        public UD apply(UD ud) throws Exception {
                            Log.e(TAG,"3-给图片添加水印（Bitmap->Bitmap） currentThread = "+Thread.currentThread());
                            return ud;
                        }
                    });
        }
    };
}

/**
 * 抽离部分内容为方法
 */
private void rxjavaGetBitmap2() {
    Log.e(TAG,"rxjavaGetBitmap currentThread = "+Thread.currentThread());
    String path = "https://pic2.zhimg.com/v2-0dda71bc9ced142bf7bb2d6adbebe4f0_r.jpg";
    //起点 类型是 path
    Observable.just(path)
            //拦截 将String->Bitmap
            .map(new Function<String, Bitmap>() {
                @Override
                public Bitmap apply(String path) throws Exception {
                    //根据path获取Bitmap
                    Log.e(TAG,"2-使用path请求Bitmap（String->Bitmap） currentThread = "+Thread.currentThread());
                    return new HttpURLUtill().getBitmap(path);
                }
            })
            .compose(rxud())
            //终点 类型是上一个拦截者的返回类型 Bitmap
            .subscribe(new Observer<Bitmap>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.e(TAG,"1-subscribe:onSubscribe currentThread = "+Thread.currentThread());
                    progressDialog = new ProgressDialog(MainActivity.this);
                    progressDialog.show();
                }

                @Override
                public void onNext(Bitmap value) {
                    //获取成功 显示图片
                    Log.e(TAG,"4-获取成功，显示图片 currentThread = "+Thread.currentThread());
                    //显示图片
                    ((ImageView)findViewById(R.id.imageview)).setImageBitmap(value);
                }

                @Override
                public void onError(Throwable e) {
                    Log.e(TAG,"4-失败 currentThread = "+Thread.currentThread());
                    if(progressDialog != null){
                        progressDialog.dismiss();
                    }
                }

                @Override
                public void onComplete() {
                    //第四步 结束执行
                    Log.e(TAG,"5-完成 currentThread = "+Thread.currentThread());
                    if(progressDialog != null){
                        progressDialog.dismiss();
                    }
                }
            });
}

简化观察者 Consumer

Observer-> Consumer

/**
 * 简化订阅者
 */
private void rxjavaGetBitmap3() {
    Log.e(TAG,"rxjavaGetBitmap currentThread = "+Thread.currentThread());
    String path = "https://pic2.zhimg.com/v2-0dda71bc9ced142bf7bb2d6adbebe4f0_r.jpg";
    //起点 类型是 path
    Observable.just(path)
            //拦截 将String->Bitmap
            .map(new Function<String, Bitmap>() {
                @Override
                public Bitmap apply(String path) throws Exception {
                    //根据path获取Bitmap
                    Log.e(TAG,"2-使用path请求Bitmap（String->Bitmap） currentThread = "+Thread.currentThread());
                    return new HttpURLUtill().getBitmap(path);
                }
            })
            .compose(rxud())
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(Disposable disposable) throws Exception {
                    Log.e(TAG,"1-subscribe:onSubscribe currentThread = "+Thread.currentThread());
                    progressDialog = new ProgressDialog(MainActivity.this);
                    progressDialog.show();
                }
            })
            //终点 类型是上一个拦截者的返回类型 Bitmap
            .subscribe(new Consumer<Bitmap>() {
                @Override
                public void accept(Bitmap bitmap) throws Exception {
                    //第三步 获取成功 显示图片
                    Log.e(TAG,"4-获取成功，显示图片 currentThread = "+Thread.currentThread());
                    //显示图片
                    ((ImageView)findViewById(R.id.imageview)).setImageBitmap(bitmap);
                    if(progressDialog != null){
                        progressDialog.dismiss();
                    }
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    Log.e(TAG,"4-失败 currentThread = "+Thread.currentThread());
                    if(progressDialog != null){
                        progressDialog.dismiss();
                    }
                }
            });
}

RxJava+Retrofit

单个接口

请求执行类型使用RxJava的Observable，网络请求后返回的直接是RxJava的被观察者Observable。

相当于通过Observable.just(path)获取到Observable。

然后可以对Observable执行拦截、订阅。

Retrofit Api接口类：使用玩Android 开放API 玩Android 开放API-玩Android - wanandroid.com

public interface NetWorkApi {
    //获取项目分类
    @GET("project/tree/json")
    Observable<ProjectBean> getProject();

    //获取项目分类下的信息
    @GET("project/list/{pageIndex}/json")
    Observable<ProjectItemBean> getProjectItem(@Path("pageIndex") int  pageIndex, @Query("cid") String cid);
}

获取到项目分类信息

//获取Retrofit
public Retrofit getRetrofit(){
    return new Retrofit.Builder()
            .baseUrl("https://www.wanandroid.com/")
            //Gson解析
            .addConverterFactory(GsonConverterFactory.create())
            //Rxjava CallAdapter
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .build();
}

public void getProjectInfo(View view) {
    //获取项目分类
    getRetrofit().create(NetWorkApi.class)  //获取Retrofit api实现类
            .getProject() //获取Observable<ProjectBean>
            .subscribeOn(Schedulers.io()) //上面代码在子线程中执行
            .observeOn(AndroidSchedulers.mainThread()) //下面代码在主线程中执行
            .subscribe(new Consumer<ProjectBean>() { //观察者，获取结果
                @Override
                public void accept(ProjectBean projectBean) throws Exception {
                    Log.e(TAG, projectBean.toString());
                }
            });
}

多个接口网络嵌套 flatMap

顺序请求两个接口的数据后更新UI。

网络嵌套的方法。

//网络嵌套写法
@SuppressLint("CheckResult")
private void networkNesting(){
    //获取项目分类
    getRetrofit().create(NetWorkApi.class)
            .getProject()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<ProjectBean>() {
                @Override
                public void accept(ProjectBean projectBean) throws Exception {
                    Log.e(TAG, projectBean.toString());
                    for (ProjectBean.DataBean datum : projectBean.getData()) {
                        //根据分类id获取项目列表数据
                        getRetrofit().create(NetWorkApi.class)
                                .getProjectItem(1, datum.getId())
                                .subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Consumer<ProjectItemBean>() {
                                    @Override
                                    public void accept(ProjectItemBean projectItemBean) throws Exception {
                                        Log.e(TAG, projectItemBean.toString());
                                    }
                                });
                    }
                }
            });
}

使用flatMap解决上述网络嵌套问题，将一个事件变为多个事件。

ps：任何一个位置发生异常，整个流程结束执行。

@SuppressLint("CheckResult")
private void noNesting(){
    //获取项目分类
   getRetrofit().create(NetWorkApi.class)
            .getProject()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
           //循环项目分类 1个Observable转多个Observable
            .flatMap(new Function<ProjectBean, ObservableSource<ProjectBean.DataBean>>() {
                @Override
                public ObservableSource<ProjectBean.DataBean> apply(ProjectBean projectBean) throws Exception {
                    return Observable.fromIterable(projectBean.getData());
                }
            })
           //根据项目分类id请求项目分类信息
            .flatMap(new Function<ProjectBean.DataBean, ObservableSource<ProjectItemBean>>() {
                @Override
                public ObservableSource<ProjectItemBean> apply(ProjectBean.DataBean dataBean) throws Exception {
                    Log.e(TAG, dataBean.toString());
                    return getRetrofit().create(NetWorkApi.class).getProjectItem(1, dataBean.getId());
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<ProjectItemBean>() {
                @Override
                public void accept(ProjectItemBean projectItemBean) throws Exception {
                    //获得项目分类信息
                    Log.e(TAG, projectItemBean.toString());
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    Log.e(TAG, "异常 一旦发生异常，整个链式调用就结束了");
                }
            });
}

多个接口过程中更新UI doOnNext

顺序请求两个接口，并在每个接口请求结束后更新相应的UI。

使用doOnNext操作符实现。

private void updateProjectAndProjectItemInfo() {
    getRetrofit().create(NetWorkApi.class)
            .getProject()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            //获取到项目分类信息后，执行UI操作
            .doOnNext(new Consumer<ProjectBean>() {
                @Override
                public void accept(ProjectBean dataBean) throws Exception {
                    ((TextView) findViewById(R.id.tv_project)).setText("我被更新了");
                }
            })
            .observeOn(Schedulers.io())
            //循环项目分类 1个Observable转多个Observable
            .flatMap(new Function<ProjectBean, ObservableSource<ProjectBean.DataBean>>() {
                @Override
                public ObservableSource<ProjectBean.DataBean> apply(ProjectBean projectBean) throws Exception {
                    return Observable.fromIterable(projectBean.getData());
                }
            })
            //根据项目分类id请求项目分类信息
            .flatMap(new Function<ProjectBean.DataBean, ObservableSource<ProjectItemBean>>() {
                @Override
                public ObservableSource<ProjectItemBean> apply(ProjectBean.DataBean dataBean) throws Exception {
                    Log.e(TAG, dataBean.toString());
                    return getRetrofit().create(NetWorkApi.class).getProjectItem(1, dataBean.getId());
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<ProjectItemBean>() {
                @Override
                public void accept(ProjectItemBean projectItemBean) throws Exception {
                    //获得项目分类信息
                    Log.e(TAG, projectItemBean.toString());
                    ((TextView) findViewById(R.id.tv_project_item)).setText("我被更新了");
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    Log.e(TAG, "异常 一旦发生异常，整个链式调用就结束了");
                }
            });
}