Como Lidar com Tarefas Assíncronas com Node.js e BullMQ

O autor escolheu a Sociedade de Engenheiras para receber uma doação como parte do programa Escreva para Doações.

Introdução

Aplicações web têm ciclos de pedido/resposta. Quando você visita um URL, o navegador envia uma solicitação para o servidor que está executando um aplicativo que processa dados ou realiza consultas no banco de dados. Enquanto isso acontece, o usuário fica aguardando até que o aplicativo retorne uma resposta. Para algumas tarefas, o usuário pode obter uma resposta rapidamente; para tarefas que consomem tempo, como processamento de imagens, análise de dados, geração de relatórios ou envio de e-mails, essas tarefas levam muito tempo para serem concluídas e podem retardar o ciclo de pedido/resposta. Por exemplo, suponha que você tenha um aplicativo onde os usuários fazem upload de imagens. Nesse caso, você pode precisar redimensionar, compactar ou converter a imagem para outro formato para preservar o espaço em disco do seu servidor antes de mostrar a imagem ao usuário. Processar uma imagem é uma tarefa intensiva em CPU, o que pode bloquear uma thread do Node.js até que a tarefa seja concluída. Isso pode levar alguns segundos ou minutos. Os usuários precisam aguardar a conclusão da tarefa para obter uma resposta do servidor.

Para evitar atrasos no ciclo de solicitação/resposta, você pode usar o bullmq, uma fila de tarefas (jobs) distribuída que permite descarregar tarefas demoradas de sua aplicação Node.js para o bullmq, liberando o ciclo de solicitação/resposta. Esta ferramenta permite que sua aplicação envie respostas rapidamente para o usuário enquanto o bullmq executa as tarefas de forma assíncrona em segundo plano e independentemente da sua aplicação. Para acompanhar os trabalhos, o bullmq utiliza o Redis para armazenar uma breve descrição de cada trabalho em uma fila. Um bullmq worker então desenfileira e executa cada trabalho na fila, marcando-o como completo quando terminar.

Neste artigo, você usará o bullmq para descarregar uma tarefa demorada para o plano de fundo, o que permitirá que uma aplicação responda rapidamente aos usuários. Primeiro, você criará uma aplicação com uma tarefa demorada sem usar o bullmq. Em seguida, você usará o bullmq para executar a tarefa de forma assíncrona. Por fim, você instalará um painel visual para gerenciar os trabalhos do bullmq em uma fila do Redis.

Pré-requisitos

Para seguir este tutorial, você precisará dos seguintes itens:

• Configuração do ambiente de desenvolvimento Node.js. Para o Ubuntu 22.04, siga nosso tutorial em Como Instalar o Node.js no Ubuntu 22.04. Para outros sistemas, consulte Como Instalar o Node.js e Criar um Ambiente de Desenvolvimento Local.

• Redis instalado no seu sistema. No Ubuntu 22, siga os Passos 1 a 3 em nosso tutorial em Como Instalar e Proteger o Redis no Ubuntu 22.04. Para outros sistemas, consulte nosso tutorial em Como Instalar e Proteger o Redis.

• Familiaridade com promessas e funções async/await, que você pode desenvolver no nosso tutorial Compreendendo o Event Loop, Callbacks, Promessas e Async/Await no JavaScript.

• Conhecimento básico sobre como usar o Express. Consulte o nosso tutorial sobre Como Começar com Node.js e Express.

• Familiaridade com JavaScript Embutido (EJS). Confira nosso tutorial sobre Como usar EJS para modelar seu aplicativo Node para mais detalhes.

• Entendimento básico de como processar imagens com sharp, que você pode aprender em nosso tutorial sobre Como processar imagens em Node.js com Sharp.

Passo 1 — Configurando o Diretório do Projeto

Neste passo, você irá criar um diretório e instalar as dependências necessárias para a sua aplicação. A aplicação que você irá construir neste tutorial permitirá que os usuários façam upload de uma imagem, que será então processada usando o pacote sharp. O processamento de imagem é intensivo em tempo e pode retardar o ciclo de solicitação/resposta, tornando a tarefa um bom candidato para o bullmq para ser descarregado para segundo plano. A técnica que você usará para descarregar a tarefa também funcionará para outras tarefas intensivas em tempo.

Para começar, crie um diretório chamado processador_de_imagens e navegue até o diretório:

mkdir image_processor && cd image_processor

Em seguida, inicialize o diretório como um pacote npm:

npm init -y

O comando cria um arquivo package.json. A opção -y diz ao npm para aceitar todas as configurações padrão.

Ao executar o comando, a saída será semelhante à seguinte:

Output
Wrote to /home/sammy/image_processor/package.json:

{
  "name": "image_processor",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "keywords": [],
  "author": "",
  "license": "ISC"
}

