Sistema de filas para manipular processos e mensagens distribuídas (microsserviços) com NodeJS e Bull

Fala devs! Tudo bem?

Meu nome é Vitor Brangioni, sou Co-Founder e responsável pela tecnologia da ufrilla.

Há um tempo atrás tivemos um probleminha técnico.

A ufrilla para quem não conhece é uma startup que conecta pessoas que querem trabalhar de freelancer ná área de eventos aos produtores de eventos, de forma muito simples. Além de ajudar esses produtores a fazerem toda essa gestão e operação, que é uma 'baita dor de cabeça'.  

O problema foi o seguinte: O contratante abre x vagas em nossa plataforma e os freelancers demonstram interesse em trabalhar, em seguida o contratante seleciona as pessoas que ele quer que trabalhe em seu evento e assim os primeiros que confirmarem de fato irão trabalhar. Exemplo: Se o contratante abrir 5 vagas de Bartender, ele poderá selecionar mais de 100 pessoas das que demonstraram interesse, mas somente as 5 primeiras que confirmarem irão trabalhar.

Como o número de usuários começou a crescer bem (+35mil usuários na época) com uma certa frequência havia freelancers confirmando na vaga ao mesmo tempo, acontecendo requisições simultâneas. E para ver se ainda há disponibilidade de vagas, temos que buscar o número de pessoas confirmadas no banco de dados e verificar se as vagas já foram preenchidas, se ainda houver disponibilidade, então confirmamos sua participação. Agora imagine várias dessas requisições sendo processadas ao mesmo tempo, ou em um intervalo muito curto, é como se houvessem várias requisições verificando se há disponibilidade na vaga ao mesmo tempo e ao verificar, de fato há disponibilidade na vaga. Então depois de verificarem que há disponibilidade, todas confirmam participação na vaga.

Exemplo: Temos 1 vaga de trabalho, 5 pessoas confirmaram participação ao mesmo tempo. Então temos 5 requisições diferentes para realizar a confirmação na vaga, porém todas essas requisições leem no banco de dados que há disponibilidade, todas ao mesmo tempo. Como todas verificaram que há disponibilidade, com isso todas vão confirmar participação. No final, todos os cinco freelancers estarão como confirmados em uma vaga, em vez de somente uma pessoa.

Esse problema deu uma 'dor de cabeça' para nossa equipe e provavelmente para os freelancers, pois tinhamos que cancelar com os freelancers que já estavam planejando todo seu dia (ou deveriam. rs) para trabalhar e ainda tudo de forma manual. 

A solução que encontrei para resolver esse problema foi aplicar filas no endpoint da API para confirmar participação na vaga. Enquanto houvesse uma requisição sendo processada, as outras estariam na fila esperando o processamento da requisição atual para depois serem processadas. Seguindo a regra de fila, que é a prioridade sobre ordem de chegada (First in, first out - FIFO).

Para facilitar o entendimento do problema, da solução e conseguirmos aplicar em vários contextos, criarei um exemplo bem simples. Vamos resolver o seguinte, temos que buscar um valor no banco de dados e somar +1 nesse valor e salvar novamente. Exemplo: Se o número do banco começar com 0 (zero) e a API receber mil requisições, então no final o número do banco será mil. Mas e se essas requisições forem simultâneas? O valor final ficará com o valor correto??

Vamos começar a implementar essa solução sem fila e veremos o que acontece. Mas antes disponibilizarei a modelagem do banco de dados e script para enviarmos várias requisições simultâneas na API.

Observação: Criei uma API em NodeJS para receber e processar as requisições, com os endpoints de 'somar +1' com fila e sem fila. Não vou mostrar aqui o código sobre a arquitetura da API, pois não é o foco, e sim o código chave sobre nossa solução. Caso queira visualizar o código todo, disponibilizarei o link do github.

MODELAGEM DO BANCO

CÓDIGO PARA ENVIAR VÁRIAS REQUISIÇÕES SIMULTÂNEAS

