Aprendiendo BullJS 🐃

Aprendiendo BullJS 🐃🔥

Es bastante común encontrarnos con un proyecto en nuestra vida de programador 🧑‍💻 en el que se nos presenten requerimientos muy específicos en cuanto a funciones y su ejecución. Por ejemplo, podría tratarse de funciones en JavaScript que necesiten ejecutarse un número determinado de veces, que deban reintentarse en caso de fallar, o que requieran priorización para saber cuál ejecutar primero, entre muchas otras cosas. En este contexto, es donde los sistemas de manejo de colas para NodeJS empiezan a tomar protagonismo. En este artículo, nos centraremos en uno en particular: BullJS 🐃.

BullJS es una librería soportada por Node.js diseñada para estos escenarios 🌟, pero que además persiste la información en una base de datos Redis 🗄️. No se queda ahí, también ofrece paralelismo de tareas, notificaciones entre productores y consumidores, y seguimiento del progreso de las tareas, entre otros.

El proyecto BullJS se autodefine como: “El sistema de colas más rápido y confiable basado en Redis para NodeJS, cuidadosamente escrito para garantizar estabilidad, solidez y atomicidad” 🚀.

Aquí encontrarás los enlaces a la documentación oficial 📚:

• Guide — Tu punto de entrada para comenzar a desarrollar con Bull
• Repositorio — Repositorio Oficial

Comenzando 📋🔧

Para instalar BullJS, primero necesitas tener NodeJS instalado y, después, ejecutar el siguiente comando:

npm install bull

Como mencionamos anteriormente, Bull necesita una base de datos Redis, ya que es el lugar donde se almacenan y administran los "jobs" y los mensajes. Si tienes Docker instalado en tu máquina 💻, puedes ejecutar:

$ docker run — name my-redis-container -p 6379:6379 -d redis

Esto iniciará una base de datos local de Redis que estará ejecutándose en 127.0.0.1:6379 🌐.

Introducción con un ejemplo 💪📖

Imagina que queremos implementar un "job" (en BullJS, las tareas se denominan "jobs") que consiste en lo siguiente: "7 días después de que alguien se suscriba a nuestro boletín de noticias, queremos enviarle un correo electrónico que contenga un enlace para calificar su experiencia de suscripción en nuestro sitio web" 💌.

BullJS tiene dos elementos principales que definen todo el ecosistema para trabajar: las colas o "Queues" y las tareas o "Jobs". Primero, veamos las colas:

Queues en BullJS 🚴‍♂️🚴‍♀️🚴‍♂️🚴‍♀️🚴‍♀️

Una cola es un objeto de JavaScript que puede producir y consumir jobs. En nuestro ejemplo, vamos a llamar a una cola newsLetterMail, pero tú puedes ponerle el nombre que desees. Al crear una instancia de una cola, debemos especificar el host y el puerto de nuestra base de datos Redis, ya que el predeterminado es 127.0.0.1:6379. A continuación, veamos cómo se vería esto:

// Importa la clase Queue de la biblioteca "bull"
import Queue from "bull";

// Crea una nueva instancia de Queue llamada "newsLetterMailQueue" para manejar el envío de correos de boletines informativos
const newsLetterMailQueue = new Queue("newsLetterMail", {
  // Configura la conexión con el servidor Redis, que se utilizará para almacenar y gestionar las tareas en la cola
  redis: {
    // Establece la dirección IP del servidor Redis (en este caso, la dirección IP local)
    host: "127.0.0.1",
    // Establece el puerto en el que está escuchando el servidor Redis (el puerto predeterminado es 6379)
    port: 6379,
  },
});

Observa que hemos importado Bull con el alias Queue y hemos creado la cola, pasándole dos argumentos: el nombre y un objeto con la configuración de Redis. Ahora que tenemos una cola, pasemos a los...

Jobs en BullJS 🚴‍♂️💼

Con nuestra Queue lista, creemos nuestro primer Job. Para esto, vamos a pasar un objeto con datos que contenga la dirección de correo electrónico a la que queremos enviar el email, además de algunas opciones. En este ejemplo, queremos procesar el job 7 días después de haber sido creado, y si el job falla, se intentará ejecutar tres veces

// Crea un objeto llamado "data" que contiene información para enviar un correo electrónico
const data = {
  // Establece la dirección de correo electrónico a la que se enviará el boletín informativo
  email: "foo@bar.com",
};

