rx-java - single - rxjava observable

Proceso observable en el hilo de fondo

Question

Estoy usando RxAndroid para las operaciones de transmisión. En mi caso de uso real, estoy obteniendo una lista del servidor (usando Retrofit ). Estoy usando programadores para hacer el trabajo en un hilo de fondo y obtener la emisión final en el hilo de la interfaz de usuario (principal) de Android.

Esto funciona bien para la llamada de red, sin embargo, me di cuenta de que mis operadores después de la llamada de red no usan el hilo de fondo, sino que me llaman en el hilo principal.

myService.fetchSomeIntegersFromServer()
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.filter(integer -> {
System.out.println(Looper.getMainLooper().getThread() == Thread.currentThread());
return true;
})
.subscribe(integer1 -> {});

¿Cómo puedo asegurarme de que todas las operaciones se ejecutan en un hilo de fondo?

Answer 1

TL; DR : mueve observeOn(AndroidSchedulers.mainThread()) debajo del filter(...) .

subscribeOn(...) se utiliza para designar en qué hilo comenzará a funcionar el Observable . Las llamadas subsiguientes a subscribeOn serán ignoradas.

Por lo tanto, si tuviera que escribir lo siguiente, todo se ejecutaría en Schedulers.newThread() :

myService.fetchSomeIntegersFromServer()
.subscribeOn(Schedulers.newThread())
.filter(integer -> {
System.out.println(Looper.getMainLooper().getThread() == Thread.currentThread());
return true;
})
.subscribe(integer1 -> { doSomething(integer1); });

Ahora, por supuesto, esto no es lo que quieres: quieres hacer algo en el hilo principal.
Ahí es donde entra en observeOn . Todas las acciones después de observeOn se ejecutan en ese programador. Por lo tanto, en su ejemplo, el filter se ejecuta en el hilo principal.

En su lugar, mueva observeOn hacia abajo justo antes de subscribe :

myService.fetchSomeIntegersFromServer()
.subscribeOn(Schedulers.newThread())
.filter(integer -> {
System.out.println(Looper.getMainLooper().getThread() == Thread.currentThread());
return true;
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(integer1 -> { doSomething(integer1) });

Ahora, el filter ocurrirá en el ''nuevo hilo'', y doSomething en el hilo principal.

Para ir aún más lejos, puede usar observeOn varias veces:

myService.fetchSomeIntegersFromServer()
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.computation())
.filter(integer -> {
System.out.println(Looper.getMainLooper().getThread() == Thread.currentThread());
return true;
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(integer1 -> { doSomething(integer1) });

En este caso, la recuperación se realizará en un nuevo subproceso, el filtrado en un subproceso de cálculo y doSomething en el subproceso principal.

Checkout ReactiveX - Suscriba al operador para la documentación oficial.