前言

上一节我们学习了使用Observable解决上、下游发射事件速度不平衡的问题，之所以学习 Observable，是因为Observable有很多的使用场景，而这一节我们要学习一个新的操作符 —— Flowable操作符，这个操作符可以解决绝大部分的问题，但是Observable、Flowable各有应用场景，也各有优势和缺点。

1. Flowable

---

Flowable如下图所示：

Flowable.png

• Observable的上、下游分别是Observable、Observer，建立连接是subscribe()；
• Flowable的上、下游分别是Flowable、Subscriber，建立连接是subscribe()；

Flowable最基本用法如下：

/**
     * Flowable最基本用法
     */
    public static void demo1(){
        // 创建一个上游：Flowable
        Flowable<Integer> upStream = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.e("TAG" , "emit 1") ;
                emitter.onNext(1);
                Log.e("TAG" , "emit 2") ;
                emitter.onNext(2);
                Log.e("TAG" , "emit 3") ;
                emitter.onNext(3);

                Log.e("TAG" , "emit complete") ;
                emitter.onComplete();

            }   // 参数BackpressureStrategy.ERROR作用：
                // 用来选择背压，用于解决上下游发射数据速度不平衡问题，如果速度不一致，
                // 直接抛异常MissingBackpressureException
        } , BackpressureStrategy.ERROR) ;


        // 创建一个下游：Subscriber
        Subscriber<Integer> downStream = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.e("TAG" , "subscribe") ;
                s.request(Long.MAX_VALUE);  
            }

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

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

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

        // 建立连接
        upStream.subscribe(downStream) ;
    }

运行结果如下：

TAG: subscribe
TAG: emit 1
TAG: next -> 1
TAG: emit 2
TAG: next -> 2
TAG: emit 3
TAG: next -> 3
TAG: emit complete
TAG: complete

Flowable写法与Observable稍有区别：

• 第一个区别是：Flowable上游中新增一个参数，BackpressureStrategy.ERROR，用来选择背压，用于是解决上下游发射数据速度不平衡问题，如果速度不一致，直接抛出异常MissingBackpressureException；
• 第二个区别是：在下游的onSubscribe()方法中，接收的是Subscription ，而不是Observable中的Disposable，这两个作用都是用于切断水管，调用Subscription.cancel()、调用Disposable.dispose()，不同的是在 Subscription中需要调用 request()，即就是代码中的

s.request(Long.MAX_VALUE); 

• 如果不添加 s.request(Long.MAX_VALUE); ，直接报错MissingBackpressureException，示例代码如下：

    /**
     * Flowable用法：在onSubscribe()中不加 s.request(Long.MAX_VALUE);
     */
    public static void demo2(){
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.e("TAG" , "emit 1") ;
                emitter.onNext(1);
                Log.e("TAG" , "emit 2") ;
                emitter.onNext(2);
                Log.e("TAG" , "emit 3") ;
                emitter.onNext(3);

                Log.e("TAG" , "emit complete") ;
                emitter.onComplete();
            }
        } , BackpressureStrategy.ERROR).subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.e("TAG" , "subscribe") ;
            }

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

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

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

    }

图片.png

可以看到，在上游发送第一个事件后，下游直接抛出著名异常MissingBackpressureException异常，且下游并没有收到任何其余的事件，上、下游在同一个线程，是同步关系，按道理来讲，上游每次发送事件都应该等下游处理完事件后，才会继续发送事件，下边我们看异步线程的情况。

2. Flowable让上下游在异步线程中执行

代码如下：

/**
     * Flowable：
     *          让上、下游处于异步线程中，也就是说让上游在子线程中执行，下游在主线程中执行
     */
    public static void demo3(){
        Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.e("TAG" , "emit 1") ;
                emitter.onNext(1);
                Log.e("TAG" , "emit 2") ;
                emitter.onNext(2);
                Log.e("TAG" , "emit 3") ;
                emitter.onNext(3);

                Log.e("TAG" , "emit complete") ;
                emitter.onComplete();
            }
        } , BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .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 -> " + t) ;
                    }

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

运行结果如下：

TAG: subscribe
TAG: emit 1
TAG: emit 2
TAG: emit 3
TAG: emit complete

可以看到，上游发送所有事件，但是下游一个都没有收到

• 在onSubscribe()方法中调用的s.request(Long.MAX_VALUE); 可以把request()方法当做一种能力，当成下游处理事件的能力，下游能处理几个，就告诉上游我要几个，可以解决上游一次性发射全部数据到下游，导致OOM，可以解决上一节学习的两种处理方式的缺陷：
  1>：取样事件会导致事件丢失；
  2>：减速导致性能丢失；
  而调用s.request(Long.MAX_VALUE)方法 这种 方式既解决了事件丢失又解决了性能丢失；

3. 认识 request()方法

---

1>：对于上、下游二者在同一个线程中，上游发射第一个事件，下游直接抛出异常MissingBackpressureException异常，是因为下游没有调用request()方法，表示上游认为下游没有处理事件的能力，上游发送事件，下游必须在onSubscribe()方法中调用request()方法，表示下游有处理事件的能力，比如 request(3)或者request(Long.MAX_VALUE)都是可以的；

2>：对于上、下游二者没有在同一个线程中，即异步操作，上游可以正确发送所有事件，因为在 Flowable中默认有一个 128的水缸，当上下游在不同的线程中工作时，上游会先把事件发送到这个水缸中，所以说，即使下游没有调用 request()方法，但是水缸中保存着上游发射的事件，只有当下游调用request()方法时，才从水缸中取出事件发送给 下游；

注意：

1>：Flowable中的 水缸大小只有 128，如果是129，直接抛MissingBackpressureException异常；
2>：我们这里把上游发射的事件全部存储到 水缸中，下游一个都没有消费，只要下游快速消费一个，就不会OOM，如果下游10秒之后再消费也会OOM；