// Crea un objeto llamado "options" que contiene opciones de configuración para agregar una tarea a la cola
const options = {
  // Establece un retraso para la ejecución de la tarea, en este caso, 7 días (86400000 milisegundos por día)
  delay: 86400000 * 7,
  // Establece el número máximo de intentos para procesar la tarea en caso de que falle
  attempts: 3,
};

// Agrega una nueva tarea a la cola "newsLetterMailQueue" con la información del correo electrónico y las opciones configuradas
newsLetterMailQueue.add(data, options);

Para añadir un job a una cola, utilizamos la función add que viene en el objeto de JavaScript que nos devuelve la creación de la cola. Esto hace que BullJS añada el job a la base de datos con las opciones que hemos especificado.

Procesando un Job 🚵‍♂️🔄

Para procesar un Job, necesitamos especificar una función que pueda ser llamada por cada job en una cola, sin importar cuántos sean. Esta función se llama process y forma parte del objeto de la cola que hemos definido:

newsLetterMailQueue.process(async (job) => {
  await sendNewsLetterMailTo(job.data.email);
});

Hemos extraído la propiedad email del Job mediante job.data y luego llamamos a una función que se encarga de enviar el correo. Si esta función llega a fallar por algún error de JavaScript, BullJS controlará dicho error e intentará ejecutarlo de nuevo hasta un máximo de 3 veces o las veces que hayamos especificado en las opciones del Job.

Completando un Job 🥇🚴‍♂️🎉

Ahora, imaginemos que la ejecución ha finalizado. ¿Cómo podemos saber esto? O mejor aún, ¿cómo sabemos si algo falló? Cada vez que finalice el proceso de un Job, necesitamos resolver una promesa o ejecutar un callback. Veamos estas dos opciones:

newsLetterMailQueue.process(async (job, done) => {
  await sendNewsLetterMailTo(job.data.email);
  done(null, { message: "Email sent" });
});

En este ejemplo anterior, el callback done recibe dos parámetros: error y resultado. Como todo salió bien, hemos enviado el error en null y en el resultado un objeto con el mensaje de éxito.

newsLetterMailQueue.process(async (job) => {
  await sendNewsLetterMailTo(job.data.email);
  return Promise.resolve({ message: "Email sent" });
});

Ahora, utilizando promesas, tenemos la opción de retornar una promesa resuelta o fallida. En este caso, como queremos completar el job sin errores y completo, retornamos el resultado dentro del resolvede nuestra promesa a retornar.

Además, es posible notificar sobre el progreso de un Job mediante job.progress, ya que si tenemos alguna otra entidad escuchando por Jobs en una cola, será una excelente señal de notificación entre ambos sistemas 📡.

Por ejemplo, podríamos actualizar el progreso del Job de la siguiente manera:

newsLetterMailQueue.process(async (job) => {
  job.progress(50); // Actualiza el progreso al 50%await sendNewsLetterMailTo(job.data.email);
  job.progress(100); // Actualiza el progreso al 100%return Promise.resolve({ message: "Email sent" });
});

Esto nos permitirá mantener un seguimiento del progreso de cada Job y comunicarlo entre diferentes partes del sistema que estén interesadas en el estado de los Jobs.Manejando errores en Jobs con BullJS 🚫🚴‍♀️

Bull es una biblioteca muy útil para manejar colas y trabajos en Node.js. En este artículo, veremos cómo manejar errores, gestionar la concurrencia de trabajos y escuchar el estado de los trabajos utilizando BullJS.

🎯 Manejo de errores en un Job

Cuando trabajamos con BullJS, es importante saber que los bloques try-catch no funcionan dentro de la función .process(). Por lo tanto, debemos manejar los errores utilizando el objeto done o retornando una promesa rechazada. Aquí tienes un ejemplo:

newsLetterMailQueue.process(async (job, done) => {
  await sendNewsLetterMailTo(job.data.email);
  done(new Error("Algo salió muy mal"));
});

newsLetterMailQueue.process(async (job, done) => {
  await sendNewsLetterMailTo(job.data.email);
  return Promise.reject({ "message": "Algo salió muy mal" });
});

🏃‍♂️ Concurrencia de Jobs

BullJS nos permite manejar la concurrencia de jobs utilizando los procesadores de nuestro computador. Para ello, debemos colocar la función process en un archivo independiente y definir la concurrencia máxima de Jobs. Aquí tienes un ejemplo:

// path/to/funcion/file.jsconst processJob = async (job) => {
  // Do somethingawait sendNewsLetterMailTo(job.data.email);
};
module.exports = processJob;

De esta manera, BullJS puede ejecutar varios trabajos de manera simultánea sin que se bloqueen entre sí.

🎧 Escuchando el estado de Jobs

Una de las características más interesantes de BullJS es que podemos escuchar el estado y el resultado de un trabajo en la misma aplicación o en una aplicación externa. Para hacerlo, debemos crear una cola con el mismo nombre y escuchar un evento que nos indique cuándo se ha completado un trabajo.

import Queue from "bull";

const newsLetterMailQueue = new Queue("newsLetterMail", {
  redis: {
    host: "127.0.0.1",
    port: 6379,
  },
});

newsLetterMailQueue.on("global:completed", async (jobId, result) => {
  // aquí en result obtenemos el resultado que se envía y// además un identificador único al Jobawait sendSMS();
});

Dos cosas que son importantes notar aquí es 🕵️‍♂️:

i) El servidor que está escuchando por los mensajes debe tener BullJS instalado.

ii) Debe apuntar exactamente al mismo redis que está usando el otro servidor.

El evento 'global:completed' es el evento que se usa cross servidor es decir para servidores externos al que está procesando el job, pero si lo que quieres es hacerlo junto todo en el mismo servidor o proyecto, simplemente debes escuchar por el evento ‘completed’.

Además del evento completed hay una lista enorme de eventos por los que se puede escuchar, aquí veremos una lista:

.on('error', function(error) {// An error occured.}).on('waiting', function(jobId){// A Job is waiting to be processed as soon as a worker is idling.});.on('active', function(job, jobPromise){// A job has started. You can use `jobPromise.cancel()`` to abort it.}).on('stalled', function(job){// A job has been marked as stalled. This is useful for debugging job// workers that crash or pause the event loop.}).on('progress', function(job, progress){// A job's progress was updated!}).on('completed', function(job, result){// A job successfully completed with a `result`.}).on('failed', function(job, err){// A job failed with reason `err`!}).on('paused', function(){// The queue has been paused.}).on('resumed', function(job){// The queue has been resumed.}).on('cleaned', function(jobs, type) {// Old jobs have been cleaned from the queue. `jobs` is an array of cleaned// jobs, and `type` is the type of jobs cleaned.});.on('drained', function() {// Emitted every time the queue has processed all the waiting jobs (even if there can be some delayed jobs not yet processed)});.on('removed', function(job){// A job successfully removed.});

💡 Nota importante: El servidor que está escuchando por los mensajes debe tener BullJS instalado y debe apuntar al mismo Redis que está utilizando el otro servidor.

Existen muchos otros eventos que puedes escuchar, como 'error', 'waiting', 'active', 'stalled', 'progress', 'completed', 'failed', 'paused', 'resumed', 'cleaned', 'drained' y 'removed'. Puedes agregar "global:" a cualquiera de ellos para usarlos entre proyectos.

⛳️ Conclusiones

BullJS es una librería potente, optimizada y fácil de usar que te permite controlar tus tareas, concurrencia y notificaciones entre proyectos. Te animamos a probar BullJS y a explorar todos sus ejemplos.

💡 Ejercicios propuestos

1. Implementa un sistema de cola para procesar imágenes usando BullJS.
2. Añademanejo de errores y concurrencia a tu implementación del sistema de cola.
3. Crea una aplicación separada que escuche y responda a eventos de trabajos completados, mostrando notificaciones en tiempo real a los usuarios.

📝 Resumen en 3 puntos

1. BullJS nos permite manejar errores en trabajos mediante el objeto done o retornando una promesa rechazada.
2. Podemos gestionar la concurrencia de trabajos en BullJS colocando la función process en un archivo independiente y definiendo la concurrencia máxima de trabajos.
3. BullJS nos permite escuchar y responder a eventos de trabajos en la misma aplicación o en aplicaciones externas, facilitando la comunicación entre servidores.

¡Esperamos que este artículo te haya sido útil! Si tienes alguna duda o comentario, no dudes en dejarlo en la sección de comentarios a continuación. Si te gustó, comparte este artículo con tus amigos en las redes sociales utilizando los enlaces a continuación. ¡Hasta la próxima!