前言

上一节中，我们一次性发送128个事件没有任何问题，但是一旦超过128个立马抛出MissingBackpressureException异常，提示你上游发送事件太多，下游处理不过来，如何解决？这一节就来解决下。

1. 解决方法

---

思路一：

发送128个事件没问题是因为Flowable内部有一个大小为128的水缸，超过则会溢出，既然水缸这么小，就换一个大点的水缸试下，代码如下：

    /**
     * BUFFER：大的水缸，
     *          让上游无限的发送事件，下游一个也不处理，结果容易造成OOM
     */
    public static void demo1(){
        // 创建一个上游：Flowable
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {          // 无限for循环
                    Log.e("TAG" , "emit " + i) ;
                    emitter.onNext(i);
                }
            }
        } , BackpressureStrategy.BUFFER).subscribeOn(Schedulers.io())  // 让上游的for循环在子线程中执行
            .observeOn(AndroidSchedulers.mainThread())   // for循环执行完后切回到主线程
            .subscribe(new Subscriber<Integer>() {
                @Override
                public void onSubscribe(Subscription s) {
                    Log.e("TAG" , "subscribe") ;
                    mSubscription = s ;
                }

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

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

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

换大的水缸，在上游使用无限for循环发射数据，下游一个也不处理，导致OOM；

图片.png

思路二：

之前解决Observable如何解决上游发送事件太快，有两种方法：从数量和速度两个方面解决，同样的，Flowable也有对应的两种方法：BackpressureStrategy.DRAP和BackpressureStrategy.LATEST。

• Drap：直接把存不下的事件丢掉；
• Latest：只保留最新的数据；

Drop代码如下：

    /**
     * Drop：丢弃存储不下的事件
     */
    public static void demo2(){
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i < 10000; i++) {
                    emitter.onNext(i);
                }
            }
        } , BackpressureStrategy.DROP).subscribeOn(Schedulers.io())    // 让上游的for循环在子线程中执行
                .observeOn(AndroidSchedulers.mainThread())    // 切换到主线程执行下边的操作
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.e("TAG" , "subscribe -> ") ;
                        mSubscription = s ;
                        s.request(128);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("TAG" , "next -> " + integer) ;
                    }

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

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

效果如如下：

图片.png

• 结果就是：第一次调用 s.request(128); 方法时候，下游收到 0-127共128个事件，第二次调用 s.request(128)就不确定，上游在一直发送事件，内存正常，drop丢弃了存不下事件；

latest代码如下：

让上游无限发送事件，把Subscription保存起来，方便在外边调用 s.request(128)方法

    /**
     * latest：只保留最新数据
     */
    public static void demo3(){
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i < 10000; i++) {
                    emitter.onNext(i);
                }
            }
        } , BackpressureStrategy.LATEST).subscribeOn(Schedulers.io())   // 让上游的for循环在子线程中执行
                .observeOn(AndroidSchedulers.mainThread())  // 切换到主线程中执行下边操作
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.e("TAG" , "subscribe") ;
                        mSubscription = s ;
                        s.request(128);
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.e("TAG" , "next -> " + integer) ;
                    }

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

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

interval操作符：

/**
     *  interval操作符
     */
    public static void demo4(){
        Flowable.interval(1 , TimeUnit.MICROSECONDS)
                .onBackpressureDrop()    // 添加背压策略
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.e("TAG" , "subscribe") ;
                        mSubscription = s ;
                        s.request(Long.MAX_VALUE);
                    }

                    @Override
                    public void onNext(Long aLong) {
                        Log.e("TAG" , "next -> " + aLong) ;
                    }

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

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

八篇文章所有代码已上传至github：
https://github.com/shuai999/RxJava2Demo.git