转载

Rxjava2(二)、五种观察者模式及背压

Android进阶系列之第三方库知识点整理。

知识点总结，整理也是学习的过程，如有错误，欢迎批评指出。

上一篇： Rxjava2(一)、基础概念及使用

Rxjava2(二)、五种观察者模式及背压

直接开整，上一篇基础概念里面说了， rxjava2 扩展于观察者模式，我们上篇的只是简单的介绍了用 Observable 来创建使用，其实 rxjava2 给我们提供了五种观察者模式的创建方式。

Rxjava2(二)、五种观察者模式及背压

1、Observable 和 Observer

能够发射0或n个数据，并以成功或错误事件终止，在第一篇中已经举例说明了，这里就不再详细说明。

2、Flowable 和 Subscriber

能够发射0或n个数据，并以成功或错误事件终止。 支持 背压 ，可以控制数据源发射的速度。

我们看到 Observable 和 Flowable 这两个的区别就是后者支持背压，那么何为背压？

2.1、什么是背压

背压是一种现象，简单来说就是在 异步操作 中，上游发送数据速度快于下游处理数据的速度，下游来不及处理， Buffer 溢出 ，导致事件阻塞，从而引起的各种问题，比如事件丢失，OOM等。

在 rxjava1 中并不支持背压，当出现事件阻塞时候，会直接抛出 MissingBackpressureException 异常，但是在 rxjava2 中，提供了 Flowable 来创建被观察者，通过 Flowable 来处理背压问题，我们可以简单通过demo分析。

Rxjava2(二)、五种观察者模式及背压 
A：我们上游模拟循环发送数据。
B：线程切换，异步操作。
C：下游每隔一秒获取数据。

我们 Observable 创建，来模拟了背压这个现象，我们在上游模拟无限循环的发送数据，下游每次都休眠一秒再获取数据，这样肯定会造成我们前面提的问题，就是上游发送太他丫的快了，下游根本处理不过来，我们先看结果。

Rxjava2(二)、五种观察者模式及背压

看日志，打印结果停留在了13就没有继续打印了？同时可以看到程序已经崩了，是因为在 rxjava2 中， Observable 并不支持背压操作，遇到背压问题，它并不会报错，也不会抛 MissingBackpressureException 异常，但是内存会一直飙高，最后导致内存不足程序直接挂掉。

Rxjava2(二)、五种观察者模式及背压

可以看到内存一直在往上飙，针对背压这种现象， rxjava2 中提出用 Flowable 来处理。

下面由浅入深，慢慢揭开 Flowable 的神秘面纱。

我们先用 Flowable 创建一个基本的demo： 

       Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                emitter.onNext("事件一");
                LogUtil.d(TAG + "--subscribe  发送事件一");
                emitter.onNext("事件二");
                LogUtil.d(TAG + "--subscribe  发送事件二");
                emitter.onNext("事件三");
                LogUtil.d(TAG + "--subscribe  发送事件三");
                emitter.onNext("事件四");
                LogUtil.d(TAG + "--subscribe  发送事件四");
                emitter.onComplete();
                LogUtil.d(TAG + "--subscribe  发送完成");
            }
        }, BackpressureStrategy.ERROR) // 这里需要传入背压策略，跟线程池里面饱和策略类似，当缓存区存满时候采取的处理策略
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()) // 线程切换，异步操作
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        // 决定观察者能接收多少个事件，多余事件放入缓存区
                        // Flowable 默认缓存区大小为128，即最大能存放128个事件
                        s.request(3);
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t.getLocalizedMessage());
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

可以看到 Flowable 创建和 Observable 基本差不多，只是在 create 方法中多传入 BackpressureStrategy.ERROR 这么一个背压策略，这个后面会详讲。

在 onSubscribe 的回调中，参数变成了 Subscription ，我们可以通过这个参数，让观察者自己设置要接收多少个事件，如果发送的事件大于观察者设置接收的事件，多余事件将会存入 Flowable 缓存区中。

Flowable 缓存区队列大小只能存放128个事件，如果超过，就会报异常。

结果：

Rxjava2(二)、五种观察者模式及背压