A saída confirma que o arquivo package.json foi criado. Propriedades importantes incluem o nome do seu aplicativo (name), o número da versão da sua aplicação (version) e o ponto de partida do seu projeto (main). Se você quiser saber mais sobre as outras propriedades, você pode revisar a documentação de package.json do npm.

A aplicação que você irá construir neste tutorial exigirá as seguintes dependências:

• express: um framework web para construir aplicativos web.
• express-fileupload: um middleware que permite que seus formulários façam upload de arquivos.
• sharp: uma biblioteca de processamento de imagens.
• ejs: uma linguagem de modelo que permite gerar marcação HTML com Node.js.
• bullmq: uma fila de tarefas distribuída.
• bull-board: um painel que se baseia no bullmq e exibe o status dos trabalhos com uma interface de usuário (UI) agradável.

Para instalar todas essas dependências, execute o seguinte comando:

npm install express express-fileupload sharp ejs bullmq @bull-board/express

Além das dependências que você instalou, você também usará a seguinte imagem mais tarde neste tutorial:

Use curl para baixar a imagem para o local de sua escolha em seu computador local

curl -O https://deved-images.nyc3.cdn.digitaloceanspaces.com/CART-68886/underwater.png

Você tem as dependências necessárias para construir um aplicativo Node.js que não possui bullmq, o que você fará em seguida.

Passo 2 — Implementando uma Tarefa Intensiva em Tempo Sem bullmq

Neste passo, você irá construir uma aplicação com Express que permite aos usuários fazer upload de imagens. A aplicação iniciará uma tarefa intensiva em tempo usando sharp para redimensionar a imagem em várias dimensões, que serão exibidas ao usuário após o envio de uma resposta. Este passo ajudará você a entender como tarefas intensivas em tempo afetam o ciclo de solicitação/resposta.

Usando o nano, ou seu editor de texto preferido, crie o arquivo index.js:

nano index.js

No seu arquivo index.js, adicione o seguinte código para importar as dependências:

const path = require("path");
const fs = require("fs");
const express = require("express");
const bodyParser = require("body-parser");
const sharp = require("sharp");
const fileUpload = require("express-fileupload");

Na primeira linha, você importa o módulo path para calcular caminhos de arquivo com o Node. Na segunda linha, você importa o módulo fs para interagir com diretórios. Em seguida, você importa o framework web express. Você importa o módulo body-parser para adicionar middleware para analisar dados em solicitações HTTP. Depois, você importa o módulo sharp para processamento de imagem. Por fim, você importa o express-fileupload para lidar com uploads de um formulário HTML.

Em seguida, adicione o seguinte código para implementar middleware na sua aplicação:

...
const app = express();
app.set("view engine", "ejs");
app.use(bodyParser.json());
app.use(
  bodyParser.urlencoded({
    extended: true,
  })
);

Primeiro, defina a variável app para uma instância do Express. Em seguida, utilizando a variável app, o método set() configura o Express para usar a linguagem de modelo ejs. Em seguida, adicione o middleware do módulo body-parser com o método use() para transformar dados JSON em solicitações HTTP em variáveis que podem ser acessadas com JavaScript. Na linha seguinte, faça o mesmo com a entrada codificada em URL.

Em seguida, adicione as seguintes linhas para incluir mais middleware para lidar com upload de arquivos e servir arquivos estáticos:

...
app.use(fileUpload());
app.use(express.static("public"));

Adicione middleware para analisar arquivos enviados chamando o método fileUpload(), e defina um diretório onde o Express procurará e servirá arquivos estáticos, como imagens e CSS.

Com o middleware configurado, crie uma rota que exibe um formulário HTML para fazer upload de uma imagem:

...
app.get("/", function (req, res) {
  res.render("form");
});

Aqui, utilize o método get() do módulo Express para especificar a rota / e o retorno de chamada que deve ser executado quando o usuário visita a página inicial ou a rota /. No retorno de chamada, invoque res.render() para renderizar o arquivo form.ejs no diretório views. Você ainda não criou o arquivo form.ejs ou o diretório views.

Para criá-lo, primeiro, salve e feche seu arquivo. No terminal, insira o seguinte comando para criar o diretório views no diretório raiz do seu projeto:

mkdir views

Mova-se para o diretório views:

cd views

Crie o arquivo form.ejs no seu editor:

nano form.ejs

No seu arquivo form.ejs, adicione o seguinte código para criar o formulário:

<!DOCTYPE html>
<html lang="en">
  <%- include('./head'); %>
  <body>
    <div class="home-wrapper">
      <h1>Image Processor</h1>
      <p>
        Resizes an image to multiple sizes and converts it to a
        <a href="https://en.wikipedia.org/wiki/WebP">webp</a> format.
      </p>
      <form action="/upload" method="POST" enctype="multipart/form-data">
        <input
          type="file"
          name="image"
          placeholder="Select image from your computer"
        />
        <button type="submit">Upload Image</button>
      </form>
    </div>
  </body>
