Android RxJava 2.x入门例子详解（五）

前言

上一节我们讲解了线程调度器，这个时候我们可能需要考虑一个问题。上下游工作在不同的线程，这时候收发数据无需等待，当上游发事件的速度太快, 下游处理事件的速度太慢会发生什么事情？这也就是所谓的背压（Backpressure）问题。

我们先看一个例子：

Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; ; i++) {    //无限循环发事件
                    Log.d(TAG, "Observable发出："+i);
                    e.onNext(i);
                }
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Thread.sleep(1000);//一秒处理一次数据
                Log.d(TAG, "" + integer);
            }
        });

没错，我们的内存会爆掉，最后OOM，那要怎么解决这个问题呢？
不用担心，RxJava2.x已经为你想到这个问题。

Flowable和Subscriber的使用

Flowable<Integer> flowable=Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "flowable：B1");
                e.onNext(1);//向下游（观察者）发射内容1
                Log.d(TAG, "flowable：B2");
                e.onNext(2);
                Log.d(TAG, "flowable：B3");
                e.onNext(3);
                Log.d(TAG, "flowable：B4");
                e.onNext(4);
                Log.d(TAG, "flowable：B5");
                e.onNext(5);
                e.onComplete();
            }
        }, BackpressureStrategy.ERROR);//增加了一个参数

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                s.request(Long.MAX_VALUE);  //注意这句代码，这里告诉上游，下游的处理能力
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);
            }

            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        };

        flowable.subscribe(subscriber);

我们稍微修改一下代码：

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "flowable：1");
                e.onNext(1);//向下游（观察者）发射内容1
                Log.d(TAG, "flowable：2");
                e.onNext(2);
                Log.d(TAG, "flowable：3");
                e.onNext(3);
                Log.d(TAG, "flowable：4");
                e.onNext(4);
                Log.d(TAG, "flowable：5");
                e.onNext(5);
                e.onComplete();
            }
        }, BackpressureStrategy.ERROR)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        s.request(3);  //这里告诉上游，下游的处理能力为3
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        //Log.d(TAG, "onError");
                        Log.w(TAG, "onError: ", t);//上游多发事件报警告
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });

运行结果报异常警告，为什么呢？因为s.request(3)告诉上游处理能力为3，上游多发事件，则直接报警告。
而多发事件的处理是由BackpressureStrategy.ERROR决定
该参数还有以下选项：
BackpressureStrategy.DROP(多发则丢弃不处理)
BackpressureStrategy.BUFFER(多发则先放到缓存)
BackpressureStrategy.LATEST(多发则只保留最后的事件)

再看一个例子：

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
                for (int i = 0;i<100 ; i++) {
                    Log.d(TAG, "flowable："+i);
                    e.onNext(i);
                }
            }
        }, BackpressureStrategy.BUFFER)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        mSubscription=s;
                        Log.d(TAG, "onSubscribe");
                        s.request(10);  //这里告诉上游，下游的处理能力为10
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        //Log.d(TAG, "onError");
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });

我们使用BackpressureStrategy.BUFFER把多发的事件保存到缓存里，运行代码没有警告异常。
这时候我们可以通过外部控制mSubscription.request，例如点击按钮调用mSubscription.request(10);
系统会继续运行输出onNext。

上面都是同一线程下的结果，如果是不同线程会发生什么？

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
                for (int i = 0; ; i++) {
                    Log.d(TAG, "flowable："+i);
                    e.onNext(i);
                }
            }
        }, BackpressureStrategy.ERROR)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        s.request(10);  //这里告诉上游，下游的处理能力为10，实际上并没什么卵用
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                    }

                    @Override
                    public void onError(Throwable t) {
                        //Log.d(TAG, "onError");
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });

运行代码，下游虽然通过s.request(10)告诉上游处理能力为10，但是上游并不理会，而继续不断发送事件。这是为什么呢？是不是上游还不知道下游的处理能力？那有没有办法呢？

答案就是FlowableEmitter的requested()

再看看下面的例子：

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
                for (int i = 0; i<300 ; i++) {
                    //Log.d(TAG, "flowable："+i);
                    Log.d(TAG,"requested："+e.requested());
                    e.onNext(i);
                }
            }
        }, BackpressureStrategy.ERROR)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        s.request(3);  //异步情况，上下游不同线程，这里告诉上游处理能力为3，实际是128+3,128是由Flowable初始化
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                        //mSubscription.request(1);
                    }

                    @Override
                    public void onError(Throwable t) {
                        //Log.d(TAG, "onError");
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });

运行结果发现e.requested()获取到的结果是128，每调用一次onNext减1。
为什么是128？是因为异步情况下，上下游不同线程，这里告诉上游处理能力为3，实际是128+3,128是由Flowable初始化。
如果是同一线程则得到结果一样哦，这里不展示（同学们自己尝试吧）

既然有e.requested()，那么好办了
我们稍微修改一下程序：

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<Integer> e) throws Exception {
                boolean flag;

                for (int i = 0; ; i++) {    //无限循环发事件
                    Log.d(TAG,"requested："+e.requested());
                    flag = false;
                    while (e.requested()==0){
                        if (!flag){
                            Log.d(TAG,"下游不能处理，暂不能发送");
                            flag=true;
                        }
                    }
                    e.onNext(i);
                }
            }
        }, BackpressureStrategy.ERROR)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    private Subscription mSubscription;
                    @Override
                    public void onSubscribe(Subscription s) {
                        mSubscription=s;
                        Log.d(TAG, "onSubscribe");
                        s.request(3);  //异步情况，上下游不同线程，这里告诉上游处理能力为3，实际是128+3,128是由Flowable初始化
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "onNext: " + integer);
                        try {
                            Thread.sleep(1000);
                            mSubscription.request(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onError(Throwable t) {
                        //Log.d(TAG, "onError");
                        Log.w(TAG, "onError: ", t);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete");
                    }
                });

运行程序发现，首次上游调用了128次事件，随后下游处理96次后，上游再次调用96次事件，这样一直循环下去。可见程序很有规律，没有出现OOM。
为什么会下游处理96次后上游会继续调用事件？
这是因为Flowable内部实现了s.request请求，也就是说下游处理了一部分事件后，Flowable内部自动调用s.request(95)，自动增加处理能力。