发送四个事件，观察者通过 Subscription.request(3) 设置只接收三个事件，所以下游只接收三个，剩下一个放入 Flowable 缓存区中。

如果我们观察者不设置 Subscription.request(x) ,即不接收事件，被观察者仍然会发送事件，并存入缓存区中，观察者可以动态调用 Subscription.request(x) 方法来获取事件。 

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {

                for (int x = 0; x <= 10; x++) {
                    LogUtil.d(TAG + "--subscribe  发送了" + x + "个事件");
                    emitter.onNext(x + "事件");
                }
            }
        }, BackpressureStrategy.ERROR) 
                // 线程切换，异步操作
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        subscription = s;
                        // s.request(3);  这里不指定观察者接收事件个数
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t.getLocalizedMessage());
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

动态获取

findViewById(R.id.bt_get_event).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                if (subscription != null) {
                    LogUtil.d(TAG + "--onClick");
                    subscription.request(4);
                }
            }
        });
复制代码

可以看到我们观察者一开始并没有指定接收多少个事件，而是通过外接点击事件，来动态设置接收事件个数，我们看结果，当点击触发后，我们收到了最先存入队列的四个事件。

结果：

Rxjava2(二)、五种观察者模式及背压

2.2、背压策略

我们前面提到， Flowable 默认的缓存区队列大小为128，即只能存放上游发送的128个事件，如果上游发送的事件超过128，就需要我们指定相应的背压策略来做不同的处理， BackpressureStrategy 为我们提供了五种背压策略。

Rxjava2(二)、五种观察者模式及背压

整理如下：

策略 作用
MISSING 当缓存区大小存满（128），被观察者仍然继续发送下一个事件时，抛出异常 MissingBackpressureException , 提示缓存区满了
ERROR 当缓存区大小存满（128）（默认缓存区大小128），被观察者仍然继续发送下一个事件时，直接抛出异常 MissingBackpressureException
BUFFER 当缓存区大小存满（128），被观察者仍然继续发送下一个事件时，缓存区大小设置无限大, 即被观察者可无限发送事件，但实际上是存放在缓存区
DROP 当缓存区大小存满，被观察者仍然继续发送下一个事件时， 超过缓存区大小（128）的事件会被全部丢弃
LATEST 当缓存区大小存满，被观察者仍然继续发送下一个事件时，只保存最新/最后发送的事件， 其他超过缓存区大小（128）的事件会被全部丢弃

2.2.1、MISSING

当缓存区大小存满（128），被观察者仍然继续发送下一个事件时，抛出异常 MissingBackpressureException , 提示缓存区满了 

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // 发送129个事件，模拟超出缓存区
                for (int x = 0; x < 129; x++) {
                    emitter.onNext(x + "事件");
                    LogUtil.d(TAG + "--subscribe  发送了" + x + "个事件");
                }
            }
        }, BackpressureStrategy.MISSING) // 使用BackpressureStrategy.MISSING背压策略
                // 线程切换，异步操作
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        s.request(Integer.MAX_VALUE);
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t);
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

我们使用BackpressureStrategy.MISSING背压策略，观察者接收request(Integer.MAX_VALUE)，此值也为推荐值。

结果：

Rxjava2(二)、五种观察者模式及背压

我们看到，当发送了128个事件后，再发送第129个事件时候，抛了 MissingBackpressureException 异常，而且我们设置了观察者接收也未接收到数据，说明是先存入缓存区队列，再发送，当缓存区中抛异常后，就停止了 onNext() 事件，我们可以验证一下，当设置被观察者发送 128 事件。 

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // *******  发送128个事件  ********
                for (int x = 0; x < 128; x++) {
                    emitter.onNext(x + "事件");
                    LogUtil.d(TAG + "--subscribe  发送了" + x + "个事件");
                }
            }
        }, BackpressureStrategy.MISSING)
                // 线程切换，异步操作
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        s.request(Integer.MAX_VALUE);
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t);
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

就是在上面demo的基础上，改了发送的事件个数，上游发送128个事件，刚好为缓存区大小，并不抛异常。

结果：

Rxjava2(二)、五种观察者模式及背压

我们看到程序没有抛异常，并且正常打印了缓存区中的128个数据（从0开始），可以印证两点

