Bull-Ish Your Queues With BullMQ

Hola, y bienvenidos a esta masterclass donde hablo sobre colas y Cola Booleana. Así que, primero que todo, comencemos desde quién soy. Soy Luca Del Pupo, un desarrollador de software senior en NearForm, un amante de JavaScript y TypeScript. En mi tiempo libre, trato de administrar mi canal de YouTube, pero sin éxito en el último año. Y también me encanta escribir publicaciones técnicas para TechP. También me encanta correr y hacer senderismo en mis hermosas Dolomitas, y me encanta cuidar a los gatos de mis vecinos.

Por cierto, comencemos con el tema. Así que hoy quiero contarles una historia, una historia de un equipo, uno de los mejores equipos que he visto en mi vida. Así que este equipo estaba trabajando en un proyecto, en un proyecto para un cliente, y en algún momento necesitaban manejar una cola simple. Una cola simple, así que necesitaban entender cómo manejar un sistema de colas en su proyecto. Así que no eran los dueños de la infraestructura, así que empezaron a hablar con el equipo de plataforma. Así que fueron al equipo de plataforma para pedir ayuda para tener una especie de sistema de colas dentro de su entorno. El equipo de plataforma comenzó a pensar en las posibilidades reales y, básicamente, el producto se ejecutaba dentro de AWS. Decidieron darles una cola AWS SQS, pero con un problema simple. Solo una cola.

Y no era suficiente para el equipo, porque sí, pidieron una cola, pero no solo una cola. Necesitaban una especie de sistema de colas que pudiera manejar diferentes colas dentro del proyecto y ayudarles a desacoplar o descentralizar algunas cosas. Así que este es básicamente el resultado. El equipo siempre estuvo ahí para satisfacer su necesidad y alcanzar su objetivo de obtener una cola, pero el equipo de plataforma dijo, sí, puedo darte una cola, pero solo una. Así que durante algunos días, la noche de este equipo fue un poco una pesadilla en realidad. Una lista de pesadillas durante los días siguientes. Así que esto es básicamente lo que sucedió durante la noche.

Cada compañero piensa que el compañero que es miembro del equipo estaba pensando en, no sé, algo más. Pero en realidad, cada compañero de equipo estaba pensando en cómo podemos manejar este sistema de colas dentro de WK. Así que en algún momento, empezaron a pensar en lo que ya tenían dentro del proyecto. Y ya había dos herramientas diferentes que podían usar para crear una especie de sistema de colas alrededor del proyecto. Una era Posgas SQL para la base de datos, y la otra era Elastic Cache, la implementación de Redis dentro de AWS. Ya tenían estas dos herramientas, así que no pidieron nada más al equipo de plataforma. Pero lo único que tenían que entender es cómo podían manejar un sistema de colas dentro de una de estas dos herramientas. Así que la primera solución posible fue pgboss.

Si no conoces pgboss, es un paquete npm simple que puedes instalar en tu aplicación node y básicamente crear una especie de sistema de colas sobre Posgas SQL. Y exponer una API para llamar a las colas y manejar las colas dentro de tu aplicación. Esta herramienta requiere un esquema de base de datos específico para manejar. Es una buena práctica y una mala práctica tener un esquema de base de datos específico para manejar todas las colas y no usar el común, para evitar mezclar los datos reales con las tablas de colas. Y esto significa que si quieres usar pgboss, aumentas la carga dentro de la base de datos.

Y uno de los problemas del equipo era que la base de datos ya estaba bajo presión en algunos momentos del día. Así que aumentar nuevamente la carga de la base de datos puede ser otra pesadilla para ellos. Así que decidieron evitar usar esta solución. Así que la única solución que queda es Elastic Cache y Redis. Y en este escenario particular, lo que brilla es BoolMQ. BoolMQ es una biblioteca npm simple que implementa un sistema de colas rápido y robusto sobre Redis. Y puedes usarlo para desacoplar tu código, puedes usarlo para básicamente crear una arquitectura de microservicios alrededor de tu aplicación, o cosas como esta. Y la API es realmente, realmente directa.

