Rxjava3文档级教程三：实战演练

Naiveee 2020-09-11

展开全文

商业转载请联系作者获得授权，非商业转载请注明出处。

Rxjava3文档级教程一：介绍和基本使用

3.1 Observable.unsubscribe

3.2 disposable.dispose

3.3 CompositeDisposable

3.4 Rxlifecycle

参考文章：

一结合RxBinding

RxBinding 的 GitHub 地址

RxBinding 能够把 Android 平台的兼容包内的 UI 控件变为 Observaber 对象. 可以把 UI 控件的事件当作 RxJava 中的数据流来使用。

依赖如下：

Platform bindings:

implementation 'com.jakewharton.rxbinding3:rxbinding:3.1.0'

AndroidX library bindings:

implementation 'com.jakewharton.rxbinding3:rxbinding-core:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-appcompat:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-drawerlayout:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-leanback:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-recyclerview:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-slidingpanelayout:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-swiperefreshlayout:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-viewpager:3.1.0'
implementation 'com.jakewharton.rxbinding3:rxbinding-viewpager2:3.1.0'

Google 'material' library bindings:

implementation 'com.jakewharton.rxbinding3:rxbinding-material:3.1.0'

RxBinding 的优点:

它是对 Android View 事件的扩展, 它使得开发者可以对 View 事件使用 RxJava 的各种操作。
提供了与 Rxjava 一致的回调, 使得代码简洁明了。
几乎支持所有的常用控件及事件, 还对 Kotlin 支持。
可以应用于整个 App 的所有 UI 事件。

进阶案例：

1.1 按钮防抖功能：

相比之前的定时器或者标志位，来实现的快速点击下的防抖功能，采用RxView可以大大简化代码：

         /**
         * 按钮点击防抖
         */
        RxView.clicks(mBt)
                .throttleFirst(1, TimeUnit.SECONDS)
                .subscribeOn(AndroidSchedulers.mainThread())
                .subscribe(v -> Log.i(TAG, "点击按钮"));
         /**
         * 按钮长按
         */
        RxView.longClicks(mBt)...

1.2 输入框输入实时网络请求步长控制：

很多App都有这种顶部的搜索框，而现有的实现，一般都会在用户进行输入时，用EditText里自带的addTextChangedListener(new TextWatcher()来监听文本变化，进行实时搜索。

 etPriceBegin.addTextChangedListener(new TextWatcher() {
            @Override
            public void onTextChanged(CharSequence s, int start, int before,
                                      int count) {
                //s:变化后的所有字符
                Toast.makeText(getContext(), "变化:"+s+";"+start+";"+before+";"+count, Toast.LENGTH_SHORT).show();
                Log.i("Seachal:","变化:"+s+";"+start+";"+before+";"+count);
            }
            @Override
            public void beforeTextChanged(CharSequence s, int start, int count,
                                          int after) {
                //s:变化前的所有字符； start:字符开始的位置； count:变化前的总字节数；after:变化后的字节数
                Toast.makeText(getContext(), "变化前:"+s+";"+start+";"+count+";"+after, Toast.LENGTH_SHORT).show();
                Log.i("Seachal:","变化前:"+s+";"+start+";"+count+";"+after);
            }
            @Override
            public void afterTextChanged(Editable s) {
                //S：变化后的所有字符；start：字符起始的位置；before: 变化之前的总字节数；count:变化后的字节数
                Toast.makeText(getContext(), "变化后:"+s+";", Toast.LENGTH_SHORT).show();
                Log.i("Seachal:","变化后:"+s+";");
            }
        });

但有了RxView后，一切更加方便了起来。下面这个例子里，可以跳过用户的第一次输入不做处理， debounce(1, TimeUnit.SECONDS)又实现了用户在快速删除或者输入时，不会因为每次的实时变化都去进行网络请求或者相关操作，而是在间隔1s后才去进行这些操作。

        ed = findViewById(R.id.ed);
        RxTextView.textChanges(ed)
                .debounce(1, TimeUnit.SECONDS)
                //跳过第1次请求 因为初始输入框的空字符状态
                .skip(1)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<CharSequence>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }
                    @Override
                    public void onNext(CharSequence charSequence) {
                        Log.i(TAG,"收到的字符： " + charSequence.toString());
                    }
                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "onError： " + e.getMessage() );
                    }
                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });

1.3 联合/表单判断

        Observable<CharSequence> ObservableName = RxTextView.textChanges(mEtPhone);
        Observable<CharSequence> ObservablePassword = RxTextView.textChanges(mEtPassword);
        Observable.combineLatest(
                ObservableName, ObservablePassword,
                (phone, password) -> {
                    return isPhoneValid(phone.toString()) && isPasswordValid(password.toString());
                })
                .subscribe(aBoolean -> {
                    mBtLogin.setEnabled(aBoolean);
//                    Toast.makeText(getApplicationContext(), "默认的Toast", Toast.LENGTH_SHORT).show();
                });
        //Lambda写法
        Observable.combineLatest(ObservableName, ObservablePassword
                , (phone, password) -> isPhoneValid(phone.toString()) && isPasswordValid(password.toString()))
                .subscribe(mBtLogin::setEnabled);

1.4 定时器任务

在我们的登录注册页，少不了一个获得验证码的倒计时定时器，一般开发者都会选择继承CountDownTimer类，神说：太麻烦了！于是有了Rx。下面的代码中，mBt在第一次点击后就处于setEnabled(false)的状态,不可点击。

    private Observable<Boolean> verifyCodeObservable;
    private static int SECOND = 20;
    verifyCodeObservable = RxView.clicks(mBt)
                .throttleFirst(SECOND, TimeUnit.SECONDS) //防止20秒内连续点击,或者只使用doOnNext部分
                .subscribeOn(AndroidSchedulers.mainThread())
                .map(o -> false)
                .doOnNext(mBt::setEnabled);
        verifyCodeObservable.subscribe(s -> {
            Observable.interval(1, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
                    .take(SECOND)
                    .subscribe(aLong -> {
                                RxTextView.text(mBt).accept("剩余" + (SECOND - aLong) + "秒");
                            }
                            , Throwable::printStackTrace
                            , () -> {
                                RxTextView.text(mBt).accept("获取验证码");
                                RxView.enabled(mBt).accept(true);
                            });
        });

二

2.1 嵌套网络请求

        // flatMap
        MyRxView.clicks(btn3)
                .throttleFirst(1000, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .flatMap(new Function<Object, ObservableSource<ProjectBean>>() {
                    @Override
                    public ObservableSource<ProjectBean> apply(Object o) throws Exception {
                        return wanAndroidApi.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 wanAndroidApi.getProjectItem(1,dataBean.getId());
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<ProjectItem>() {
                    @Override
                    public void accept(ProjectItem projectItem) throws Exception {
                        Log.i(TAG, "accept: " + projectItem);
                    }
                });
        //lambda方式
//        RxView.clicks(btn3)
//                .throttleFirst(1000, TimeUnit.MILLISECONDS)
//                .observeOn(Schedulers.io())
//                .flatMap( o -> wanAndroidApi.getProject())
//                .flatMap(projectBean -> Observable.fromIterable(projectBean.getData()))
//                .flatMap(dataBean -> wanAndroidApi.getProjectItem(1,dataBean.getId()))
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(projectItem -> Log.i(TAG, "projectItem: " + projectItem));

三防泄漏

3.1 Observable.unsubscribe

@Override
    protected void onDestroy() {
        super.onDestroy();
        Observable.unsubscribeOn(AndroidSchedulers.mainThread()); //防止泄露
    }

3.2 disposable.dispose

@Override
    protected void onDestroy() {
        super.onDestroy();
        if(disposable!=null && !disposable.isDisposed()){
            disposable.dispose();
        }
    }

3.3 CompositeDisposable

如果在请求过程中，UI层destroy了怎么办，不及时取消订阅，可能会造成内存泄漏。因此，CompositeDisposable就上场了，它可以对我们订阅的请求进行统一管理。CompositeDisposable的clear方法内部，实际上就会调用Disposable的dispose方法。
大致三步走：
1、在UI层创建的时候（比如onCreate之类的），实例化CompositeDisposable；
2、把subscribe订阅返回的Disposable对象加入管理器；
3、UI销毁时清空订阅的对象。

private CompositeDisposable mCompositeDisposable;
// when create UI
mCompositeDisposable = new CompositeDisposable();
// when request data
if (mCompositeDisposable != null && !mCompositeDisposable.isDisposed()) {
    mCompositeDisposable.add(disposable);
}
// when destroy UI
if (mCompositeDisposable != null) {
    mCompositeDisposable.clear(); // clear时网络请求会随即cancel
    mCompositeDisposable = null;
}

这样我们就可以管理脱缰的网络请求了，相当于将它与UI的生命周期绑定。只要稍稍将上述模板封装一哈，就比较方便了。比如add操作可以封装一个方法，每次网络请求时add一发就好。
对于MVP架构的项目，CompositeDisposable完全可以封装到Presenter当中。这里就不展开了。

3.4 Rxlifecycle

github官网

我们可以利用Rxlifecycle来解决内存泄漏问题。

依赖：

implementation 'com.trello.rxlifecycle3:rxlifecycle:3.1.0'
// If you want to bind to Android-specific lifecycles
implementation 'com.trello.rxlifecycle3:rxlifecycle-android:3.1.0'
// If you want pre-written Activities and Fragments you can subclass as providers
implementation 'com.trello.rxlifecycle3:rxlifecycle-components:3.1.0'
// If you want pre-written support preference Fragments you can subclass as providers
implementation 'com.trello.rxlifecycle3:rxlifecycle-components-preference:3.1.0'
// If you want to use Android Lifecycle for providers
implementation 'com.trello.rxlifecycle3:rxlifecycle-android-lifecycle:3.1.0'
// If you want to use Kotlin syntax
implementation 'com.trello.rxlifecycle3:rxlifecycle-kotlin:3.1.0'
// If you want to use Kotlin syntax with Android Lifecycle
implementation 'com.trello.rxlifecycle3:rxlifecycle-android-lifecycle-kotlin:3.1.0'
// If you want to use Navi for providers
// DEPRECATED: Use rxlifecycle-android-lifecycle instead. This will be removed in a future release.
implementation 'com.trello.rxlifecycle3:rxlifecycle-navi:3.1.0'

You can bind when the lifecycle emits anything:

myObservable
    .compose(RxLifecycle.bind(lifecycle))
    .subscribe();

Or you can bind to when a specific lifecyle event occurs:

myObservable
    .compose(RxLifecycle.bindUntilEvent(lifecycle, ActivityEvent.DESTROY))
    .subscribe();

Alternatively, you can let RxLifecycle determine the appropriate time to end the sequence:

myObservable
    .compose(RxLifecycleAndroid.bindActivity(lifecycle))
    .subscribe();

例子：

public class RxLifeActivity extends RxAppCompatActivity {
    /**
     * RxLifecycle使用：在当前activity中继承RxAppCompatActivity
     * <p>
     * ActivityEvent:手动设置指定在什么时候取消订阅，下列枚举
     * CREATE,
     * START,
     * RESUME,
     * PAUSE,
     * STOP,
     * DESTROY
     */
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx_life);
        Flowable.interval(3, 2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .compose(this.bindUntilEvent(ActivityEvent.PAUSE)) //手动设置在activity onPause的时候取消订阅
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(aLong -> Log.e("liuqiang", "RxLifeActivityOnCreate:" + aLong));
    }
    @Override
    protected void onStart() {
        super.onStart();
        Flowable.interval(6, 3, TimeUnit.SECONDS)
                //bindToLifecycle的自动取消订阅示例，因为是在onStart的时候调用，所以在onStop的时候自动取消订阅
                .compose(this.bindToLifecycle()) //设置在默认的生命周期中取消订阅
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(aLong -> Log.e("liuqiang", "RxLifeActivityOnStart:" + aLong));
    }
}