1、缓存区大小确实为128

2、先存入缓存区后再获取（如果异常， onNext 直接不调用）

2.2.2、ERROR

当缓存区大小存满（128）（默认缓存区大小128），被观察者仍然继续发送下一个事件时，直接抛出异常 MissingBackpressureException 

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // 发送129个事件，模拟超出缓存区
                for (int x = 0; x < 129; x++) {
                    emitter.onNext(x + "事件");
                    LogUtil.d(TAG + "--subscribe  发送了" + x + "个事件");
                }
            }
        }, BackpressureStrategy.ERROR)
                // 线程切换，异步操作
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        s.request(Integer.MAX_VALUE);
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t);
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

使用 BackpressureStrategy.ERROR 背压策略

结果：

Rxjava2(二)、五种观察者模式及背压

跟Missing一样，直接抛了 MissingBackpressureException 异常且下游未接收到数据，同理，如果上游发送数据小于等于128，正常发送和接收。

2.2.3、BUFFER

当缓存区大小存满（128），被观察者仍然继续发送下一个事件时，缓存区大小设置无限大, 即被观察者可无限发送事件，但实际上是存放在缓存区。

       Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // 发送129个事件，模拟超出缓存区
                for (int x = 0; x < 129; x++) {
                    emitter.onNext(x + "事件");
                    LogUtil.d(TAG + "--subscribe  发送了" + x + "个事件");
                }
            }
        }, BackpressureStrategy.BUFFER)
                // 线程切换，异步操作
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        s.request(Integer.MAX_VALUE);
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t);
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

使用 BackpressureStrategy.BUFFER 背压策略

更改缓存区大小，不做限制。

结果：

Rxjava2(二)、五种观察者模式及背压

可以看到，我们发送的129个事件全部发送且接收到了。

2.2.4、DROP

当缓存区大小存满，被观察者仍然继续发送下一个事件时， 超过缓存区大小（128）的事件会被全部丢弃

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // 发送129个事件，模拟超出缓存区
                for (int x = 0; x < 129; x++) {
                    emitter.onNext(x + "事件");
                    LogUtil.d(TAG + "--subscribe  发送了" + x + "个事件");
                }
            }
        }, BackpressureStrategy.DROP)
                // 线程切换，异步操作
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        s.request(Integer.MAX_VALUE);
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t);
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

使用 BackpressureStrategy.DROP 背压策略

丢掉大于缓存区的事件。

结果：

Rxjava2(二)、五种观察者模式及背压

结果很明了，并没有抛异常同时也正常打印了，但是超过缓存区的那个事件被抛弃，并没有获取到。

2.2.5、LATEST

当缓存区大小存满，被观察者仍然继续发送下一个事件时，只保存最新/最后发送的事件， 其他超过缓存区大小（128）的事件会被全部丢弃

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // 发送150个事件
                for (int x = 0; x < 150; x++) {
                    emitter.onNext(x + "事件");
                    LogUtil.d(TAG + "--subscribe  发送了" + x + "个事件");
                }
            }
        }, BackpressureStrategy.LATEST)
                // 线程切换，异步操作
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtil.d(TAG + "--onSubscribe");
                        s.request(Integer.MAX_VALUE);
                    }

                    @Override
                    public void onNext(String s) {
                        LogUtil.d(TAG + "--onNext  接收到:" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.d(TAG + "--onError  error=" + t);
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.d(TAG + "--onComplete");
                    }
                });
复制代码

使用 BackpressureStrategy.LATEST 背压策略

发送了150个事件

当超出128时，会保存最新的一个事件，即会接收129个事件。

结果：

Rxjava2(二)、五种观察者模式及背压

我们可以看到，观察者端接收到129个数据，分别为缓存区内数据，加上最新/最后一条数据，中间数据均被丢弃。

2.3、同步情况下`Flowable`

