1.Flowable(可觀察者)

   private void test5(){
        Log.d(TAG, "test5執行了");;
        Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(final FlowableEmitter<Integer> emitter) throws Exception {
                emitter.setCancellable(new Cancellable() {
                    @Override
                    public void cancel() throws Exception {
                        Log.d(TAG, "取消訂閱了噢噢噢噢");
                    }
                });
                if(callBack == null){
                    callBack=new CallBack() {
                        @Override
                        public void test(String msg) {
                            Log.d(TAG, "發射了訊息");
                            emitter.onNext(1);
                        }
                    };
                }

            }
        }, BackpressureStrategy.BUFFER); //增加了一個引數

        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                //  s.request(20);//策略BUFFER ,消費能力20 結果只收到20條，但是資料依然在發射
                // s.request(0);//策略BUFFER ,消費能力0 結果收不到資料,但是資料依然在發射
                s.request(129);
                //策略DROP,消費能力20,結果只收到連續的20條,但是資料依然在發射
                //策略DROP,消費能力10000,結果資料128條之後，發現直接拋到844了，這說明預設緩衝區滿了之後
                //129條沒有能進入到緩衝區，為什麼呢，因為緩衝區滿了，庫存就那麼大，根據DROP策略，只能丟棄
                //簡單的說,DROP可能導致新產生的事件被丟棄，收到的事件可能會斷層
            }
            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);

            }
            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
        };
        /**
         * .doOnCancel(new Action() {
        @Override
        public void run() throws Exception {
        Log.d(TAG, "取消註冊了");
        }
        })
         */

        flowable.doOnCancel(new Action() {
            @Override
            public void run() throws Exception {
                Log.d(TAG, "取消註冊了");
            }
        })
                .doOnSubscribe(new Consumer<Subscription>() {
            @Override
            public void accept(Subscription subscription) throws Exception {
                Log.d(TAG, "您訂閱了");
                mSubscription=subscription;
            }
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(subscriber);
}