single - scheduler rx android

Creando Observable sin usar Observable.create

Question

Estoy usando RxJava en mi aplicación de Android y quiero cargar datos de la base de datos.

De esta manera, estoy creando un nuevo Observable usando Observable.create() que devuelve una lista de EventLog

public Observable<List<EventLog>> loadEventLogs() {
    return Observable.create(new Observable.OnSubscribe<List<EventLog>>() {
        @Override
        public void call(Subscriber<? super List<EventLog>> subscriber) {
            List<DBEventLog> logs = new Select().from(DBEventLog.class).execute();
            List<EventLog> eventLogs = new ArrayList<>(logs.size());
            for (int i = 0; i < logs.size(); i++) {
                eventLogs.add(new EventLog(logs.get(i)));
            }
            subscriber.onNext(eventLogs);
        }
    });
}

Aunque funciona correctamente, leí que el uso de Observable.create() no es realmente una mejor práctica para Rx Java (ver here ).

Así que cambié este método de esta manera.

public Observable<List<EventLog>> loadEventLogs() {
    return Observable.fromCallable(new Func0<List<EventLog>>() {
        @Override
        public List<EventLog> call() {
            List<DBEventLog> logs = new Select().from(DBEventLog.class).execute();
            List<EventLog> eventLogs = new ArrayList<>(logs.size());
            for (int i = 0; i < logs.size(); i++) {
                eventLogs.add(new EventLog(logs.get(i)));
            }
            return eventLogs;
        }
    });
}

¿Es este un mejor enfoque usando Rx Java? ¿Por qué? ¿Cuál es realmente la diferencia entre los dos métodos?

Además, dado que la base de datos carga una lista de elementos, ¿tiene sentido emitir toda la lista a la vez? ¿O debería emitir un artículo a la vez?

Answer 1

Como se explica en la respuesta que vinculó, con Observable.create es posible que tenga que infringir los requisitos avanzados de RxJava.

Por ejemplo, tendrá que implementar una backpressure o cómo darse de baja.

En su caso, desea emitir un artículo, sin tener que lidiar con contrapresión o suscripción. Así que Observable.fromCallable es una buena opción. RxJava se ocupará del resto.

Answer 2

Los dos métodos pueden parecer similares y comportarse de manera similar, pero fromCallable ocupa de las dificultades de la contrapresión para usted, mientras que la versión de create no lo hace. El OnSubscribe contrapresión dentro de una implementación de OnSubscribe abarca desde un simple derretimiento de la mente; sin embargo, si se omite, puede obtener MissingBackpressureException a lo largo de límites asincrónicos (como observeOn ) o incluso en límites de continuación (como concat ).

RxJava intenta ofrecer soporte de contrapresión adecuado para la mayor cantidad posible de fábricas y operadores, sin embargo, hay bastantes fábricas y operadores que no pueden soportarlo.

El segundo problema con la implementación manual de OnSubscribe es la falta de soporte de cancelación, especialmente si genera muchas llamadas en onNext . Muchos de estos pueden ser reemplazados por métodos de fábrica estándar (como from ) o clases de ayuda (como SyncOnSubscribe ) que tratan con toda la complejidad para usted.

Puede encontrar muchas presentaciones y ejemplos que (aún) usan create por dos razones.

1. Es mucho más fácil introducir flujos de datos basados ​​en push mostrando cómo el impulso de los eventos funciona de manera imperativa. En mi opinión, estas fuentes pasan demasiado tiempo create proporcional en lugar de hablar sobre los métodos estándar de fábrica y mostrar cómo determinadas tareas comunes (como la suya) se pueden lograr de forma segura.
2. Muchos de estos ejemplos se crearon en el momento en que RxJava no requería soporte de contrapresión o incluso compatibilidad de cancelación síncrona adecuada o simplemente se transfirieron de los ejemplos de Rx.NET (que hasta la fecha no admite contrapresión y la cancelación sincrónica funciona de alguna manera, cortesía de C # I adivinar). La generación de valores al invocar a Next no tenía preocupaciones en ese momento. Sin embargo, tal uso conduce a la saturación del buffer y al uso excesivo de la memoria, por lo tanto, el equipo de Netflix ideó una forma de limitar el uso de la memoria al requerir que los observadores establezcan cuántos elementos están dispuestos a realizar. Esto se conoce como contrapresión.

Para la segunda pregunta, es decir, si uno debe crear una Lista o una secuencia de valores, depende de su fuente. Si su fuente admite algún tipo de iteración o transmisión de elementos de datos individuales (como JDBC), puede conectar y emitir uno por uno (vea SyncOnSubscribe ). Si no lo admite o si lo necesita en forma de lista de todos modos, guárdelo como está. Siempre puede convertir entre los dos formularios a través de toList y flatMapIterable si es necesario.