前面说过，背压前提是异步操作下，在同步下，我们并不会有背压一说，因为在同一个线程，发送数据后总是要等下游处理了才会发送第二条数据， 不会存在缓冲区 ，如下： 

       Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                LogUtil.d(TAG + "--subscribe  发送事件一");
                emitter.onNext("事件一");
                LogUtil.d(TAG + "--subscribe  发送事件二");
                emitter.onNext("事件二");
                LogUtil.d(TAG + "--subscribe  发送事件三");
                emitter.onNext("事件三");
                LogUtil.d(TAG + "--subscribe  发送完成");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribe(new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                LogUtil.d(TAG + "--onSubscribe");
                s.request(3);
            }

            @Override
            public void onNext(String s) {
                LogUtil.d(TAG + "--onNext  接收到:" + s);
            }

            @Override
            public void onError(Throwable t) {
                LogUtil.d(TAG + "--onError  error=" + t);
            }

            @Override
            public void onComplete() {
                LogUtil.d(TAG + "--onComplete");
            }
        });
复制代码

结果：

Rxjava2(二)、五种观察者模式及背压

可以看到，事件都是顺序执行，发送一条接收一条，然后再执行下一条。

但是，我们可能会遇到这个一个情况，当上游发送了四条数据，但是下游只接收三条？我们改一下demo如下：

       Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                LogUtil.d(TAG + "--subscribe  发送事件一");
                emitter.onNext("事件一");
                LogUtil.d(TAG + "--subscribe  发送事件二");
                emitter.onNext("事件二");
                LogUtil.d(TAG + "--subscribe  发送事件三");
                emitter.onNext("事件三");
                LogUtil.d(TAG + "--subscribe  发送事件四");
                emitter.onNext("事件四");
                LogUtil.d(TAG + "--subscribe  发送完成");
                emitter.onComplete();
            }
        }, BackpressureStrategy.ERROR).subscribe(new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                LogUtil.d(TAG + "--onSubscribe");
                s.request(3);

            }

            @Override
            public void onNext(String s) {
                LogUtil.d(TAG + "--onNext  接收到:" + s);
            }

            @Override
            public void onError(Throwable t) {
                LogUtil.d(TAG + "--onError  error=" + t);
            }

            @Override
            public void onComplete() {
                LogUtil.d(TAG + "--onComplete");
            }
        });
复制代码

可以看到，被观察者发送了四个事件，但是观察者只接收了三条。

结果：

Rxjava2(二)、五种观察者模式及背压

可以看到，同样抛了 MissingBackpressureException 异常

这里可以使用BUFFER的背压策略来处理，但是我们为了说明观察者反向控制被观察者，我们采用如下方案：

        Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> emitter) throws Exception {
                // 通过emitter.requested()获取观察者设置的接收的事件数目
                long requested = emitter.requested();
                LogUtil.d(TAG + "--subscribe 观察者设置接收的事件数目:" + requested);

                for (int x = 0; x < requested; x++) {
                    LogUtil.d(TAG + "--subscribe  发送事件" + x);
                    emitter.onNext("发送事件" + x);
                }
                LogUtil.d(TAG + "--subscribe  发送完成");
                emitter.onComplete();
            }
        }, BackpressureStrategy.BUFFER).subscribe(new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                LogUtil.d(TAG + "--onSubscribe");
                // 设置观察者接收事件数目为3
                s.request(3);

            }

            @Override
            public void onNext(String s) {
                LogUtil.d(TAG + "--onNext  接收到:" + s);
            }

            @Override
            public void onError(Throwable t) {
                LogUtil.e(TAG + "--onError  error=" + t);
            }

            @Override
            public void onComplete() {
                LogUtil.d(TAG + "--onComplete");
            }
        });
复制代码

我们在 subscribe 中通过 emitter.requested() 获取观察者中设置的接收事件数目，来动态的发送数据，这样就避免了上下游数据不同步问题。

结果：

Rxjava2(二)、五种观察者模式及背压

2.4、使用操作符时背压处理

我们前面都是通过create来创建 Flowable ，可以在 Create 第二个参数中传入相应的背压策略， Flowable 所有的操作符都支持背压，但是通过操作符创建的背压策略默认为BackpressureStrategy.ERROR，我们可以通过 

onBackpressureBuffer()
onBackpressureDrop()
onBackpressureLatest()

