Attività asincrona RxJava in Android

Question

Sto cercando di implementare un'attività asincrona utilizzando RxJava in Android. Ho provato il seguente codice e non ha funzionato. Esegue sul thread dell'interfaccia utente. Sto usando la seguente versione di RxAndroid 0.24.0.

try { 
      Observable.just(someMethodWhichThrowsException()) 
        .subscribeOn(Schedulers.newThread()) 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(s -> onMergeComplete()); 
     } catch (IOException e) { 
      e.printStackTrace(); 
     } 

Tuttavia, il seguente funziona in modo asincrono per me.

Observable observable = Observable.create(new Observable.OnSubscribe<String>() { 
      @Override 
      public void call(Subscriber<? super String> subscriber) { 
       try { 
        someMethodWhichThrowsException(); 
       } catch (IOException e) { 
        e.printStackTrace(); 
       } 

       subscriber.onCompleted(); 
      } 
     }); 
     observable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(); 

io sto cercando di capire quanto segue: 1. Qual è la differenza tra loro? 2. Qual è la migliore pratica durante la creazione di compiti asincroni?

Grazie in anticipo.

Answer 1

1. Qual è la differenza tra loro?

Observable.just(someMethodWhichThrowsException()) 
    .subscribeOn(Schedulers.newThread()) 

Questo è equivalente alla seguente:

Object someResult = someMethodWhichThrowsException(); 
Observable.just(someResult) 
    .subscribeOn(Schedulers.newThread()) 

Come si può vedere questo rende il metodo chiamata sincrona prima, poi passa a Observable.just per diventare osservabile.

Observable.create(new Observable.OnSubscribe<String>() { 
     @Override 
     public void call(Subscriber<? super String> subscriber) { 
      ... 
     } 
    }) 
    .subscribeOn(Schedulers.newThread()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(); 

Questo metodo, tuttavia, verrà eseguito il codice nel blocco callin abbonamento. Hai detto che vuoi iscriverti su una nuova discussione (subscribeOn(Schedulers.newThread())), quindi l'iscrizione avviene su una nuova discussione e il codice che viene eseguito all'abbonamento (il blocco call) viene eseguito anche su quel thread. Questo è un comportamento simile a chiamare Observable.defer.

2. Qual è la procedura migliore durante la creazione di attività asincrone?

Bene, dipende da voi e dal vostro comportamento desiderato. A volte si desidera che il codice asincrono inizi immediatamente a essere eseguito (in tal caso si consiglia di memorizzarlo nella cache utilizzando uno degli operatori a tale scopo). Sicuramente prenderemo in considerazione l'utilizzo della libreria Async Utils per questo.

Altre volte, ti consigliamo di eseguire solo su abbonamento (che è il comportamento negli esempi qui) - per esempio se ci sono effetti collaterali, o se non ti interessa quando è gestito e vuole solo utilizzare i built-in per ottenere qualcosa dal thread dell'interfaccia utente. Dan Lew mentions che Observable.defer è molto utile per prendere il vecchio codice e rimuoverlo dal thread dell'interfaccia utente, durante una conversione in Rx.

Answer 2

Utilizzare Async.start() dalla libreria RxJava Async Utils. Questo chiamerà la funzione che fornisci su un altro thread.

Esempio:

Observable<String> observable = Async.start(new Func0<String>() { 
    @Override 
    public String call() { 
     try { 
      return someMethodWhichThrowsException(); 
     } catch (IOException e) { 
      throw new RuntimeException(e); 
     } 
    } 
}); 

Come si nota, eccezioni controllate devono essere avvolti in RuntimeExceptions.

Vedi anche https://github.com/ReactiveX/RxJava/wiki/Async-Operators#start