</html>

Primeiro, você faz referência ao arquivo head.ejs, que ainda não criou. O arquivo head.ejs conterá o elemento HTML head que você pode referenciar em outras páginas HTML.

No tag body, você cria um formulário com os seguintes atributos:

• action especifica a rota para onde os dados do formulário devem ser enviados quando o formulário for enviado.
• method especifica o método HTTP para enviar dados. O método POST incorpora os dados em uma solicitação HTTP.
• encytype especifica como os dados do formulário devem ser codificados. O valor multipart/form-data permite que os elementos HTML input enviem dados de arquivo.

No elemento form, você cria uma tag input para enviar arquivos. Em seguida, você define o elemento button com o atributo type definido como submit, que permite enviar formulários.

Depois de terminar, salve e feche seu arquivo.

Em seguida, crie um arquivo head.ejs:

nano head.ejs

No seu arquivo head.ejs, adicione o seguinte código para criar a seção head do aplicativo:

<head>
  <meta charset="UTF-8" />
  <meta http-equiv="X-UA-Compatible" content="IE=edge" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>Image Processor</title>
  <link rel="stylesheet" href="css/main.css" />
</head>

Aqui, você faz referência ao arquivo main.css, que você irá criar no diretório public mais tarde neste passo. Esse arquivo conterá os estilos para esta aplicação. Por enquanto, você continuará configurando os processos para ativos estáticos.

Salve e feche o arquivo.

Para lidar com os dados enviados pelo formulário, você deve definir um método post no Express. Para fazer isso, retorne ao diretório raiz do seu projeto:

cd ..

Abra o seu arquivo `index.js` novamente:

nano index.js

No seu arquivo `index.js`, adicione as linhas destacadas para definir um método para lidar com envios de formulários na rota `/upload`:

app.get("/", function (req, res) {
  ...
});

app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);

});

Você utiliza a variável `app` para chamar o método `post()`, que irá lidar com o formulário enviado na rota `/upload`. Em seguida, você extrai os dados da imagem enviada na requisição HTTP para a variável `image`. Depois disso, você define uma resposta para retornar o código de status `400` se o usuário não enviar uma imagem.

Para configurar o processo para a imagem enviada, adicione o seguinte código destacado:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
});

Essas linhas representam como seu aplicativo irá processar a imagem. Primeiro, você remove a extensão da imagem enviada e salva o nome na variável `imageName`. Em seguida, você define a função `processImage()`. Esta função recebe o parâmetro `size`, cujo valor será usado para determinar as dimensões da imagem durante o redimensionamento. Na função, você invoca `sharp()` com `image.data`, que é um buffer contendo os dados binários da imagem enviada. `sharp` redimensiona a imagem de acordo com o valor do parâmetro `size`. Você utiliza o método `webp()` de `sharp` para converter a imagem para o formato de imagem `webp`. Então, você salva a imagem no diretório `public/images/`.

A lista subsequente de números define os tamanhos que serão usados para redimensionar a imagem enviada. Em seguida, você usa o método map() do JavaScript para invocar processImage() para cada elemento no array sizes, após o que ele retornará um novo array. Cada vez que o método map() chama a função processImage(), ela retorna uma promessa para o novo array. Você usa o método Promise.all() para resolvê-las.

As velocidades de processamento do computador variam, assim como o tamanho das imagens que um usuário pode enviar, o que pode afetar a velocidade de processamento da imagem. Para atrasar este código para fins de demonstração, insira as linhas destacadas para adicionar um loop de incremento intensivo em CPU e um redirecionamento para uma página que exibirá as imagens redimensionadas com as linhas destacadas:

...
app.post("/upload", async function (req, res) {
  ...
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  }

  res.redirect("/result");
});

O loop será executado 10 bilhões de vezes para incrementar a variável counter. Você invoca a função res.redirect() para redirecionar o aplicativo para a rota /result. A rota irá renderizar uma página HTML que exibirá as imagens no diretório public/images.

A rota /result ainda não existe. Para criá-la, adicione o código destacado no seu arquivo index.js:

...

app.get("/", function (req, res) {
 ...
});

app.get("/result", (req, res) => {
  const imgDirPath = path.join(__dirname, "./public/images");
  let imgFiles = fs.readdirSync(imgDirPath).map((image) => {
    return `images/${image}`;
  });
  res.render("result", { imgFiles });
});

app.post("/upload", async function (req, res) {
  ...
});

