RxJava2配置及使用詳解

2019-12-12 02:34:00

字體：大中小

來源：轉載

供稿：網友

RxJava2.0是一個非常棒的流式編程,采用的觀察者模式思想,事件的產生者產生事間之后發送給綁定的接受者,接受順序與發送順序一致.

依賴:

compile 'io.reactivex.rxjava2:rxjava:2.0.1'compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

簡單使用:

//觀察者模式,這里產生事件,事件產生后發送給接受者,但是一定要記得將事件的產生者和接收者捆綁在一起,否則會出現錯誤Observable.create(new ObservableOnSubscribe<String>() {  @Override  public void subscribe(ObservableEmitter<String> e) throws Exception {    //這里調用的方法會在產生事件之后會發送給接收者,接收者對應方法會收到    e.onNext("hahaha");    e.onError(new Exception("wulala"));    e.onComplete();  }/*--}).subscribe(new Observer<String>() {  //接受者,根據事件產生者產生的事件調用不同方法  @Override  public void onSubscribe(Disposable d) {    Log.e(TAG, "onSubscribe: ");  }  @Override  public void onNext(String value) {    Log.e(TAG, "onNext: " + value);  }  @Override  public void onError(Throwable e) {    Log.e(TAG, "onError: ", e);  }  @Override  public void onComplete() {    Log.e(TAG, "onComplete: ");  }});

我們來用圖解一下這其中發生了什么事:

上游朝下游發送數據,經過subscribe使上下游產生關系,即達成訂閱.

解析1:

ObservableEmitter,這是個啥東西?Emitter:顧名思義,即Rxjava的發射器,通過這個發射器,即可發送事件-----通過調用onNext,onError,onComplete方法發送不同事件.

注意:

雖然RxJava可以進行事件發送,但這并不意味著你可以隨便發送,這其中需要遵循一些規則.

onNext:你可以發送無數個onNext,發送的每個onNext接受者都會接收到.

onError:當發送了onError事件之后,發送者onError之后的事件依舊會繼續發送,但是接收者當接收到onError之后就會停止接收事件了.

onComplete:當發送了onComplete事件之后,發送者的onComplete之后的事件依舊會繼續發送,但是接收者接收到onComplete之后就停止接收事件了.

onError事件和onComplete事件是互斥的,但是這并不代表你配置了多個onError和onComplete一定會崩潰,多個onComplete是可以正常運行的,但是只會接收到第一個,之后的就不會再接收到了,多個onError時,只會接收到第一個,第二個會直接造成程序崩潰.

解析2:

Disposable又是個啥東西,翻譯之后百度告訴我這東西叫做一次性的,是用來控制發送者和接受者之間的紐帶的,默認為false,表示發送者和接受者直接的通信閥門關閉,可以正常通信,在調用dispose()方法之后,閥門開啟,會阻斷發送者和接收者之間的通信,從而斷開連接.

重載方法:

     subscribe();     //表示發送者隨意發送數據,接受者什么都不管,什么都不接收.     subscribe(Consumer<? super T> onNext) {}   //只響應onNext()事件,其他的事件忽略.     subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}     //含義同上     subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}     //含義同上     subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}   //含義同上

解析3:

默認情況下,發送者和接收者都運行在主線程,但是這顯然是不符合實際需求的,我們在日常使用中,通常用的最多的就是在子線程進行各種耗時操作,然后發送到主線程進行,難道我們就沒有辦法繼續用這個優秀的庫了?想多了你,一個優秀的庫如果連這都想不到,怎么能被稱為優秀呢,RxJava中有線程調度器,通過線程調度器,我們可以很簡單的實現這種效果,下面放代碼.

Observable.create(new ObservableOnSubscribe<String>() {  @Override  public void subscribe(ObservableEmitter<String> e) throws Exception {    e.onNext("hahaha");    e.onNext("hahaha");    e.onNext("hahaha");    Log.e(TAG,"運行在什么線程" + Thread.currentThread().getName());    e.onComplete();  }}).subscribeOn(Schedulers.newThread())        //線程調度器,將發送者運行在子線程 .observeOn(AndroidSchedulers.mainThread())     //接受者運行在主線程 .subscribe(new Observer<String>() {  @Override  public void onSubscribe(Disposable d) {    Log.e(TAG, "onSubscribe: ");    Log.e(TAG,"接收在什么線程" + Thread.currentThread().getName());  }  @Override  public void onNext(String value) {    Log.e(TAG, "onNext: " + value);  }  @Override  public void onError(Throwable e) {    Log.e(TAG, "onError: ", e);  }  @Override  public void onComplete() {    Log.e(TAG, "onComplete: ");  }});

最終結果: