Cuando se trabaja con bases de datos distribuidas como Couchbase, el rendimiento y la eficiencia son consideraciones clave, especialmente cuando se recupera una gran cantidad de datos. Muchas veces, cuando los clientes vienen de diferentes entornos de desarrollo o bases de datos, preguntan sobre la capacidad de Couchbase para hacer operaciones “multi-get” o “bulk get”. Muchas bases de datos ofrecen “multi-get” como método para recuperar múltiples documentos basados en sus claves. La mayoría de los SDKs de Couchbase no ofrecen APIs explícitas para batching porque la programación reactiva provee la flexibilidad de implementar batching a la medida de tu caso de uso específico y es a menudo más efectivo que un método genérico de talla única.
¿Qué es Bulk Get?
Una operación de obtención masiva permite solicitar varios documentos en una sola operación, en lugar de realizar repetidas llamadas GET individuales. En los almacenes tradicionales de clave-valor, cada petición suele dirigirse a un nodo específico. Sin embargo, en un entorno distribuido como Couchbase, repartir estas operaciones entre nodos puede introducir sobrecarga si se gestiona manualmente.
Soporte SDK para operaciones masivas
Los SDKs de Couchbase (incluyendo Java, .NET y Go) ofrecen soporte integrado para operaciones de obtención masiva. Estos métodos SDK están diseñados para aceptar una lista de claves de documentos y gestionar automáticamente la ejecución en paralelo de las operaciones individuales. GET de forma eficaz por tres razones principales.
-
- Paralelismo: En lugar de buscar cada documento secuencialmente, los SDK inician varias solicitudes simultáneamente.
- Selección de nodos: Los SDK dirigen de forma inteligente cada solicitud al nodo correcto del clúster donde residen los datos.
- Ejecución asíncrona: Aprovechando las capacidades asíncronas de cada SDK, las operaciones se gestionan de forma no bloqueante, lo que garantiza un mayor rendimiento y una mejor utilización de los recursos.
Couchbase proporciona dos formas principales de lograr la capacidad de obtención masiva usando Programación Reactiva y Programación Asíncrona.
API reactiva
Si tu objetivo es optimizar las operaciones de obtención masiva en Couchbase, la programación reactiva proporciona un enfoque eficiente y más fácil. El protocolo binario de Couchbase tiene ejecución fuera de orden y tiene un fuerte soporte para operaciones asíncronas en KV. Al gestionar eficientemente los flujos de datos asíncronos, permite un alto rendimiento y una baja latencia, lo que lo hace ideal para sistemas distribuidos. Para aprovechar al máximo sus capacidades, lo ideal es una pila totalmente reactiva en la que cada capa, desde la base de datos hasta el cliente, soporte flujos reactivos. Couchbase ColecciónReactiva se integra a la perfección con Project Reactor, permitiendo el acceso sin bloqueos a las operaciones Key-Value (KV) de Couchbase. Esta integración se alinea perfectamente con las arquitecturas reactivas modernas, permitiendo a las aplicaciones manejar cargas de trabajo de alto rendimiento de manera más eficiente al evitar el bloqueo innecesario de hilos.
Dicho esto, migrar toda una aplicación existente a una arquitectura reactiva puede implicar un trabajo importante. Si se trata de un proyecto nuevo, adoptar un framework reactivo como Spring WebFlux es muy recomendable. Sin embargo, incluso en aplicaciones no reactivas, la introducción de un enfoque reactivo en la capa CRUD de Couchbase por sí sola puede proporcionar ganancias significativas. Al hacerlo, puedes minimizar el bloqueo de hilos y reducir el estrangulamiento de la CPU, lo que lleva a una mejor eficiencia de los recursos y una mejor escalabilidad.
A continuación se muestra un ejemplo de un código Java que puede maximizar el rendimiento de Couchbase utilizando Reactive API y puede trabajar con una pila no reactiva.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
/** * @param collection The collection to get documents from. * @param documentIds The IDs of the documents to return. * @param mapSupplier Factory for the returned map. Suggestion: * Pass {@code TreeMap::new} for sorted results, * or {@code HashMap::new} for unsorted. * @param concurrency Limits the number of Couchbases requests in flight * at the same time. Each invocation of this method has a separate quota. * Suggestion: Start with 256 and tune as desired. * @param mapValueTransformerScheduler The scheduler to use for converting * the result map values. Pass {@link Schedulers#immediate()} * to use the SDK's IO scheduler. Suggestion: If your value converter does IO, * pass {@link Schedulers#boundedElastic()}. * @param mapValueTransformer A function that takes a document ID and a result, * and returns the value to associated with that ID in the returned map. * @param <V> The return map's value type. * @param <M> The type of the map you'd like to store the results in. * @return a Map (implementation determined by {@code mapSupplier}) * where each given document ID is associated with the result of * getting the corresponding document from Couchbase. */ public static <V, M extends Map<String, V>> Map<String, V> bulkGet( ReactiveCollection collection, Iterable<String> documentIds, int concurrency, Supplier<M> mapSupplier, Scheduler mapValueTransformerScheduler, BiFunction<String, SuccessOrFailure<GetResult>, V> mapValueTransformer ) { return Flux.fromIterable(documentIds) .flatMap( documentId -> Mono.zip( Mono.just(documentId), collection.get(documentId) .map(SuccessOrFailure::success) .onErrorResume(error -> Mono.just(SuccessOrFailure.failure(error))) ), concurrency ) .publishOn(mapValueTransformerScheduler) .collect( mapSupplier, (map, idAndResult) -> { String documentId = idAndResult.getT1(); SuccessOrFailure<GetResult> successOrFailure = idAndResult.getT2(); map.put(documentId, mapValueTransformer.apply(documentId, successOrFailure)); } ) .block(); } } |
Este enfoque reactivo busca los documentos utilizando sus ID y devuelve un archivo Mapa donde cada clave es un ID de documento y el valor es el resultado procesado. Aunque no está mal recopilar los resultados en una Lista y volver a procesarlos más tarde, una estrategia mejor (tanto en términos de rendimiento como de claridad del código) es recopilar los resultados en un archivo ConcurrentHashMap indexados por ID de documento. De este modo se evitan las exploraciones repetidas y las búsquedas de resultados se realizan en tiempo constante. Veamos cómo funciona paso a paso.
-
- Creación de un flujo reactivo a partir de identificadores de documentos
En la línea 19, estamos creando un Flux (flujo reactivo) a partir de la lista de ID de documentos. Para cada ID de documento, llama a collection.get(documentId) para obtener el documento de forma reactiva. - La envoltura da lugar a ÉxitoOfracaso
Para garantizar la resistencia, cada operación asíncrona envuelve el resultado en un archivo ÉxitoOrFallo objeto. Esta envoltura captura tanto las búsquedas exitosas como las fallidas. Por defecto, si collection.get(documentId) arroja un error (por ejemplo, un problema de red, un documento que falta), todo el Flux fallará y dejará de procesarse. Esto no es ideal para operaciones masivas, ya que queremos seguir procesando otros documentos incluso si uno falla. Así que en lugar de propagar el error, convierte el fallo en un SuccessOrFailure.failure(error) objeto. De esta forma, el flujo descendente sigue obteniendo un valor válido (ÉxitoOfracaso) para cada ID de documento, tanto si ha tenido éxito como si ha fallado. - Emparejamiento de identificadores de documentos con resultados mediante Mono.zip
Utilizando Mono.zip hace explícito que está combinando el documentId y el async consiga en una tupla. Esto ayuda a identificar la asociación entre el documentID y el resultado, especialmente cuando los resultados llegan fuera de orden debido a la concurrencia. - Concurrencia controla cuántas búsquedas de documentos se ejecutan en paralelo (cuántas solicitudes están en vuelo a la vez).
- Paralelismo y transferencia de planificadores
Los flujos reactivos no se bloquean por defecto, pero la lógica de transformación (por ejemplo, el análisis sintáctico de JSON o la conversión de datos) puede consumir mucha CPU. Antes de recopilar las tuplas resultantes, el flujo cambia a un planificador especificado por la persona que llama utilizando publishOn(...). Esto descarga el trabajo de transformación de los hilos IO a un grupo de hilos separado. Esto garantiza que los subprocesos de E/S no se bloqueen por el trabajo de transformación debido a una computación pesada.
- Recopilación en un mapa
Una vez obtenidos todos los resultados, el flujo recopila los pares de tuplas en un mapa. Utiliza mapProveedor para crear el mapa. Para cada (documentId, resultado) par, se aplica mapValueTransformer para transformar el resultado bruto en un tipo específico del dominio V y, a continuación, introduce el valor transformado en el mapa.
- Bloqueo para recuperar el resultado final
Como aquí todo es asíncrono (no bloqueante), bloquear() se utiliza para esperar a que termine todo el flujo y devolver el flujo construido mapa a la persona que llama.
- Creación de un flujo reactivo a partir de identificadores de documentos
API asíncrona
Aunque recomendamos el uso de las APIs reactivas por su rendimiento, flexibilidad y manejo de la contrapresión, Couchbase también ofrece una API asíncrona de bajo nivel para escenarios en los que necesites un control aún más preciso y un ajuste del rendimiento. Sin embargo, escribir código asíncrono eficiente conlleva sus propios retos, requiere una gestión cuidadosa de la concurrencia y la contrapresión para prevenir el agotamiento de recursos y evitar timeouts.
A continuación se muestra un ejemplo que demuestra cómo utilizar la API Async para mejorar el rendimiento de la obtención masiva en Couchbase:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
// async api to call get() for an array of keys List<CompletableFuture<GetResult>> futures = new LinkedList<>(); for (int i = 0; i < keys.size(); i++) { CompletableFuture<GetResult> f = collection.async().get( keys.get(i), (GetOptions) options ); futures.add(f); } // Wait for all Get operations to complete CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); // Convert results to JsonObjects List<JsonObject> results = new LinkedList<>(); for (CompletableFuture<GetResult> future : futures) { try { JsonObject json = future.join().contentAsObject(); results.add(json); } catch (CompletionException e) { e.printStackTrace(); results.add(null); // or skip / handle differently } } |
Veamos cómo funciona paso a paso.
-
- Obtener documentos
Aquí iteramos sobre las claves y para cada clave, llamamos a collection.async().get(clave, opciones), que devuelve un CompletableFuture y luego almacenamos todos esos futuros en una lista.
- Esperar a que terminen todas las búsquedas
CompletableFuture.allOf(...) crea un nuevo futuro que se completa cuando se completan todos los futuros de la matriz..join() bloquea el subproceso actual hasta que se hayan realizado todas las búsquedas asíncronas.
- Transformar resultados
Una vez realizadas todas las búsquedas, creamos otra lista para guardar los valores finales en formato plano List. Para cada CompletableFuture, recuperamos y transformamos el resultado. Dependiendo del requisito, puede manejar el error de fallo añadiendo null a la lista en lugar del resultado fallido o de un objeto marcador de error.
El paso de transformación asume que la obtención de los documentos se ha completado antes de transformar los resultados, sin embargo, si el objetivo es continuar encadenando operaciones asíncronas entonces podemos crear una lista if futures Lista<CompletableFuture> y envolver la transformación en otra envoltura asíncrona.
- Obtener documentos
Recomendamos utilizar esta API sólo si estás escribiendo código de integración para mecanismos de concurrencia de alto nivel o si realmente necesitas hasta la última gota de rendimiento. En todos los demás casos, la API reactiva (por la riqueza de operadores) es probablemente la mejor opción.
Conclusión
La programación reactiva ofrece una de las maneras más eficientes de lograr un alto rendimiento para las operaciones de obtención masiva con Couchbase. Su verdadero poder se desbloquea cuando se aplica a través de una pila completamente reactiva, donde el comportamiento no bloqueante y la escalabilidad están completamente optimizados.
Dicho esto, no necesitas una arquitectura completamente reactiva para empezar a cosechar los beneficios. Un primer paso práctico e impactante es migrar sólo la capa CRUD de Couchbase a reactiva. Hacerlo puede reducir drásticamente el bloqueo de hilos y minimizar el estrangulamiento de la CPU, lo que lleva a una mejor capacidad de respuesta del sistema y la utilización de los recursos sin necesidad de una revisión completa de la arquitectura.
Si el rendimiento y la escalabilidad son prioridades, la programación reactiva bien merece la inversión, incluso en una implementación parcial.
El autor agradece al equipo de Couchbase SDK y su excelente explicación sobre cómo podemos lograr la dosificación de manera eficiente sin la necesidad de una función genérica bulk get, gracias.