Você define a rota /result com o método app.get(). Na função, você define a variável imgDirPath com o caminho completo para o diretório public/images. Você usa o método readdirSync() do módulo fs para ler todos os arquivos no diretório fornecido. A partir daí, você encadeia o método map() para retornar um novo array com os caminhos das imagens prefixados com images/.

Finalmente, você chama res.render() para renderizar o arquivo result.ejs, que ainda não existe. Você passa a variável imgFiles, que contém um array com todos os caminhos relativos das imagens, para o arquivo result.ejs.

Salve e feche seu arquivo.

Para criar o arquivo result.ejs, retorne ao diretório views:

cd views

Crie e abra o arquivo result.ejs em seu editor:

nano result.ejs

No seu arquivo result.ejs, adicione as seguintes linhas para exibir imagens:

<!DOCTYPE html>
<html lang="en">
  <%- include('./head'); %>
  <body>
    <div class="gallery-wrapper">
      <% if (imgFiles.length > 0){%>
      <p>The following are the processed images:</p>
      <ul>
        <% for (let imgFile of imgFiles){ %>
        <li><img src=<%= imgFile %> /></li>
        <% } %>
      </ul>
      <% } else{ %>
      <p>
        The image is being processed. Refresh after a few seconds to view the
        resized images.
      </p>
      <% } %>
    </div>
  </body>
</html>

Primeiro, você referencia o arquivo head.ejs. Na tag body, você verifica se a variável imgFiles está vazia. Se tiver dados, você itera sobre cada arquivo e cria uma imagem para cada elemento do array. Se imgFiles estiver vazio, você imprime uma mensagem que diz ao usuário para Atualizar após alguns segundos para ver as imagens redimensionadas..

Salve e feche seu arquivo.

Em seguida, retorne ao diretório raiz e crie o diretório public que conterá seus ativos estáticos:

cd .. && mkdir public

Mova para o diretório public:

cd public

Crie um diretório images que irá manter as imagens enviadas:

mkdir images

Em seguida, crie o diretório css e navegue até ele:

mkdir css && cd css

No seu editor, crie e abra o arquivo main.css, que você referenciou anteriormente no arquivo head.ejs:

nano main.css

No seu arquivo main.css, adicione os seguintes estilos:

body {
  background: #f8f8f8;
}

h1 {
  text-align: center;
}

p {
  margin-bottom: 20px;
}

a:link,
a:visited {
  color: #00bcd4;
}

/** Estilos para o botão "Escolher Arquivo" **/
button[type="submit"] {
  background: none;
  border: 1px solid orange;
  padding: 10px 30px;
  border-radius: 30px;
  transition: all 1s;
}

button[type="submit"]:hover {
  background: orange;
}

/** Estilos para o botão "Enviar Imagem" **/
input[type="file"]::file-selector-button {
  border: 2px solid #2196f3;
  padding: 10px 20px;
  border-radius: 0.2em;
  background-color: #2196f3;
}

ul {
  list-style: none;
  padding: 0;
  display: flex;
  flex-wrap: wrap;
  gap: 20px;
}

.home-wrapper {
  max-width: 500px;
  margin: 0 auto;
  padding-top: 100px;
}

.gallery-wrapper {
  max-width: 1200px;
  margin: 0 auto;
}

Essas linhas irão estilizar elementos no aplicativo. Usando atributos HTML, você estiliza o plano de fundo do botão Escolher Arquivo com o código hex #2196f3 (um tom de azul) e a borda do botão Enviar Imagem para laranja. Você também estiliza os elementos na rota /result para torná-los mais apresentáveis.

Depois de terminar, salve e feche seu arquivo.

Volte para o diretório raiz do projeto:

cd ../..

Abra o arquivo index.js no seu editor:

nano index.js

No seu arquivo index.js, adicione o seguinte código, que irá iniciar o servidor:

...
app.listen(3000, function () {
  console.log("Server running on port 3000");
});

O arquivo completo index.js agora corresponderá ao seguinte:

const path = require("path");
const fs = require("fs");
const express = require("express");
const bodyParser = require("body-parser");
const sharp = require("sharp");
const fileUpload = require("express-fileupload");

const app = express();
app.set("view engine", "ejs");
app.use(bodyParser.json());
app.use(
  bodyParser.urlencoded({
    extended: true,
  })
);

app.use(fileUpload());

app.use(express.static("public"));

app.get("/", function (req, res) {
  res.render("form");
});

app.get("/result", (req, res) => {
  const imgDirPath = path.join(__dirname, "./public/images");
  let imgFiles = fs.readdirSync(imgDirPath).map((image) => {
    return `images/${image}`;
  });
  res.render("result", { imgFiles });
});

app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  }

  res.redirect("/result");
});

app.listen(3000, function () {
  console.log("Server running on port 3000");
});

Depois de terminar de fazer as alterações, salve e feche seu arquivo.

Execute o aplicativo usando o comando node:

