Modo efficiente per manipolare i thread RxJava

Question

Nel mio progetto ho bisogno di elaborare gli oggetti in thread diversi. Per manipolare il comportamento di flusso io creo nuovi osservabili a cambiare il loro observeOn() questo modo:

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 
       }); 

Ma credo che in RxJava c'è molto di più modo bello ed efficiente per elaborare una risposta a diversi thread. Ho provato a google, ma non ho trovato nulla.

Grazie,
Anton

Answer 1

In Rx, di solito è consigliato per evitare effetti collaterali a 'fare' blocchi (che sarà eseguito solo se il flusso viene sottoscritto su), e preferiscono iscriversi codice.

Nel tuo caso è possibile sfruttare cache() o publish()...connect(), ad esempio:

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 
      }); 

Con publish() invece di cache(), il codice è identico, ma si può decidere quando attivare la query da collegando il flusso (si chiamare query.connect() dopo aver collegato i 2 abbonamenti).

Se il lavoro di abbonamento è un calcolo in background, è preferibile il Schedulers.computation() su Schedulers.io().

Nota che AFAICT il codice funziona bene senza la linea map(Observable::just), come le dichiarazioni 'observeOn' impatto solo il flusso, più in basso (e non precedente 'fare' dichiarazioni)