RxJava2.0----Flowable(被观察者)/Subs

作者: Calllanna | 来源:发表于2017-11-12 17:48 被阅读113次

    一 引言

    这个模式与模式一最大区别在于可以支持背压(Backpressure),什么是背压?我们先看看一个例子:

    Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(ObservableEmitter<String> e) throws Exception {
                   int i = 0;                
                   while (true){
                        e.onNext(“data:”+(i++));
                    }
                 }
               }) .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) throws Exception {
                    Thread.sleep(2000);
                    print(s);
                }
            }); 
    
    结果:
    
    MissingBackpressureException: create: could not emit value due to lack of requests
    

    我们看到结果出现,这个例子最终会产生OOM异常,因为被观察者发射数据的速度大于观察者接收处理数据的速度,造成观察者的调度器中数据缓冲池无限堆积,超出了缓冲池的最大容量,导致OOM
    而此时Flowable的背压策略就很好的解决这个问题。例如我们使用如下方式:

      Flowable.create(new FlowableOnSubscribe<String>() {
                @Override
                public void subscribe(@NonNull FlowableEmitter<String> e) throws Exception {
                     int i = 0;
                    while (true){
                        e.onNext("data:"+(i++));
                    }
                }
            }, BackpressureStrategy.DROP)//超出缓冲池的数据丢弃
                    .subscribeOn(Schedulers.computation())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Subscriber<String>() {
                Subscription subscription;
                @Override
                public void onSubscribe(Subscription s) {
                    subscription = s;
                    subscription.request(1);
                }
                @Override
                public void onNext(String s) {
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    printThread(s);
                    subscription.request(1); //处理完了,在请求数据
                }
                @Override
                public void onError(Throwable t) {
    
                }
                @Override
                public void onComplete() {
    
                }
            });
    

    该背压策略是超出缓冲池的数据被丢弃,而观察者要求处理一个发送我一个数据,那么是否体验到背压策略的强大了呢?是否觉得好多问题都不再是问题了?哈哈,下面就好好给你安利一下Flowable背压策略。

    二.Backpressure的策略

    (1)被观察者和观察者在异步线程的情况下,如果被观察者发射事件的速度大于观察者接收事件的速度,就会产生Backpressure问题。但是同步情况下,Backpressure问题不会存在。
    (2)Backpressure的策略仅仅是调度Subscriber接收事件,并不影响Flowable发送事件。观察者可以根据自身实际情况按需拉取数据,而不是被动接收(也就相当于告诉上游观察者把速度慢下来),最终实现了上游被观察者发送事件的速度的控制,实现了背压的策略。
    (3)Backpressure的策略有5种:ERROR,BUFFER,DROP,LATEST,MISSING
    ERROR策略
    之前在分析Scheduler调度的时候,在观察者线程里有一个数据缓存池SimpleQueue<T>
    queue,queue在ObserveOnObserver()构造函数里实例化一个128个大小的队列:
    queue = new SpscLinkedArrayQueue<T>(bufferSize);,
    用来缓存观察者处理不了暂时缓存下来的数据,缓存池的默认大小为128,即只能缓存128个事件。这个缓存池的大小目前我还没有发现可以直接设置的方法(如果你发现了要及时@我哦)。

    在ERROR策略下,如果缓存池溢出,就会立刻抛出MissingBackpressureException异常。 ERROR即保证在异步操作中,事件累积不能超过128,超过即出现异常。

    Flowable.create(new FlowableOnSubscribe<String>() {
               @Override
               public void subscribe(@NonNull FlowableEmitter<String> e) throws Exception {
                   int i = 0;
                   while (true){
                       e.onNext("data:"+(i++));
                   }
               }
           }, BackpressureStrategy.ERROR)//超出缓冲池的数据128个时,报Error
                   .subscribeOn(Schedulers.computation())
                   .observeOn(Schedulers.newThread())
                   .subscribe(new Subscriber<String>() {
                       @Override
                       public void onSubscribe(Subscription s) {
                           subscription = s;
                           subscription.request(128);
                       }
    
                       @Override
                       public void onNext(String s) {
                           try {
                               Thread.sleep(100);
                           } catch (InterruptedException e) {
                               e.printStackTrace();
                           }
                           printThread(s);
    
                       }
    
                       @Override
                       public void onError(Throwable t) {
                           printThread(t.getMessage());
                       }
    
                       @Override
                       public void onComplete() {
    
                       }
                   });
    输出:
    data:0:-------RxNewThreadScheduler-1
    create: could not emit value due to lack of requests:-------RxNewThreadScheduler-1
    

    BUFFER策略
    该策略是Flowable的默认策略,这里有一点要注意, 很多人认为BUFFER策略就是把RxJava中默认的只能存128个事件的缓存池换成一个大的缓存池,支持存很多很多的数据。其实不是,大小没有改变,依然是128个。看看源码:

    所以使用这个策略,相当于无作为,相当大的概率出现OOM.
    DROP策略
    在DROP策略下,当观察者来不及处理的数据将会被丢弃。即丢弃后面超出缓存区的
    例子:

     Subscription subscription1;
        private void doAct2() {
            //DROP---------------分三次打印第一次0...49,第二次50...99...第三次100...127
            if(subscription1 == null) {
                tx_console.setText("DROP");
                Flowable.create(new FlowableOnSubscribe<String>() {
                    @Override
                    public void subscribe(@NonNull FlowableEmitter<String> e) throws Exception {
                        int i = 0;
                        while (true) {
                            e.onNext("data:" + (i++));
                            if (i == 1000) {
                                e.onComplete();
                                return;
                            }
                        }
                    }
                }, BackpressureStrategy.DROP)//超出缓冲池的数据丢弃,丢弃最新的,保留原始的
                        .subscribeOn(Schedulers.computation())
                        .observeOn(Schedulers.newThread())
                        .subscribe(new Subscriber<String>() {
                            @Override
                            public void onSubscribe(Subscription s) {
                                subscription1 = s;
                                subscription1.request(50);
                            }
    
                            @Override
                            public void onNext(String s) {
                                try {
                                    Thread.sleep(100);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                printThread(s);
                            }
    
                            @Override
                            public void onError(Throwable t) {
    
                            }
    
                            @Override
                            public void onComplete() {
    
                            }
                        });
            }else{
                subscription1.request(50);
            }
        }
    输出:
    data:0
    data:1
    ......
    data:127
    

    LATEST策略
    LATEST策略和DROP策略略同,区别在于LATEST策略会接收最后一个数据,即当观察者来不及处理的数据将会被丢弃。但是会保留最后一个。

    例子:
     Subscription subscription2;
        private void doAct2() {
            //DROP---------------分三次打印第一次0...49,第二次50...99...第三次100...127,999
            if(subscription2 == null) {
                tx_console.setText("DROP");
                Flowable.create(new FlowableOnSubscribe<String>() {
                    @Override
                    public void subscribe(@NonNull FlowableEmitter<String> e) throws Exception {
                        int i = 0;
                        while (true) {
                            e.onNext("data:" + (i++));
                            if (i == 1000) {
                                e.onComplete();
                                return;
                            }
                        }
                    }
                }, BackpressureStrategy.DROP)//超出缓冲池的数据丢弃,丢弃最新的,保留原始的
                        .subscribeOn(Schedulers.computation())
                        .observeOn(Schedulers.newThread())
                        .subscribe(new Subscriber<String>() {
                            @Override
                            public void onSubscribe(Subscription s) {
                                subscription2 = s;
                                subscription2.request(50);
                            }
    
                            @Override
                            public void onNext(String s) {
                                try {
                                    Thread.sleep(100);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                printThread(s);
                            }
    
                            @Override
                            public void onError(Throwable t) {
    
                            }
    
                            @Override
                            public void onComplete() {
    
                            }
                        });
            }else{
                subscription2.request(50);
            }
        }
    输出:
    data:0
    data:1
    ......
    data:127
    data:999
    

    MISSING策略
    在MISSING策略下,没有缓冲池,接收第一个数据以后,后面的都丢弃
    例子:

     Flowable.create(new FlowableOnSubscribe<String>() {
                @Override
                public void subscribe(@NonNull FlowableEmitter<String> e) throws Exception {
                    int i = 0;
                    while (true){
                        e.onNext("data:"+(i++));
                    }
                }
            }, BackpressureStrategy.MISSING)//没有缓冲池,接收第一个数据以后,后面的都丢弃
                    .subscribeOn(Schedulers.computation())
                    .observeOn(Schedulers.newThread())
                    .subscribe(new Subscriber<String>() {
                        @Override
                        public void onSubscribe(Subscription s) {
                            subscription = s;
                            subscription.request(128);
                        }
    
                        @Override
                        public void onNext(String s) {
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            printThread(s);
    
                        }
    
                        @Override
                        public void onError(Throwable t) {
    
                        }
    
                        @Override
                        public void onComplete() {
    
                        }
                    });
    输出:
     data:0:-------RxNewThreadScheduler-1
    

    三.Flowable的创建

    Flowable的创建是和Observable一样,可以通过操作函数,例如:create(),defer(),just(),from(),rang(),timer(),interval()等方法来创建。
    也可以直接通过Observable转换创建:
    Observable.toFlowable()
    具体使用例子可以参考上一篇Observable的创建。

    四.Subscriber的创建

    在上面的例子中,我们可以看到Subscriber的创建,也需要重写四个方法onSubscribe(Subscription s),onNext(Object s), onError(Throwable t),onComplete()
    与Observer不同的是onSubscribe中回调的是Subscription,不是Disposable ,而Subscription比
    Disposable 多了一个request(),其他方法功能一样。 当调用了request()方法后,被观察者便发送对应数量的事件给观察者接收并处理。 Flowable在设计的时候采用了一种新的思路,也就是响应式拉取的模式,即观察者要求多少,被观察者便传给你多少。

    相关文章

      网友评论

      本文标题:RxJava2.0----Flowable(被观察者)/Subs

      本文链接:https://www.haomeiwen.com/subject/wginmxtx.html