node index.js

Você receberá uma saída como esta:

Output
Server running on port 3000

Esta saída confirma que o servidor está em execução sem problemas.

Abra seu navegador preferido e visite http://localhost:3000/.

ssh -L 3000:localhost:3000  your-non-root-user@yourserver-ip

Quando a página carregar, ela corresponderá ao seguinte:

Em seguida, pressione o botão Escolher Arquivo e selecione a imagem underwater.png em sua máquina local. O display mudará de Nenhum arquivo escolhido para underwater.png. Depois disso, pressione o botão Enviar Imagem. O aplicativo carregará por um tempo enquanto processa a imagem e executa o loop de incremento.

Assim que a tarefa terminar, a rota /result será carregada com as imagens redimensionadas:

Agora você pode parar o servidor com CTRL+C. O Node.js não recarrega automaticamente o servidor quando os arquivos são alterados, então você precisará parar e reiniciar o servidor sempre que atualizar os arquivos.

Agora você sabe como uma tarefa intensiva em tempo pode afetar o ciclo de solicitação/resposta de uma aplicação. Você executará a tarefa de forma assíncrona na próxima vez.

Passo 3 — Executando Tarefas que Consomem Tempo de Forma Assíncrona com bullmq

Neste passo, você irá transferir uma tarefa que consome muito tempo para o plano de fundo usando bullmq. Este ajuste irá liberar o ciclo de solicitação/resposta e permitir que seu aplicativo responda imediatamente aos usuários enquanto a imagem está sendo processada.

Para fazer isso, você precisa criar uma descrição sucinta da tarefa e adicioná-la a uma fila com bullmq. Uma fila é uma estrutura de dados que funciona de forma semelhante a uma fila na vida real. Quando as pessoas fazem fila para entrar em um espaço, a primeira pessoa da fila será a primeira a entrar no espaço. Qualquer pessoa que chegar depois se alinhará no final da fila e entrará no espaço após todos que a precederem na fila até que a última pessoa entre no espaço. Com o processo de primeiro a entrar, primeiro a sair (FIFO) da estrutura de dados de fila, o primeiro item adicionado à fila é o primeiro item a ser removido (desenfileirado). Com bullmq, um produtor irá adicionar um trabalho em uma fila, e um consumidor (ou trabalhador) irá remover um trabalho da fila e executá-lo.

A fila em bullmq está no Redis. Ao descrever um trabalho e adicioná-lo à fila, é criada uma entrada para o trabalho em uma fila do Redis. A descrição do trabalho pode ser uma string ou um objeto com propriedades que contenham dados mínimos ou referências aos dados que permitirão ao bullmq executar o trabalho posteriormente. Depois de definir a funcionalidade para adicionar trabalhos à fila, mova o código intensivo em tempo para uma função separada. Mais tarde, o bullmq chamará essa função com os dados armazenados na fila quando o trabalho for retirado da fila. Uma vez concluída a tarefa, o bullmq a marcará como concluída, retirará outro trabalho da fila e o executará.

Abra o index.js no seu editor:

nano index.js

No seu arquivo index.js, adicione as linhas destacadas para criar uma fila no Redis com o bullmq:

...
const fileUpload = require("express-fileupload");
const { Queue } = require("bullmq");

const redisOptions = { host: "localhost", port: 6379 };

const imageJobQueue = new Queue("imageJobQueue", {
  connection: redisOptions,
});

async function addJob(job) {
  await imageJobQueue.add(job.type, job);
}
...

Comece extraindo a classe Queue do bullmq, que é usada para criar uma fila no Redis. Em seguida, defina a variável redisOptions como um objeto com propriedades que a instância da classe Queue usará para estabelecer uma conexão com o Redis. Defina o valor da propriedade host como localhost porque o Redis está sendo executado na sua máquina local.

Em seguida, defina a variável imageJobQueue para uma instância da classe Queue, levando o nome da fila como seu primeiro argumento e um objeto como segundo argumento. O objeto possui uma propriedade connection com o valor definido como um objeto na variável redisOptions. Após instanciar a classe Queue, uma fila chamada imageJobQueue será criada no Redis.

Por fim, defina a função addJob() que você usará para adicionar um trabalho na imageJobQueue. A função recebe um parâmetro job contendo as informações sobre o trabalho (você chamará a função addJob() com os dados que deseja salvar em uma fila). Na função, invoque o método add() da imageJobQueue, levando o nome do trabalho como primeiro argumento e os dados do trabalho como segundo argumento.

Adicione o código destacado para chamar a função addJob() e adicionar um trabalho na fila:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  ...
  await addJob({
    type: "processUploadedImages",
    image: {
      data: image.data.toString("base64"),
      name: image.name,
    },
  });

  res.redirect("/result");
});
...