Solo una nota rápida. Hay dos implementaciones de bool. El bool sin mq es el más antiguo. Y funciona en modo de mantenimiento y también funciona muy bien en la versión actual de Redis sin ningún problema. Pero esta implementación no soporta lo mismo. BoolMQ es la nueva. La más rápida. Y básicamente esto utiliza el stream de Redis. Si no recuerdo mal, funciona desde Redis 6 en adelante. Y es la versión actual de BoolMQ. Así que si tienes una versión más antigua de Redis, tienes que usar bool, de lo contrario puedes usar BoolMQ. Si quieres migrar de uno a otro, no es simple porque la API ha cambiado, pero es factible sin ningún problema. Por cierto, tenemos la característica expuesta por BoolMQ. Así que BoolMQ, obviamente, expone colas. Así que puedes crear colas, agregar elementos a las colas y hacer todas las cosas que conoces con la cola. Luego, lo que puedes hacer es, usando el worker, puedes manejar los datos dentro de las colas para desencolar todos tus datos. Luego también puedes crear un trabajo usando BoolMQ y Redis.

Y la guinda del pastel es que también puedes crear el flujo. Así que puedes crear una especie de objeto que depende de diferentes colas. Y puedes básicamente crear el resultado basado en el resultado de cada cola. Déjame explicar mejor lo que dije. Imagina que necesitas crear una cotización de un producto. Este producto está compuesto de diferentes piezas y para obtener la cotización de cada pieza, necesitas llamar a una API diferente. Lo que puedes hacer es crear una especie de objeto que represente tu producto. Y para cada pieza, puedes tener una cola específica que maneje la cotización de cada pieza. Cuando todas las piezas de tu producto estén cotizadas, tienes la última cola que es la cola del producto que básicamente obtiene todos los resultados de las otras colas y crea el resultado de tu cotización y guarda los datos, lo que quieras. Puedes encontrar más sobre flujos en la documentación de BoolMQ.

Entonces, ¿qué puedes tener fuera de la caja? Paralelismo y concurrencia. Obviamente, si quieres trabajar con colas, necesitas concurrencia y paralelismo. Fuera de la caja, ambos ya están implementados. La concurrencia por defecto está deshabilitada y puedes habilitarla. Simplemente añade la concurrencia y el número de trabajadores que deseas dentro de tu proceso. Básicamente, dentro de tu proceso puedes tener diferentes trabajadores que usan la memoria y puedes crear cuántos trabajadores quieras. Obviamente hay una limitación, la limitación de la máquina donde se está ejecutando tu aplicación. Entonces, si quieres tener paralelismo, puedes ejecutar tu aplicación muchas veces y esto te permite tener múltiples instancias de tu aplicación y así el trabajador que trabaja en paralelo en la instancia del usuario.

Luego, otros beneficios. Hay dos beneficios importantes. El primero. Fuera de la caja, puedes tener OpenTelemetry ya configurado. Solo es cuestión de configurar OpenTelemetry con un paquete simple. Y si quieres un panel de UI de lo que sucede dentro de tus colas, puedes construir un panel simple ya creado por un paquete y exponerlo usando Fastify, Express o lo que quieras. Pero pasemos al código para mostrar lo que sucede con PullMQ. Así que, pasemos al código y déjame cambiar la rama. Vamos a la primera. Ok, este es el primer ejemplo. Este es un monorepo simple, ok, nada especial, con dos paquetes, el publicador y la consola. El publicador es un proceso node simple, que importa las colas, un UID aleatorio, para el inicio de sesión y el setTimeout.

Ahora podemos crear una cola PullMQ utilizando una nueva cola simple de PullMQ. Este es el nombre de tu cola. Y la segunda parte es la conexión al Redis. Esta es la cadena de conexión al Redis. En nuestro escenario, esto es una especie de seguimiento temporal falso con el ID del sensor. El PushTemperature es el método utilizado para empujar los datos dentro de la cola. La Temperature es el nombre de tu evento. Y el Message es el ID del sensor, el valor y el anuncio detectado.

Y luego tienes el LogReinfoMessage. Lo que sucede en el código, básicamente, es un método run que usando un while true cada 10 milisegundos empuja una nueva temperatura dentro de él. Y cuando quieras cerrar el proceso, el sistema también cierra la conexión con el Redis. ¿Qué sucede en el consumidor? En el consumidor, tienes un trabajador. En este caso, el trabajador es otra clase expuesta por PullMQ. Puedes tener pno, la conexión a la base de datos para guardar los datos dentro de la base de datos, nada especial.

Y aquí, lo que sucede es que tienes el mismo nombre de la cola. Así que aquí, mi cola es el mismo nombre que debes tener dentro de la cola. Y este es el trabajador. Así que el trabajador básicamente utiliza una promesa para manejar los datos. En este escenario, básicamente, obtiene los datos, añade el resultado dentro de la tabla de temperatura usando sensor temperature y detected ad. Ejecuta la consulta y bla bla bla. Como puedes ver aquí, la concurrencia es 1, pero puedes usar el número que prefieras para tener múltiples trabajadores dentro de la misma Europa.

