Estamos usando Micronaut con Mongo para exponer datos a través de algunos controladores. Debido a que el tamaño de las entidades de respuesta está creciendo, nuestras aplicaciones a veces se quedan sin memoria. Por lo tanto, estamos investigando el cambio al controlador mongo asíncrono y el uso de respuestas reactivas para transmitir los datos a los clientes. Lamentablemente, no podemos cambiar las estructuras de respuesta de la API ni los tipos de contenido (todas las application/json )
Una de nuestras API devolvió entidades estructuradas así:
[ { "field": "value" }, { "field": "value" }, ... { "field": "value" } ] Esto lo conseguimos usando este controlador, donde dataStore devuelve un Publisher<Example> :
@Get("all") Flowable<Example> getAllExamples() { return Flowable.fromPublisher(dataStore.find()).map(SomeMapper::toPublic); }Esto funciona bien, la enorme lista de ejemplos no tiene que estar completamente cargada en la memoria antes de transmitirla al cliente.
Otras API devuelven la estructura (en mi opinión, más sensata):
{ "list": [ { "field": "value" }, { "field": "value" }, ... { "field": "value" } ], "meta": { ... } }¿Podemos aplicar un patrón de flujo/publicador similar para entidades como esta, o estamos atascados cargando datos para tales respuestas en la memoria antes de enviarlas?
Probamos firmas como:
@Get("all/dev") Single<ExamplesWrapper> getAllDev() { Publisher<Example> dev = dataStore.find(); return Flowable.fromPublisher(dev) .map(mapper::map) .collect((Callable<ArrayList<Example>>) ArrayList::new, ArrayList::add) .map(ExampleWrapper::new); }Donde el contenedor agregaría algunos metadatos. Pero esto nuevamente lo carga todo en la memoria antes de enviarlo, bloqueando la aplicación.
Agregar el Flowable en el contenedor de respuesta:
public class ExamplesWrapper { private final Flowable<Example> examples; @ConstructorProperties({"examples"}) public ExamplesWrapper(Flowable<Example> examples) { this.examples = examples; } public Flowable<Example> getExamples() { return examples; } }También falla con alguna buena excepción de mapeo de Jackson.
Los metadatos no dependen de los datos de ejemplo reales (agregan información estática de la empresa). ¿Podemos implementar de alguna manera tal punto final sin tener que cargar todos los datos en la memoria?
De la documentación :
6.20 Escritura de datos de respuesta
Escritura reactiva de datos de respuesta
El servidor HTTP de Micronaut admite la escritura de fragmentos de datos de respuesta al devolver un Publisher que emite objetos que se pueden codificar en la respuesta HTTP.
La siguiente tabla resume las firmas de tipo de devolución de ejemplo y el comportamiento que exhibe el servidor para manejarlas: Tipo de devolución Descripción
- Flowable<byte[]>: un Flowable que emite cada fragmento de contenido como un byte[] sin bloquear
- Flux<ByteBuf>: un flujo de reactor que emite cada fragmento como un Netty ByteBuf
- Publisher<String>: un editor que emite cada fragmento de contenido como una cadena
- Flowable<Book> Al emitir un POJO, cada objeto emitido se codifica como JSON de forma predeterminada sin bloquear
Al devolver un tipo reactivo, el servidor utiliza una codificación de transferencia fragmentada y sigue escribiendo datos hasta que se llama al método Publisher onComplete.
Entiendo esto, por lo que si desea que el mecanismo de Micronaut transmita sus cosas, debe tener una firma como Flowable<item> o Flux<item> o Publisher<item> , donde item es una parte de su respuesta, no un elemento completo . Micronaut luego responderá con fragmentos a medida que provengan de Flowable o equivalente.
En este caso, una cosa que pensé es que usted mismo puede dividir en partes adecuadas. De esa manera, la transmisión de respuestas grandes sin almacenarlas en la memoria debería funcionar.
Así que algo como esto:
@Get("all") public Flowable<String> getAllExamples() { ObjectMapper objectMapper = new ObjectMapper(); Publisher<Example> dev = dataStore.find(); return Flowable.fromPublisher(dev) .map(mapper::map) .concatMap(item -> Flowable.just(objectMapper.writeValueAsString(item), ",")) .startWith("{\"list\": [") .concatWith(Flowable.just("],\"meta\":\"whatever\"}")); }es hacky, pero parece funcionar para tal caso.
Algunos enfoques que no funcionaron:
Probé escribiendo directamente en JsonGenerator en un asignador de Jackson personalizado, descargando objetos a medida que avanzan como se describe en Jackson Streaming Api, pero el micronauta RoutingInboundHandler no parece enviar la respuesta al usuario final, sino que la almacena en el búfer, lo que resulta en una falta de memoria. Approach funciona con Spring Boot, por lo que posiblemente sea una característica que falta en Micronaut.
También me sucedió el mismo almacenamiento en búfer cuando usaba respuestas de Micronaut Writeable (bloqueo) y trataba de vaciar los datos tal como están escritos. Abrí un tema sobre eso a micronaut core .