Aqui, você chama a função addJob() com um objeto que descreve o trabalho. O objeto possui o atributo type com um valor igual ao nome do trabalho. A segunda propriedade, image, é configurada como um objeto contendo os dados da imagem que o usuário carregou. Como os dados da imagem em image.data estão em um buffer (forma binária), você invoca o método toString() do JavaScript para convertê-lo em uma string que pode ser armazenada no Redis, o que definirá a propriedade data como resultado. A propriedade image é configurada com o nome da imagem carregada (incluindo a extensão da imagem).

Agora você definiu as informações necessárias para o bullmq executar este trabalho mais tarde. Dependendo do seu trabalho, você pode adicionar mais informações ou menos.

Salve e feche seu arquivo.

Em seguida, crie e abra o arquivo utils.js que conterá o código de processamento de imagem:

nano utils.js

No seu arquivo utils.js, adicione o seguinte código para definir a função de processamento de imagem:

const path = require("path");
const sharp = require("sharp");

function processUploadedImages(job) {
}

module.exports = { processUploadedImages };

Você importa os módulos necessários para processar imagens e calcular caminhos nas duas primeiras linhas. Em seguida, você define a função processUploadedImages(), que conterá a tarefa de processamento de imagem intensiva em tempo. Esta função recebe um parâmetro job que será preenchido quando o trabalhador buscar os dados do trabalho na fila e, em seguida, invoca a função processUploadedImages() com os dados da fila. Você também exporta a função processUploadedImages() para que possa referenciá-la em outros arquivos.

Salve e feche seu arquivo.

Volte ao arquivo index.js:

nano index.js

Copie as linhas destacadas do arquivo index.js, em seguida, exclua-as deste arquivo. Você precisará do código copiado momentaneamente, então salve-o na área de transferência. Se estiver usando o nano, você pode destacar essas linhas e clicar com o botão direito do mouse para copiá-las:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage))
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  };
...
  res.redirect("/result");
});

O método post para a rota upload agora será semelhante ao seguinte:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);

  await addJob({
    type: "processUploadedImages",
    image: {
      data: image.data.toString("base64"),
      name: image.name,
    },
  });

  res.redirect("/result");
});
...

Salve e feche este arquivo, em seguida, abra o arquivo utils.js:

nano utils.js

No seu arquivo utils.js, cole as linhas que você acabou de copiar para o retorno de chamada da rota /upload na função processUploadedImages:

...
function processUploadedImages(job) {
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  };
}
...

Agora que você moveu o código para processar uma imagem, você precisa atualizá-lo para usar os dados da imagem do parâmetro job da função processUploadedImages() que você definiu anteriormente.

Para fazer isso, adicione e atualize as linhas destacadas abaixo:

function processUploadedImages(job) {
  const imageFileData = Buffer.from(job.image.data, "base64");
  const imageName = path.parse(job.image.name).name;
  const processImage = (size) =>
    sharp(imageFileData)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);
  ...
}

Você converte a versão string dos dados da imagem de volta para binário com o método Buffer.from(). Em seguida, você atualiza path.parse() com uma referência ao nome da imagem salva na fila. Depois disso, você atualiza o método sharp() para receber os dados binários da imagem armazenados na variável imageFileData.

O arquivo completo utils.js agora corresponderá ao seguinte:

const path = require("path");
const sharp = require("sharp");

function processUploadedImages(job) {
  const imageFileData = Buffer.from(job.image.data, "base64");
  const imageName = path.parse(job.image.name).name;
  const processImage = (size) =>
    sharp(imageFileData)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  };
}

module.exports = { processUploadedImages };

Salve e feche seu arquivo, depois retorne para o index.js:

nano index.js

A variável sharp não é mais necessária como dependência, pois a imagem agora é processada no arquivo utils.js. Delete a linha destacada do arquivo:

const bodyParser = require("body-parser");
const sharp = require("sharp");
const fileUpload = require("express-fileupload");
const { Queue } = require("bullmq");
...

Salve e feche seu arquivo.

Você agora definiu a funcionalidade para criar uma fila no Redis e adicionar um trabalho. Você também definiu a função processUploadedImages() para processar imagens enviadas.

A tarefa restante é criar um consumidor (ou trabalhador) que irá retirar um trabalho da fila e chamar a função processUploadedImages() com os dados do trabalho.

Crie um arquivo worker.js no seu editor:

nano worker.js

No seu arquivo worker.js, adicione o seguinte código:

const { Worker } = require("bullmq");

const { processUploadedImages } = require("./utils");

const workerHandler = (job) => {
  console.log("Starting job:", job.name);
  processUploadedImages(job.data);
  console.log("Finished job:", job.name);
  return;
};