const axios = require("axios"); // package para enviar as requisições
const host = "http://localhost:3000/api/count";
const endpointWithQueue = `${host}/add-queue`; // endpoint com fila
const endpointWithoutQueue = `${host}/sum`; // endpoint sem fila
const nReqs = 500; // número de requisições para enviar
const reqs = []; // array para inserir as requisições
// Preparando array de requisições
for (let i = 0; i < nReqs; i++) {
  reqs.push(axios.post(endpointWithQueue, { sum: 1 })); // altere qual endpoint você quer testar, com fila ou sem fila.
}
// Enviando requisções para a api de forma simultânea.
Promise.all(reqs).then(
  (_) => console.log("SUCESSO! Todas as requisições foram enviadas."),
  (err) => console.log(err)
);

SOLUÇÃO SEM FILA

Endpoint da API para chamar o método de 'somar +1':

router.post('/sum', (req, res) => {
    controller.sum(req, res)
});

Método para adicionar +1 na coluna 'sum' do banco de dados:

const { Count } = require("./../../config/models");
exports.sum = async (req, res) => {
  let { sum } = req.body;
  this._sum(sum)
    .then((_) => res.sendStatus(200))
    .catch((err) => res.sendStatus(500));
};
exports._sum = async (sum) => {
  const myCount = await Count.findOne({ where: { id: 1 } });
  sum = myCount.sum + sum;
  return Count.update({ sum }, { where: { id: 1 } }).then(
    (rows) => {
      console.log(`${myCount.sum} + 1 = ${sum}`);
      return rows;
    },
    (err) => {
      console.log(err);
      throw err;
    }
  );
};

Ao realizar o envio de várias requisições simultâneas para esse end-point sem fila, irá notar que o valor no banco de dado estará totalmente errado do que esperavamos. Como enviamos 500 requisições simultâneas, esperavamos o valor "500" no banco de dados, mas o valor ficou somente "1".

SOLUÇÃO COM FILA

Para implementar a solução com fila, usei um package chamado de 'Bull' (https://github.com/OptimalBits/bull). É uma biblioteca que te ajuda com o controle de trabalhos distribuídos, ela fornece algumas soluções muito útil para esse tipo de trabalho, em que conseguimos realizar trabalhos em background, como filas com prioridades ( FIFO, LIFO e outras) e outras soluções. O 'Bull' utiliza o redis para armazenamento da fila, então caso sua aplicação 'caia' por algum motivo, após ela voltar ao ar, continuará executando os processos que encontram-se na fila. Em nosso caso, usaremos a solução de fila FIFO (First in, first out), ou seja, prioridade por ordem de chegada.

Código para end-points e processador da fila:

const { Router } = require("express");
const controller = require("./controller");
const router = new Router();
const Bull = require("bull");
const Queue = new Bull("Queue", { redis: { port: 6379, host: "redis" } });
router.post("/add-queue", (req, res) => {
  Queue.add({ ...req.body });
  return res.sendStatus(200);
});
router.post("/sum", (req, res) => {
  controller.sum(req, res);
});
Queue.process(async (job) => {
  const { sum } = job.data;
  return controller._sum(sum);
});
exports.router = router;

Ao enviarmos novamente as 500 requisições simultâneas, notaremos que agora o valor do banco ficará correto. Pois nossa aplicação agora organizou as requisições em uma fila, agora será executada uma requisição por vez. Ao entrar em nosso log da API, notaremos que o processo estará acontecendo em background:

Github: https://github.com/VitorBrangioni/http-requests-queue

Essa é a solução que encontrei para resolver esse problema, espero que esse conteúdo consiga ajudar vocês. Depois é só adequarem essa solução no problema que estão enfrentando. 

Me deem um feedback do que acharam dessa soluçao, se ajudou ou não. Mas independente, fiz de coração!! 🙂 

E claro... O que podemos melhorar nisso? Conhece alguma solução melhor? Se sim, compartilhe com a gente e juntos ficaremos melhor. Pois nada é melhor do que compartilhar conhecimentos 😉

Abraços e bora codar,
Vitor Brangioni.