Y como ya hemos visto en el publicador anterior, necesitamos cerrar la conexión cuando alguien cierra la instancia de tu proceso de Node.js. Así que lo que podemos hacer es ejecutar el publicador. Y como puedes ver, comienza a empujar datos dentro de la cola. Y luego también podemos ejecutar el consumidor. Y como puedes ver, comienza a recibir los datos. Luego también podemos, si quieres, crear muchos publicadores. Si quieres, comienza el publicador.

Y bla bla bla. Muy simple. Está bien, déjame cerrar esto e ir a la segunda demo.

Así que la segunda demo es bastante simple. Solo es para mostrarte cómo configurar OpenTelemetry. Así que si quieres tener OpenTelemetry y rastrear lo que sucede dentro de tus colas, solo necesitas tener, obviamente, necesitas configurar OpenTelemetry dentro de tu entorno de Node.js. En este caso, ya configuré Jager.

Así que Jager rastrea todo. Y este es el código para configurar OpenTelemetry. Dentro de tu código, solo necesitas usar BullMQ Hotel.

Y pasar en el campo de telemetría la nueva instancia BullHotel que rastrea la telemetría de tu cola. Es lo mismo también para el consumidor. Y como puedes ver, si ejecuto de nuevo ahora el publicador... Y ejecuto de nuevo el consumidor. Y ahora podemos saltar al navegador. Esta es la interfaz de usuario de Jager. Si actualizo y encuentro, como puedes ver, aquí está el consumidor, aquí está el productor.

Y si actualizo de nuevo, ahora puedo ver que este es el rastreador para el productor, así que el tiempo necesario para manejar el empuje dentro del publicador, o el productor. Y este es el tiempo necesario para manejar el consumidor dentro del consumidor. Y como puedes ver, puedes hacer lo que quieras dentro de tu OpenTelemetry.

Ahora, saltemos a la última demo. La última demo es bastante simple. Podemos ir a la interfaz de usuario y ver qué sucede. Está bien, habilito de nuevo el inicio y el publicador. Este paquete tiene otra hermosa aplicación de Fastify dentro de la interfaz de usuario. Así que si no conoces Fastify, es un marco simple para manejar API de Node.js en Node.

Y aquí está la configuración, cómo configurar Fastify. El adaptador de Fastify es un adaptador expuesto por BullMQ, BullBoard, para crear el adaptador para Fastify en este caso. Necesitamos cargar todas las colas dentro de este adaptador, básicamente, como puedes ver aquí. Y luego lo que sucede, necesitamos exponer la API slash UI para mostrar la interfaz de usuario dentro del servidor. Lo que sucede aquí, si ejecuto npm run start UI y vuelvo al navegador. Está bien, esta es la interfaz de usuario. Puedes hacer clic en el activo, los hijos. Puedes ir a una cola específica y ver el elemento completado. Como puedes ver, el número aumentó porque ahora el publicador y el consumidor todavía están funcionando. Puedes ver la falla, el priorizador, y así sucesivamente.

Y con eso, probablemente, también completamos la demo. Puedo volver a la diapositiva para completar, para resumir todo. Así que, conclusión. Está bien, lo que aprendiste hoy es que BullMQ es una solución fácil para manejar básicamente un sistema de colas sin demasiados problemas. Y es perfecto si no puedes tener otros servicios de colas en tu sistema. No quiero decir que sea la mejor solución de todas, pero es una buena solución posible. Está bien. Entonces, la API es realmente sencilla. También es fácil de escalar porque solo puedes crear Redis. Básicamente, si configuras Redis de la mejor manera, ya se escala por sí mismo. Y tu trabajo puede escalar creando una nueva instancia o creando una diferente concurrencia de tu trabajo.

Es realmente para la mayoría de los casos que necesitas usar una cola. Está bien. Y está muy bien documentado. Si necesitas algo, puedes saltar a la documentación y encontrar todo lo que necesitas dentro de ella.

Entonces, los contras, desafortunadamente, se basan en Redis. Así que si lo apagas, perderás los datos. Así que también necesitas crear un sistema confiable a partir de este problema. Y es otra biblioteca que aprender.

Está bien, completé mi charla. Este es el código QR para la diapositiva de la presentación. Este es el código QR para la demo que te mostraré. Y estos son mis contactos. Si quieres charlar conmigo, no dudes en enviarme un problema así. Y gracias de nuevo. Espero que hayas disfrutado de esta charla. Adiós.