Na primeira linha, você importa a classe Worker de bullmq; quando instanciada, isso iniciará um trabalhador que retirará trabalhos da fila no Redis e os executará. Em seguida, você faz referência à função processUploadedImages() do arquivo utils.js para que o trabalhador possa chamar a função com os dados na fila.

Você define uma função workerHandler() que recebe um parâmetro job contendo os dados do trabalho na fila. Na função, você registra que o trabalho começou, então invoca processUploadedImages() com os dados do trabalho. Depois disso, você registra uma mensagem de sucesso e retorna null.

Para permitir que o trabalhador se conecte ao Redis, desenfileire um trabalho da fila e chame workerHandler() com os dados do trabalho, adicione as seguintes linhas ao arquivo:

...
const workerOptions = {
  connection: {
    host: "localhost",
    port: 6379,
  },
};

const worker = new Worker("imageJobQueue", workerHandler, workerOptions);

console.log("Worker started!");

Aqui, você define a variável workerOptions para um objeto contendo as configurações de conexão do Redis. Você define a variável worker para uma instância da classe Worker que recebe os seguintes parâmetros:

• imageJobQueue: o nome da fila de trabalho.
• workerHandler: a função que será executada após um trabalho ter sido desenfileirado da fila do Redis.
• workerOptions: as configurações de configuração do Redis que o trabalhador usa para estabelecer uma conexão com o Redis.

Finalmente, você registra uma mensagem de sucesso.

Depois de adicionar as linhas, salve e feche seu arquivo.

Agora você definiu a funcionalidade do trabalhador bullmq para desenfileirar trabalhos da fila e executá-los.

No seu terminal, remova as imagens no diretório public/images para começar do zero ao testar seu aplicativo:

