Estamos en el proceso de creación de prototipos de una API REST mediante programación reactiva. Como se muestra en el diagrama, mantenemos 3 capas iguales a las que usamos en nuestros diseños de API de sincronización anteriores;

http://oi59.tinypic.com/339hhki.jpg

La capa API se implementó utilizando Jersey2, que procesará la solicitud / deserialización de JSON y la transferencia a la capa de servicio.Capa de servicio que implementa la lógica de negocios. Implementado usando programación reactiva (RxJava)Dao Layer, que se utiliza para las operaciones de persistencia por parte de Service Layer. Dado que usamos CouchBase, esto utilizará CouchBase RxClient.

A mi entender, el flujo es el siguiente:

una) Llega la solicitud HTTP, Jersery procesará la solicitud / analizará el modelo de solicitud JSON / deserialize dentro de un RequestThread del "grupo de subprocesos de contenedor".

si) Con el soporte Jersey2 Async, RequestThread regresará a Container Thread Pool, y la capa de servicio se ejecutará en Schedulers.computation () Scheduler.

@Path("/resource")
public class AsyncUserResource {
    @GET
    public void asyncGet(@Suspended final AsyncResponse asyncResponse) {
 
       Observable<User> user = userService.getUser(...); //this is executed using Schedulers.computation() inside Service implementation


       user.subscribe(new Observer<User>() {

            @Override
            public void onCompleted() { 

            }

            @Override
            public void onError(Throwable e) {
                //handle error using ExceptionMappers

            }

            @Override
            public void onNext(User user) {
               asyncResponse.resume(user); 

            }});
    }        


}

C) Cualquier operación de E / S dentro de DAO utilizará Schedulers.io () para ejecutar estas operaciones de procesamiento largo en un hilo separado.

Mis preguntas son:

Al implementar DAO / Servicios, ¿debo ocultar los planificadores en uso (Threading) dentro de la implementación.

por ejemplo, Dao:

public interface UserDao {
  public Observable<User> getUser();
}

En la implementación, ¿es una buena práctica especificar el Schedular como se muestra a continuación?

public Observable<User> getUser() {

        Observable<User> ret = Observable.create((subscriber)->{
            try {

                 //Do DB call
                 User u = null;
                 subscriber.onNext(u);
                 subscriber.onCompleted();

            }catch (Exception e) {
                subscriber.onError(e);  
            }

        });
        return ret.subscribeOn(Schedulers.io());
}

¿O es mejor simplemente devolver el Observable, y la capa superior usará un Schedular particular en consecuencia?

Dado que los DAO en su mayoría invocan llamadas io / network, supongo que se debe utilizar Schedulars.io (). ¿Qué tal la lógica de negocios junto a la capa de servicio? ¿Deberían ejecutarse dentro de Schedulers.computation () (Event Loop)?

Hay dos grupos de subprocesos dentro de la JVM. Uno es "Grupo de subprocesos de contenedor" y el otro es "Grupo de subprocesos Rx" utilizado por Schedulers.io (). ¿Cómo configurar los ajustes del grupo / tamaño de RxJava?