Efektywny sposób manipulowania wątkami RxJava

Question

W moim projekcie muszę przetwarzać obiekty w różnych wątkach. Manipulować zachowanie Stream tworzę nowe obserwable zmienić swoje observeOn() ten sposób:

apiService.getObjects(token) // Retrofit 
       .compose(bindToLifecycle()) 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .doOnNext(o -> { 
        // process in Main Thread 
       }) 
       .map(Observable::just) // create new one, to change thread for it 
       .observeOn(Schedulers.io()) 
       .subscribe(o -> { 
        // process in the background thread 
       }); 

Ale myślę, że w RxJava jest dużo piękniejsza i bardziej efektywny sposób przetwarzać jedną odpowiedź w różnych wątkach. Próbowałem google, ale nic nie znalazłem.

Dzięki,
Anton

Answer 1

W Rx, to zazwyczaj zaleca się, aby uniknąć skutków ubocznych w „do” bloków (które zostaną wykonane tylko wtedy, gdy strumień zostanie subskrybowanych dalej), i wolą subskrybować kod.

W twoim przypadku można wykorzystać cache() lub publish()...connect(), np:

query = apiService.getObjects(token) 
      .compose(bindToLifecycle()) 
      .subscribeOn(Schedulers.io()) 
      .cache(); 

query.observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(o -> { 
       // process in Main Thread 
      }) 
query.observeOn(Schedulers.io()) 
      .subscribe(o -> { 
       // process in the background thread 
      }); 

Z publish() zamiast cache() kod jest identyczny, ale można zdecydować, kiedy ogień zapytanie przez łączącej strumień (ci zadzwoń pod numer query.connect() po okablowaniu 2 subskrypcji).

Jeśli praca subskrypcyjna jest obliczana w tle, może być preferowana wersja Schedulers.computation() w stosunku do Schedulers.io().

Zauważ, że AFAICT Twój kod będzie działać dobrze bez linii map(Observable::just), jako sprawozdanie „observeOn” tylko wpływ na strumień dalej w dół (a nie poprzednia „zrobić” wypowiedzi)