RxJava2——Observable基本用例
最基本的呼叫
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 "); } });
輸出結果:基本按照順序輸出
onSubscribeonNext 1onNext 2onNext 3onNext 4onNext 5onCompletesubscribe over
onComplete/onError放中間
emitter.onNext("1"); emitter.onNext("2"); emitter.onNext("3"); //emitter.onComplete(); emitter.onError(new Exception()); emitter.onNext("4"); emitter.onNext("5");
輸出結果:當中間有 onCompelete/onError 時,Observable會繼續傳送,但Obsever不會再接收事件了
onSubscribeonNext 1onNext 2onNext 3//onCompleteonErrorsubscribe over
連續兩次呼叫onComplete
emitter.onNext("1"); emitter.onNext("2"); emitter.onNext("3"); emitter.onComplete(); emitter.onComplete(); emitter.onNext("4"); emitter.onNext("5");
輸出結果:和單次呼叫 onComplete一樣,如果放中間,Observer將不會再接收事件,且無異常
onSubscribeonNext 1onNext 2onNext 3onCompletesubscribe over
連續兩次呼叫onError
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");
輸出結果:收到onError事件便截止
onSubscribeonNext 1onNext 2onNext 3onError
除此之外,連續 2 次呼叫 onError(),還報出了一個 UndeliverableException
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)
onError 和 onComplete 同時出現
emitter.onNext("1"); emitter.onNext("2"); emitter.onNext("3"); emitter.onComplete(); emitter.onError(new Exception()); emitter.onNext("4"); emitter.onNext("5");
輸出結果:輸出到onComplete截止
onSubscribeonNext 1onNext 2onNext 3onComplete
除此之外,同時也收到了 UndeliverableException 而終止程序
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)
但如果把 onError 放在 onComplete 前面呢?則輸出結果為:
onSubscribeonNext 1onNext 2onNext 3onErrorsubscribe over
說明onError 之前不能有 onComplete / onError 事件發生。
指定排程執行緒
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 "); } });
輸出結果:被觀察者在新的執行緒中執行傳送事件,觀察者在Android主執行緒中接收事件
Thread: mainonSubscribeThread: RxNewThreadScheduler-1subscribe overThread: mainonNext 1Thread: mainonNext 2Thread: mainonNext 3Thread: mainonNext 4Thread: mainonNext 5
壓力測試
被觀察者限制快速傳送資料,觀察察者每休息 5 秒鐘處理一個 onNext 事件
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 "); } });
輸出結果:記憶體溢位!!!在RxJava2中Observable是不支援背壓的
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)
總結
- 正常情況下事件會按照預期逐個由被觀察者傳送給觀察者
- 理論上來說,onComplete 和 onError 是互斥的,我們應該控制其只發送一個
- onComplete / onError 一旦傳送,被觀察者還會繼續傳送事件,但觀察者將不會再接收到任何事件
- onComplete 可以傳送多次,以收到第一次為結束,且不會有異常
- onError 只能被髮送一次,並且不能在有 onComplete 之後再發送 onError ,否則會報 UndeliverableException
- 被觀察者傳送事件的速度遠遠超過觀察者處理的速度,最終會造成 OOM。原因是 Observable 不支援背壓。