RxJava2——Observable基本用例

最基本的調用

<code>Observable.create(new ObservableOnSubscribe<string>() {
            @Override
            public void subscribe(ObservableEmitter<string> emitter) throws Exception {
                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onNext("4");
                emitter.onNext("5");
                emitter.onComplete();

                XLog.d("subscribe over");

            }
        }).subscribe(new Observer<string>() {
            @Override
            public void onSubscribe(Disposable d) {
                XLog.d("onSubscribe ");

            }

            @Override
            public void onNext(String s) {

                XLog.d("onNext " + s);
            }

            @Override
            public void onError(Throwable e) {
                XLog.d("onError ");

            }

            @Override
            public void onComplete() {
                XLog.d("onComplete ");

            }
        });

/<string>/<string>/<string>/<code>

輸出結果：基本按照順序輸出

onSubscribeonNext 1onNext 2onNext 3onNext 4onNext 5onCompletesubscribe over

onComplete/onError放中間

<code>emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
//emitter.onComplete();
emitter.onError(new Exception());
emitter.onNext("4");
emitter.onNext("5");
/<code>

輸出結果：當中間有 onCompelete/onError 時，Observable會繼續發送，但Obsever不會再接收事件了

onSubscribeonNext 1onNext 2onNext 3//onCompleteonErrorsubscribe over

連續兩次調用onComplete

<code>                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onComplete();
                emitter.onComplete();
                emitter.onNext("4");
                emitter.onNext("5");
/<code>

輸出結果：和單次調用 onComplete一樣，如果放中間，Observer將不會再接收事件，且無異常

onSubscribeonNext 1onNext 2onNext 3onCompletesubscribe over

連續兩次調用onError

<code>                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onError(new Exception());
                emitter.onError(new Exception());
                emitter.onComplete();
                emitter.onNext("4");
                emitter.onNext("5"); 

                XLog.d("subscribe over");
/<code>

輸出結果：收到onError事件便截止

onSubscribeonNext 1onNext 2onNext 3onError

除此之外，連續 2 次調用 onError(),還報出了一個 UndeliverableException

<code>09-22 12:45:18.181 8830-8830/www.ly.com.rxjavademo W/System.err: io.reactivex.exceptions.UndeliverableException: java.lang.Exception
        at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
        at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:74)
        at www.ly.com.rxjavademo.MainActivity$2.subscribe(MainActivity.java:73)
        at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
        at io.reactivex.Observable.subscribe(Observable.java:12005)
        at www.ly.com.rxjavademo.MainActivity.onCreate(MainActivity.java:81)
/<code>

onError 和 onComplete 同時出現

<code>                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onComplete();
                emitter.onError(new Exception());
                emitter.onNext("4");
                emitter.onNext("5");
/<code>

輸出結果：輸出到onComplete截止

onSubscribeonNext 1onNext 2onNext 3onComplete

除此之外，同時也收到了 UndeliverableException 而終止進程

<code>09-22 13:10:32.738 14353-14353/www.ly.com.rxjavademo W/System.err: io.reactivex.exceptions.UndeliverableException: java.lang.Exception
        at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
09-22 13:10:32.739 14353-14353/www.ly.com.rxjavademo W/System.err:     at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:74)
        at www.ly.com.rxjavademo.MainActivity$2.subscribe(MainActivity.java:73)
        at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
        at io.reactivex.Observable.subscribe(Observable.java:12005)
        at www.ly.com.rxjavademo.MainActivity.onCreate(MainActivity.java:81)
/<code>

但如果把 onError 放在 onComplete 前面呢？則輸出結果為：

onSubscribeonNext 1onNext 2onNext 3onErrorsubscribe over

說明onError 之前不能有 onComplete / onError 事件發生。

指定調度線程

<code>Observable.create(new ObservableOnSubscribe<string>() {
            @Override
            public void subscribe(ObservableEmitter<string> emitter) throws Exception {
                emitter.onNext("1");
                emitter.onNext("2");
                emitter.onNext("3");
                emitter.onNext("4");
                emitter.onNext("5");
                XLog.d("subscribe over");

            }
        })
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<string>() {
            @Override
            public void onSubscribe(Disposable d) {
                XLog.d("onSubscribe ");

            }

            @Override
            public void onNext(String s) {

                XLog.d("onNext " + s);
            }

            @Override
            public void onError(Throwable e) {
                XLog.d("onError ");

            }

            @Override
            public void onComplete() {
                XLog.d("onComplete ");

            }
        });
/<string>/<string>/<string>/<code>

輸出結果：被觀察者在新的線程中執行發送事件，觀察者在Android主線程中接收事件

Thread: mainonSubscribeThread: RxNewThreadScheduler-1subscribe overThread: mainonNext 1Thread: mainonNext 2Thread: mainonNext 3Thread: mainonNext 4Thread: mainonNext 5

壓力測試

被觀察者限制快速發送數據，觀察察者每休息 5 秒鐘處理一個 onNext 事件

<code>Observable.create(new ObservableOnSubscribe<string>() {
            @Override
            public void subscribe(ObservableEmitter<string> emitter) throws Exception {
                int i = 0;
                for (;;) {
                    emitter.onNext("send " + (i++));
                }

            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<string>() {
            @Override
            public void onSubscribe(Disposable d) {
                XLog.d("onSubscribe ");

            }

            @Override
            public void onNext(String s) {

                XLog.d("onNext " + s);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void onError(Throwable e) {
                XLog.d("onError ");

            }

            @Override
            public void onComplete() {
                XLog.d("onComplete ");
 

            }
        });
/<string>/<string>/<string>/<code>

輸出結果：內存溢出！！！在RxJava2中Observable是不支持背壓的

<code>09-22 13:36:52.613 20160-20169/www.ly.com.rxjavademo W/art: Throwing OutOfMemoryError "Failed to allocate a 28 byte allocation with 1864 free bytes and 1864B until OOM; failed due to fragmentation (required continguous free 4096 bytes for a new buffer where largest contiguous free 0 bytes)" (recursive case)

/<code>

<code>09-22 13:36:52.613 20160-20169/www.ly.com.rxjavademo W/art: Throwing OutOfMemoryError "Failed to allocate a 28 byte allocation with 1864 free bytes and 1864B until OOM; failed due to fragmentation (required continguous free 4096 bytes for a new buffer where largest contiguous free 0 bytes)" (recursive case)

/<code>

總結

1. 正常情況下事件會按照預期逐個由被觀察者發送給觀察者
2. 理論上來說，onComplete 和 onError 是互斥的，我們應該控制其只發送一個
3. onComplete / onError 一旦發送，被觀察者還會繼續發送事件，但觀察者將不會再接收到任何事件
4. onComplete 可以發送多次，以收到第一次為結束，且不會有異常
5. onError 只能被髮送一次，並且不能在有 onComplete 之後再發送 onError ，否則會報 UndeliverableException
6. 被觀察者發送事件的速度遠遠超過觀察者處理的速度，最終會造成 OOM。原因是 Observable 不支持背壓。

閱讀更多 仰簡 的文章

關鍵字: 背壓 XLog 用例