android - sirve - RxJava y datos en caché

tipos de memoria cache

Question

Todavía soy bastante nuevo en RxJava y lo estoy usando en una aplicación de Android. He leído una tonelada métrica sobre el tema, pero todavía siento que me falta algo.

Tengo el siguiente escenario:

Tengo datos almacenados en el sistema al que se accede a través de varias conexiones de servicios (AIDL) y necesito recuperar datos de este sistema (puede ocurrir una cantidad n 1 de llamadas asincrónicas). Rx me ayudó mucho a simplificar este código. Sin embargo, todo este proceso suele tardar unos segundos (más de 5 segundos +), por lo tanto, necesito almacenar estos datos en caché para acelerar la aplicación nativa.

Los requisitos en este punto son:

1. Suscripción inicial, la caché estará vacía, por lo tanto, tenemos que esperar el tiempo requerido para cargar. No es gran cosa. Después de eso, los datos deben ser almacenados en caché.

2. Las cargas posteriores deben extraer los datos de la memoria caché, pero luego los datos deben volver a cargarse y la memoria caché de disco debe estar detrás de las escenas.

El problema: tengo dos Observables - A y B. A contiene los Observables anidados que extraen datos de los servicios locales (toneladas que están sucediendo aquí). B es mucho más simple. B simplemente contiene el código para extraer los datos de la memoria caché de disco.

Necesidad de resolver: a) Devolver un elemento en caché (si está almacenado en caché) y continuar cargando nuevamente el caché de disco. b) La memoria caché está vacía, carga los datos del sistema, los almacena en caché y los devuelve. Las llamadas posteriores vuelven a "a".

He tenido algunas personas que recomiendan algunas operaciones como flatmap, merge e incluso temas, pero por alguna razón tengo problemas para conectar los puntos.

¿Cómo puedo hacer esto?

Answer 1

Aquí hay un par de opciones sobre cómo hacer esto. Trataré de explicarlos lo mejor que pueda a medida que avance. Esto es servilleta-código, y estoy usando la sintaxis lambda estilo Java8 porque soy flojo y es más bonito. :)

1. Un sujeto, como AsyncSubject , sería perfecto si pudiera mantener estos como estados de instancia en la memoria, aunque parece que necesita almacenarlos en el disco. Sin embargo, creo que este enfoque vale la pena mencionar en caso de que sea capaz de hacerlo. Además, es solo una técnica ingeniosa para saber. AsyncSubject es un Observable que solo emite el ÚLTIMO valor publicado (Un Sujeto es a la vez Observador y Observable), y solo comenzará a emitir después de que se haya llamado a onCompleted . Por lo tanto, todo lo que se suscriba después de completar recibirá el siguiente valor.

   En este caso, podría tener (en una clase de aplicación u otra instancia de singleton en el nivel de la aplicación):

   public class MyApplication extends Application {    
       private final AsyncSubject<Foo> foo = AsyncSubject.create();
   
       /** Asynchronously gets foo and stores it in the subject. */
       public void fetchFooAsync() {
           // Gets the observable that does all the heavy lifting.
           // It should emit one item and then complete.
           FooHelper.getTheFooObservable().subscribe(foo);
       }
   
       /** Provides the foo for any consumers who need a foo. */
       public Observable<Foo> getFoo() {
           return foo;
       }
   
   }

2. Aplazando lo Observable. Observable.defer le permite esperar para crear un Observable hasta que esté suscrito. Puede usar esto para permitir que la recuperación de la memoria caché de disco se ejecute en segundo plano, y luego devolver la versión almacenada en caché o, si no está en la caché, hacer el trato real.

   Esta versión asume que su código getter, tanto la captura de caché como la creación de no captura, están bloqueando llamadas, no observables, y el aplazamiento funciona en segundo plano. Por ejemplo:

   public Observable<Foo> getFoo() {
       Observable.defer(() -> {
           if (FooHelper.isFooCached()) {
               return Observable.just(FooHelper.getFooFromCacheBlocking());
           }
           return Observable.just(FooHelper.createNewFooBlocking());
       }).subscribeOn(Schedulers.io());
   }

3. Use concatWith y take . Aquí suponemos que nuestro método para obtener el Foo de la memoria caché de disco emite un solo elemento y se completa o simplemente se completa sin emitir, si está vacío.

   public Observable<Foo> getFoo() {
       return FooHelper.getCachedFooObservable()
               .concatWith(FooHelper.getRealFooObservable())
               .take(1);
   }

   Ese método solo debería intentar obtener el trato real si el elemento visible en caché termina vacío.

4. Use amb o ambWith . Esta es probablemente una de las soluciones más locas, pero divertida de señalar. amb toma básicamente un par (o más con las sobrecargas) observables y espera hasta que uno de ellos emita un objeto, luego descarta por completo el otro observable y simplemente toma el que ganó la carrera. La única forma en que esto sería útil es si es posible que el paso de cálculo de crear un nuevo Foo sea más rápido que recuperarlo del disco. En ese caso, podrías hacer algo como esto:

   public Observable<Foo> getFoo() {
       return Observable.amb(
               FooHelper.getCachedFooObservable(),
               FooHelper.getRealFooObservable());
   }

Prefiero la Opción 3. En cuanto a almacenarla en caché, podrías tener algo como esto en uno de los puntos de entrada (preferiblemente antes de que necesitemos el Foo, ya que como dijiste, esta es una operación de larga duración) Más tarde los consumidores debería obtener la versión en caché mientras haya terminado de escribir. Usar un AsyncSubject aquí también puede ser útil, para asegurarse de que no activemos el trabajo varias veces mientras esperamos que se escriba. Los consumidores solo obtendrían el resultado completo, pero nuevamente, eso solo funciona si se puede mantener razonablemente en la memoria.

if (!FooHelper.isFooCached()) {
    getFoo()
        .subscribeOn(Schedulers.io())
        .subscribe((foo) -> FooHelper.cacheTheFoo(foo));
}

Tenga en cuenta que, debe mantener alrededor de un programador de hilos único destinado a la escritura (y lectura) de disco y usar .observeOn(foo) después de .subscribeOn(...) o sincronizar el acceso a la memoria caché de disco para evitar problemas de concurrencia.

Answer 2

Echa un vistazo al proyecto a continuación. Este es mi punto de vista personal sobre las cosas y he usado este patrón en varias aplicaciones.

https://github.com/zsiegel/rxandroid-architecture-sample

Eche un vistazo al PersistenceService. En lugar de presionar la base de datos (o MockService en el proyecto de ejemplo), simplemente podría tener una lista local de usuarios que se actualizan con el método save () y simplemente devolverla en el get ().

Hazme saber si tienes alguna pregunta.

Answer 3

Recientemente publiqué una biblioteca en Github para Android y Java, llamada RxCache , que satisface sus necesidades sobre el almacenamiento en caché de datos utilizando observables.

RxCache implementa dos capas de almacenamiento en memoria caché: memoria y disco, y cuenta con varias anotaciones para configurar el comportamiento de cada proveedor.

Se recomienda encarecidamente utilizar con Retrofit datos recuperados de las llamadas http. Usando la expresión lambda, puede formular la expresión de la siguiente manera:

    rxCache.getUser(retrofit.getUser(id), () -> true).flatmap(user -> user);

Espero que les resulte interesante :)