再忆RxJava---背压策略

提莫队长

发布于 2020-06-02 15:25:46

6630

发布于 2020-06-02 15:25:46

文章被收录于专栏：刘晓杰

1 背压存在的背景

被观察者发送事件速度太快，而观察者来不及接收所有事件，从而导致观察者无法及时响应或者处理所有发送过来事件的问题，最终导致缓存区溢出、事件丢失 & OOM

2 背压策略的原理

2.1 未雨绸缪（事情在还没有发生之前做一定的处理），一共有两种（1）控制被观察者发送事件的速度---反馈控制（2）控制观察者接收事件的速度---响应式拉取
2.2 亡羊补牢（事情已经发生，如何补救）---对多余的数据进行有选择的抛弃，或者保留，或者报错

3 背压具体情况讨论

3.1 同步策略

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) {
                Log.e("emitter", "发送1");
                emitter.onNext("111");
                Log.e("emitter", "发送2");
                emitter.onNext("222");
                Log.e("emitter", "发送3");
                emitter.onNext("333");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribe(new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(3);
            }

            @Override
            public void onNext(String s) {
                Log.e("emitter", "接受" + s);
            }

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

            @Override
            public void onComplete() {

            }
        });

其实对于同步而言，讨论背压毫无意义。emitter.onNext然后直接就是Subscriber.onNext，然后再下一个emitter.onNext。因为这是同步的，不存在缓存队列。就如例子而言，s.request(n)，如果n小于3，会根据Error策略，直接走OnError方法（具体请看代码）。如果n大于3，是5，直接onComplete，不管有没有发送满5个总的来说，同步并没有采用什么背压，如果非要说的话，那也是亡羊补牢式的

3.2 异步

先来看几段代码 FlowableCreate---NoOverflowBaseAsyncEmitter的onNext方法

        public final void onNext(T t) {
            。。。。。。
            if (get() != 0) {//get最初是128，也就是buffer-size，这是子线程
                downstream.onNext(t);
                BackpressureHelper.produced(this, 1);
            } else {
                onOverflow();
            }
        }

也就是发送的时候，超过128个数据，就走onError，没有就往下一个onNext走 (可以先看一下ObserveOnSubscriber的onSubscribe函数，里面有queue的构造，以及sourceMode其实并没有赋值) 再来看BaseObserveOnSubscriber的onNext方法

        @Override
        public final void onNext(T t) {
            if (done) {
                return;
            }
            if (sourceMode == ASYNC) {
                trySchedule();
                return;
            }
            if (!queue.offer(t)) {//这个queue就是FlowableObserveOn的构造函数中的prefetch大小的一个队列。这里默认是128
                //也就是最上面get为什么是128的原因
                //此时还没到Handler，所以还是子线程
                upstream.cancel();
                error = new MissingBackpressureException("Queue is full?!");
                done = true;
            }
            trySchedule();
        }

接下来就是trySchedule，接下来就是调用自身run方法，走runAsync（ObserveOnSubscriber），然后无限循环poll直到没有数据，然后onNext runAsync主要注意produced和requested.get()