rm public/images/*

Em seguida, execute o arquivo index.js:

node index.js

O aplicativo iniciará:

Output
Server running on port 3000

Agora você vai iniciar o trabalhador. Abra uma segunda sessão no terminal e navegue até o diretório do projeto:

cd image_processor/

Inicie o trabalhador com o seguinte comando:

node worker.js

O trabalhador será iniciado:

Output
Worker started!

Visite http://localhost:3000/ em seu navegador. Pressione o botão Escolher Arquivo e selecione o arquivo underwater.png do seu computador, em seguida, pressione o botão Enviar Imagem.

Você pode receber uma resposta instantânea que pede para atualizar a página após alguns segundos:

Alternativamente, você pode receber uma resposta instantânea com algumas imagens processadas na página, enquanto outras ainda estão sendo processadas:

Você pode atualizar a página algumas vezes para carregar todas as imagens redimensionadas.

Volte ao terminal onde seu trabalhador está sendo executado. Esse terminal terá uma mensagem que corresponde ao seguinte:

Output
Worker started!
Starting job: processUploadedImages
Finished job: processUploadedImages

A saída confirma que o bullmq executou o trabalho com sucesso.

Seu aplicativo ainda pode delegar tarefas demoradas mesmo se o trabalhador não estiver em execução. Para demonstrar isso, pare o trabalhador no segundo terminal com CTRL+C.

Na sua sessão inicial do terminal, pare o servidor Express e remova as imagens em public/images:

rm public/images/*

Depois disso, inicie o servidor novamente:

node index.js

No seu navegador, visite http://localhost:3000/ e faça o upload da imagem underwater.png novamente. Quando você for redirecionado para o caminho /result, as imagens não serão exibidas na página porque o trabalhador não está em execução:

Volte ao terminal onde você executou o trabalhador e inicie o trabalhador novamente:

node worker.js

A saída será semelhante ao seguinte, o que indica que o trabalho foi iniciado:

Output
Worker started!
Starting job: processUploadedImages

Depois que o trabalho for concluído e a saída incluir uma linha que diz Trabalho concluído: processarImagensEnviadas, atualize o navegador. As imagens agora serão carregadas:

Interrompa o servidor e o worker.

Agora você pode transferir uma tarefa intensiva em tempo para o plano de fundo e executá-la de forma assíncrona usando bullmq. No próximo passo, você configurará um painel para monitorar o status da fila.

Passo 4 — Adicionando um Painel para Monitorar Filas bullmq

Neste passo, você usará o pacote bull-board para monitorar os trabalhos na fila do Redis a partir de um painel visual. Este pacote criará automaticamente um painel de interface do usuário (UI) que exibe e organiza as informações sobre os trabalhos do bullmq armazenados na fila do Redis. Usando seu navegador, você pode monitorar os trabalhos que foram concluídos, estão aguardando ou falharam sem precisar abrir o Redis CLI no terminal.

Abra o arquivo index.js no seu editor de texto:

nano index.js

Adicione o código destacado para importar o bull-board:

...
const { Queue } = require("bullmq");
const { createBullBoard } = require("@bull-board/api");
const { BullMQAdapter } = require("@bull-board/api/bullMQAdapter");
const { ExpressAdapter } = require("@bull-board/express");
...

No código anterior, você importa o método createBullBoard() do bull-board. Você também importa BullMQAdapter, que permite o acesso do bull-board às filas do bullmq, e ExpressAdapter, que fornece funcionalidade para o Express exibir o painel de controle.

Em seguida, adicione o código destacado para conectar o bull-board com o bullmq:

...
async function addJob(job) {
  ...
}

const serverAdapter = new ExpressAdapter();
const bullBoard = createBullBoard({
  queues: [new BullMQAdapter(imageJobQueue)],
  serverAdapter: serverAdapter,
});
serverAdapter.setBasePath("/admin");

const app = express();
...

Primeiro, você define o serverAdapter como uma instância do ExpressAdapter. Em seguida, você invoca createBullBoard() para inicializar o painel de controle com os dados da fila do bullmq. Você passa para a função um argumento de objeto com as propriedades queues e serverAdapter. A primeira propriedade, queues, aceita um array das filas que você definiu com o bullmq, que é a imageJobQueue aqui. A segunda propriedade, serverAdapter, contém um objeto que aceita uma instância do adaptador do servidor Express. Depois disso, você define o caminho /admin para acessar o painel de controle com o método setBasePath().

Em seguida, adicione o middleware do serverAdapter para a rota /admin:

app.use(express.static("public"))

app.use("/admin", serverAdapter.getRouter());

app.get("/", function (req, res) {
  ...
});

O arquivo index.js completo será conforme o seguinte:

const path = require("path");
const fs = require("fs");
const express = require("express");
const bodyParser = require("body-parser");
const fileUpload = require("express-fileupload");
const { Queue } = require("bullmq");
const { createBullBoard } = require("@bull-board/api");
const { BullMQAdapter } = require("@bull-board/api/bullMQAdapter");
const { ExpressAdapter } = require("@bull-board/express");

const redisOptions = { host: "localhost", port: 6379 };

const imageJobQueue = new Queue("imageJobQueue", {
  connection: redisOptions,
});

async function addJob(job) {
  await imageJobQueue.add(job.type, job);
}

const serverAdapter = new ExpressAdapter();
const bullBoard = createBullBoard({
  queues: [new BullMQAdapter(imageJobQueue)],
  serverAdapter: serverAdapter,
});
serverAdapter.setBasePath("/admin");

const app = express();
app.set("view engine", "ejs");
app.use(bodyParser.json());
app.use(
  bodyParser.urlencoded({
    extended: true,
  })
);
app.use(fileUpload());

app.use(express.static("public"));

app.use("/admin", serverAdapter.getRouter());

app.get("/", function (req, res) {
  res.render("form");
});

app.get("/result", (req, res) => {
  const imgDirPath = path.join(__dirname, "./public/images");
  let imgFiles = fs.readdirSync(imgDirPath).map((image) => {
    return `images/${image}`;
  });
  res.render("result", { imgFiles });
});

app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);

  await addJob({
    type: "processUploadedImages",
    image: {
      data: Buffer.from(image.data).toString("base64"),
      name: image.name,
    },
  });

  res.redirect("/result");
});

app.listen(3000, function () {
  console.log("Server running on port 3000");
});

Depois de terminar de fazer as alterações, salve e feche o seu arquivo.

Execute o arquivo index.js:

node index.js

Volte para o seu navegador e visite http://localhost:3000/admin. O painel de controle será carregado:

Na dashboard, você pode revisar o tipo de trabalho, os dados que ele consome e mais informações sobre o trabalho. Você também pode alternar para outras abas, como a aba Concluídos para obter informações sobre os trabalhos concluídos, a aba Falhou para mais informações sobre os trabalhos que falharam, e a aba Pausados para mais informações sobre os trabalhos que foram pausados.

Agora você pode usar o painel bull-board para monitorar filas.

Conclusão

Neste artigo, você delegou uma tarefa intensiva em tempo para uma fila de trabalhos usando bullmq. Primeiro, sem usar bullmq, você criou um aplicativo com uma tarefa intensiva em tempo que possui um ciclo de solicitação/resposta lento. Em seguida, você usou bullmq para delegar a tarefa intensiva em tempo e executá-la de forma assíncrona, o que impulsiona o ciclo de solicitação/resposta. Depois disso, você usou bull-board para criar um painel para monitorar as filas do bullmq no Redis.

Você pode visitar a documentação do bullmq para aprender mais sobre as funcionalidades do bullmq não abordadas neste tutorial, como agendamento, priorização ou reexecução de trabalhos, e configuração de configurações de concorrência para os trabalhadores. Você também pode visitar a documentação do bull-board para aprender mais sobre as funcionalidades do painel.