三种方式来指定相应的背压策略。

        Flowable.interval(1, TimeUnit.MILLISECONDS)
                .observeOn(Schedulers.io())
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        Log.d(TAG, "onSubscribe");
                        subscription = s;
                        s.request(Long.MAX_VALUE); //默认可以接收Long.MAX_VALUE个事件
                    }

                    @Override
                    public void onNext(Long aLong) {
                        LogUtil.i(TAG + "--onNext  aLong=" + aLong);
                        try {
                            // 延时一秒接收
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtil.e(TAG + "--onError  error=" + t);
                    }

                    @Override
                    public void onComplete() {
                        LogUtil.i(TAG + "--onComplete");
                    }
                });
复制代码

这里我们通过 interval 来创建 Flowable ，可以看到下游每一毫秒发送一条数据，下游一秒处理一条，上游明显快于下游，处理不过来数据放入缓存池中，当缓存池中队列满时，就会抛异常，因为其默认的背压策略为 BackpressureStrategy.ERROR

结果：

Rxjava2(二)、五种观察者模式及背压

我们可以通过 onBackpressureXXX 其指定相应的背压策略。

Rxjava2(二)、五种观察者模式及背压

结果：

Rxjava2(二)、五种观察者模式及背压

当我们指定背压策略为BUFFER后，可以看到并没有异常抛出，程序一直在打印输出。

3、Single和SingleObserver

只发射单个数据或错误事件。

        Single.create(new SingleOnSubscribe<String>() {
            @Override
            public void subscribe(SingleEmitter<String> emitter) throws Exception {
                // 只能发送onSuccess或者onError，发射多条数据，只接受第一条
                emitter.onSuccess("Success");
                emitter.onError(new NullPointerException(""));
            }
        }).subscribe(new SingleObserver<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                LogUtil.d(TAG + "--onSubscribe");
            }

            @Override
            public void onSuccess(String s) {
                LogUtil.d(TAG + "--onSuccess  s=" + s);
            }

            @Override
            public void onError(Throwable e) {
                LogUtil.e(TAG + "--onError  error=" + e.getMessage());
            }
        });
复制代码

SingleEmitter 发射器只能发送一条 onSuccess 或者 onError 数据，如果发射器发射多条数据，观察者只能接收到第一条数据。

结果：

Rxjava2(二)、五种观察者模式及背压

4、Completable和CompletableObserver

不发射数据，只处理 onComplete 和 onError 事件。

image-20191230175048374

方法 onComplete 与 onError 只可调用一个，同时调用，第一个生效。

5、Maybe和MaybeObserver

能够发射0或者1个数据，要么成功，要么失败。有点类似于Optional。

Rxjava2(二)、五种观察者模式及背压

onSuccess 方法一次订阅只能发送一次。

方法 onComplete 与 onError 只可调用一个，同时调用，第一个生效。

原文  https://juejin.im/post/5e09e2c2f265da33cf1aea3e
正文到此结束
所属分类： 编程技术 Java

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 Maven教程 hibernate教程 Docker 教程 memcached教程 Quartz指南 Hive教程 Ant教程 ANTLR教程 SpringCloud java实例教程 springcloud-demo XStream教程 Hazelcast教程 Elastic-Job-Lite 深入浅出MyBatis SVN教程 ibaties教程 rabittmq教程 solr教程 Hadoop教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 java-demo Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 Spring Cloud Consul实现选举机制
  2. 2 Spring Boot集成ShedLock实现分布式定时任务
  3. 3 利用oss进行数据库和网站图片备份
  4. 4 Spring Cloud Stream实现数据流处理
  5. 5 Python3访问MySQL数据库快速入门Demo
  6. 6 Github开源项目作者可以免费申请 JetBrains 全家桶
  7. 7 Spring Cloud Vault快速入门Demo
  8. 8 Spring Cloud Gateway快速入门Demo
  9. 9 Spring Cloud Contract快速入门Demo
  10. 10 Spring Cloud Consul快速入门Demo
网站信息
  • 文章总数：82,714 篇
  • 文件总数：284,287 个
  • 标签总数：2,393 个
  • 分类总数：85 个
  • 留言数量：2,560 条
  • 在线人数：674 人
  • 运行天数：4,408天
  • 最后更新：2024年11月21日23点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG