Artigo 51 — Filas de Tarefas e Processamento Assíncrono
Por Prof. Ricardo Matos | Mestre em Programação
Módulo 9 — Tópicos Avançados
Introdução
Imagine que um usuário faz upload de uma planilha com 10.000 produtos para importar em massa. Processar isso sincronamente — na mesma requisição HTTP — significaria deixar o cliente esperando por minutos enquanto o servidor trabalha. Se a conexão cair no meio, todo o trabalho se perde. Se dois usuários fizerem isso ao mesmo tempo, o servidor engasga.
Ou imagine enviar um email de boas-vindas ao cadastrar. Conectar ao servidor SMTP, aguardar a resposta e só então responder ao usuário adiciona latência desnecessária. Se o SMTP estiver lento ou indisponível, o cadastro falha — mesmo que isso não devesse impedir o registro.
O problema central em ambos os casos é o mesmo: estamos misturando a requisição do usuário com trabalho que pode — e deve — acontecer de forma independente e assíncrona. Filas de tarefas (job queues) resolvem isso desacoplando a aceitação de um trabalho da sua execução.
O padrão produtor-consumidor
A ideia central de uma fila de tarefas é simples. Um produtor aceita o trabalho e o coloca em uma fila. Um ou mais consumidores (workers) processam as tarefas da fila independentemente, em paralelo, no seu próprio ritmo. O produtor pode responder imediatamente ao cliente — "seu trabalho foi enfileirado" — sem esperar o resultado.
Sem filas:
Cliente → POST /importar → [processa 10.000 itens... espera 3 min] → 200 OK
Com filas:
Cliente → POST /importar → [enfileira job] → 202 Accepted (imediato)
↓
[Worker processa em background]
↓
[Notifica o cliente via WebSocket quando terminar]As vantagens vão além da velocidade de resposta. Filas permitem controle de concorrência — você decide quantos workers processam em paralelo, evitando sobrecarregar serviços externos. Permitem retry automático com backoff exponencial quando um job falha. Permitem priorizar tarefas urgentes. E dão visibilidade sobre o estado de cada job — pendente, processando, concluído, falhou.
BullMQ — filas de tarefas com Redis
O BullMQ é a biblioteca mais robusta e popular para filas de tarefas em Node.js. Usa o Redis como backend — o Redis é extremamente rápido para esta finalidade, com suporte nativo a listas, pub/sub e atomicidade.
npm install bullmq ioredisA arquitetura do BullMQ tem três componentes principais. A Queue é o ponto de entrada — onde os jobs são adicionados. O Worker é o processador — consome jobs da fila e executa a lógica. O QueueEvents é o observador — permite reagir a eventos como job concluído ou falhado.
Configuração base
Antes de criar filas específicas, vamos centralizar a configuração da conexão Redis que todas as filas vão compartilhar.
// src/infrastructure/queue/redisConfig.js
// A configuração Redis é compartilhada entre todas as filas
// O BullMQ cria conexões próprias — não compartilhamos com o Socket.IO
// para evitar interferência entre os sistemas
const redisConfig = {
host: process.env.REDIS_HOST || 'localhost',
port: Number(process.env.REDIS_PORT) || 6379,
password: process.env.REDIS_PASSWORD || undefined,
// Configurações importantes para produção
maxRetriesPerRequest: null, // BullMQ requer este valor específico
enableReadyCheck: false, // evita erro ao conectar durante startup
// Reconnect automático com backoff exponencial
retryStrategy: (tentativas) => {
if (tentativas > 10) return null; // desiste após 10 tentativas
return Math.min(tentativas * 200, 5000); // espera até 5s entre tentativas
},
};
module.exports = { redisConfig };// src/infrastructure/queue/filas.js
// Define e exporta todas as filas da aplicação
// Centralizar aqui evita nomes duplicados e facilita listar todas as filas
const { Queue } = require('bullmq');
const { redisConfig } = require('./redisConfig');
// Cada fila tem um nome único — o nome é a chave no Redis
// Agrupe jobs relacionados na mesma fila para simplificar o código dos workers
const filaEmail = new Queue('email', {
connection: redisConfig,
defaultJobOptions: {
// Tenta até 3 vezes antes de mover para a fila de falhados
attempts: 3,
// Backoff exponencial: 1s, 2s, 4s entre tentativas
// Evita bombardear um serviço que está temporariamente indisponível
backoff: {
type: 'exponential',
delay: 1000,
},
// Remove jobs bem-sucedidos após 24h — libera memória no Redis
removeOnComplete: { age: 86400, count: 100 },
// Mantém jobs falhados por 7 dias para análise
removeOnFail: { age: 604800 },
},
});
const filaImportacao = new Queue('importacao', {
connection: redisConfig,
defaultJobOptions: {
// Importações têm mais tentativas — são operações longas e mais suscetíveis
attempts: 2,
backoff: { type: 'exponential', delay: 5000 },
removeOnComplete: { age: 3600 }, // 1 hora — importações são únicas
removeOnFail: { age: 604800 },
},
});
const filaNotificacao = new Queue('notificacao', {
connection: redisConfig,
defaultJobOptions: {
attempts: 5, // notificações têm mais tentativas — importante entregá-las
backoff: { type: 'exponential', delay: 2000 },
removeOnComplete: { age: 3600 },
removeOnFail: { age: 604800 },
},
});
const filaRelatorio = new Queue('relatorio', {
connection: redisConfig,
defaultJobOptions: {
attempts: 2,
backoff: { type: 'fixed', delay: 10000 },
removeOnComplete: { age: 86400 },
removeOnFail: { age: 604800 },
},
});
module.exports = { filaEmail, filaImportacao, filaNotificacao, filaRelatorio };Workers — processando os jobs
Cada worker é um processo ou módulo separado que fica ouvindo a fila e processando jobs conforme eles chegam. Em produção, workers podem rodar em processos completamente separados do servidor HTTP — isso isola falhas e permite escalar independentemente.
// src/workers/emailWorker.js
// Worker responsável por processar todos os jobs de email
// Este arquivo pode rodar como processo separado: node src/workers/emailWorker.js
const { Worker } = require('bullmq');
const nodemailer = require('nodemailer');
const { redisConfig } = require('../infrastructure/queue/redisConfig');
// Configura o transportador de email uma vez — reutilizado por todos os jobs
const transportador = nodemailer.createTransport({
host: process.env.SMTP_HOST,
port: Number(process.env.SMTP_PORT) || 587,
secure: false,
auth: {
user: process.env.SMTP_USER,
pass: process.env.SMTP_PASS,
},
});
// Mapa de templates — cada tipo de email tem sua própria função de renderização
// Separar templates do worker facilita manutenção e testes
const templates = {
boasVindas: (dados) => ({
subject: `Bem-vindo(a), ${dados.nome}!`,
html: `
<h1>Olá, ${dados.nome}!</h1>
<p>Sua conta foi criada com sucesso.</p>
<p>Acesse: <a href="${process.env.FRONTEND_URL}/login">Entrar</a></p>
`,
}),
tarefaCriada: (dados) => ({
subject: `Nova tarefa: ${dados.tarefa.titulo}`,
html: `
<h2>Tarefa criada com sucesso</h2>
<p><strong>${dados.tarefa.titulo}</strong></p>
${dados.tarefa.prazo ? `<p>Prazo: ${new Date(dados.tarefa.prazo).toLocaleDateString('pt-BR')}</p>` : ''}
<p><a href="${process.env.FRONTEND_URL}/tarefas">Ver minhas tarefas</a></p>
`,
}),
redefinirSenha: (dados) => ({
subject: 'Redefinição de senha',
html: `
<h2>Redefinição de senha</h2>
<p>Clique no link abaixo para redefinir sua senha:</p>
<p><a href="${process.env.FRONTEND_URL}/nova-senha?token=${dados.token}">
Redefinir senha
</a></p>
<p>O link expira em 1 hora.</p>
<p>Se não foi você, ignore este email.</p>
`,
}),
};
// O worker escuta a fila 'email' e processa um job por vez (concurrency: 1)
// Para emails, processamento serial evita sobrecarregar o SMTP
// Aumente a concorrência para outros tipos de jobs se necessário
const worker = new Worker(
'email',
async (job) => {
const { tipo, para, dados } = job.data;
// Registra o progresso — visível no painel do BullMQ
await job.updateProgress(10);
// Valida que temos um template para este tipo de email
const gerarTemplate = templates[tipo];
if (!gerarTemplate) {
// Lançar erro aqui faz o BullMQ tentar novamente (até o limite de attempts)
throw new Error(`Template de email desconhecido: ${tipo}`);
}
const template = gerarTemplate(dados);
await job.updateProgress(30);
// Envia o email — pode lançar erro se o SMTP estiver indisponível
// BullMQ vai retentar automaticamente com o backoff configurado
const resultado = await transportador.sendMail({
from: `"${process.env.EMAIL_NOME || 'Gestão App'}" <${process.env.EMAIL_FROM}>`,
to: para,
subject: template.subject,
html: template.html,
});
await job.updateProgress(100);
// O retorno do processador é salvo no job — útil para debugging
return {
messageId: resultado.messageId,
aceito: resultado.accepted,
};
},
{
connection: redisConfig,
concurrency: 1, // processa um email por vez para não sobrecarregar SMTP
// Limita a taxa de processamento: no máximo 10 emails por segundo
// Útil para respeitar limites de API de provedores de email
limiter: {
max: 10,
duration: 1000,
},
}
);
// Eventos do worker para logging e monitoramento
worker.on('completed', (job) => {
console.log(`[EmailWorker] ✅ Job ${job.id} (${job.data.tipo}) concluído`);
});
worker.on('failed', (job, erro) => {
console.error(
`[EmailWorker] ❌ Job ${job?.id} (${job?.data?.tipo}) falhou: ${erro.message}`
);
// Em produção, você enviaria este erro para o Sentry
});
worker.on('error', (erro) => {
console.error('[EmailWorker] Erro no worker:', erro.message);
});
console.log('[EmailWorker] 🚀 Aguardando jobs na fila "email"...');
module.exports = worker; // exporta para testes// src/workers/importacaoWorker.js
// Worker para importação de dados em massa — jobs de longa duração
const { Worker } = require('bullmq');
const { parse } = require('csv-parse/sync');
const { redisConfig } = require('../infrastructure/queue/redisConfig');
const Produto = require('../infrastructure/database/models/ProdutoModel');
const { notificarUsuario } = require('../socketHandlers/notificacoes');
// 'io' é injetado na criação — permite notificar o usuário via WebSocket
function criarImportacaoWorker(io) {
const worker = new Worker(
'importacao',
async (job) => {
const { usuarioId, conteudoCSV, nomeArquivo } = job.data;
const resultados = { criados: 0, atualizados: 0, erros: [] };
await job.updateProgress(5);
// Parseia o CSV recebido como string
let registros;
try {
registros = parse(conteudoCSV, {
columns: true, // usa a primeira linha como cabeçalho
skip_empty_lines: true,
trim: true,
});
} catch (erro) {
throw new Error(`CSV inválido: ${erro.message}`);
}
const total = registros.length;
console.log(`[ImportacaoWorker] Processando ${total} registros de ${nomeArquivo}`);
// Processa em lotes de 100 para não sobrecarregar o banco
// e para poder reportar progresso gradualmente
const TAMANHO_LOTE = 100;
for (let i = 0; i < registros.length; i += TAMANHO_LOTE) {
const lote = registros.slice(i, i + TAMANHO_LOTE);
// Processa cada registro do lote com tratamento de erro individual
// Um registro inválido não deve abortar toda a importação
for (const registro of lote) {
try {
// Upsert: cria ou atualiza o produto pelo código
await Produto.findOneAndUpdate(
{ codigo: registro.codigo },
{
nome: registro.nome,
preco: parseFloat(registro.preco),
estoque: parseInt(registro.estoque, 10),
categoria: registro.categoria,
},
{ upsert: true, new: true, runValidators: true }
);
resultados.criados++;
} catch (erro) {
// Registra o erro mas continua o processamento
resultados.erros.push({
linha: i + lote.indexOf(registro) + 2, // +2: 1-indexed + header
codigo: registro.codigo,
erro: erro.message,
});
}
}
// Atualiza o progresso proporcionalmente ao lote concluído
const progresso = Math.round(((i + lote.length) / total) * 90) + 5;
await job.updateProgress(progresso);
// Notifica o usuário do progresso via WebSocket
if (io) {
notificarUsuario(io, usuarioId, {
tipo: 'importacao:progresso',
mensagem: `Importando... ${progresso}%`,
jobId: job.id,
});
}
}
await job.updateProgress(100);
// Notifica o usuário que a importação terminou
if (io) {
notificarUsuario(io, usuarioId, {
tipo: 'importacao:concluida',
mensagem: `Importação concluída: ${resultados.criados} produtos, ${resultados.erros.length} erros`,
jobId: job.id,
resultados,
});
}
return resultados;
},
{
connection: redisConfig,
// Importações são pesadas — processa uma por vez para não travar o banco
concurrency: 1,
}
);
worker.on('failed', (job, erro) => {
console.error(`[ImportacaoWorker] ❌ Job ${job?.id} falhou:`, erro.message);
if (io && job?.data?.usuarioId) {
notificarUsuario(io, job.data.usuarioId, {
tipo: 'importacao:falhou',
mensagem: `Importação falhou: ${erro.message}`,
jobId: job.id,
});
}
});
return worker;
}
module.exports = { criarImportacaoWorker };Produtores — enfileirando jobs a partir da API
Agora que os workers existem, precisamos enfileirar jobs a partir dos controllers e dos eventos de domínio.
// src/infrastructure/queue/produtores.js
// Funções de alto nível para enfileirar cada tipo de job
// Encapsular a lógica aqui evita que os controllers conheçam o BullMQ diretamente
const { filaEmail, filaImportacao, filaNotificacao } = require('./filas');
// Enfileira um email para envio assíncrono
// Retorna imediatamente — o envio acontece em background
async function enfileirarEmail(tipo, para, dados, opcoes = {}) {
const job = await filaEmail.add(
tipo, // nome do job — aparece no painel de monitoramento
{ tipo, para, dados },
{
// Jobs de email de redefinição de senha têm prioridade maior
priority: opcoes.prioridade || 0,
// Atraso opcional — útil para emails de "lembrete" agendados
delay: opcoes.atrasoMs || 0,
// ID único para evitar duplicatas — útil para emails de boas-vindas
jobId: opcoes.jobId,
}
);
console.log(`[Fila] Email enfileirado: ${tipo} → ${para} (job ${job.id})`);
return job;
}
// Enfileira uma importação de CSV
async function enfileirarImportacao(usuarioId, conteudoCSV, nomeArquivo) {
const job = await filaImportacao.add(
'importar-csv',
{ usuarioId, conteudoCSV, nomeArquivo },
{
// Importações podem demorar muito — aumenta o timeout
// O default do BullMQ é sem timeout — ajuste conforme necessário
}
);
console.log(`[Fila] Importação enfileirada: ${nomeArquivo} (job ${job.id})`);
return job;
}
// Enfileira geração de relatório — pode ser agendado para horário de baixo tráfego
async function enfileirarRelatorio(usuarioId, tipo, filtros, agendar) {
const opcoes = {};
if (agendar) {
// Agenda para um horário específico — útil para relatórios diários
const agora = new Date();
const horarioAlvo = new Date(agendar);
const atrasoMs = horarioAlvo.getTime() - agora.getTime();
if (atrasoMs > 0) {
opcoes.delay = atrasoMs;
}
}
return filaRelatorio.add('gerar-relatorio', { usuarioId, tipo, filtros }, opcoes);
}
module.exports = { enfileirarEmail, enfileirarImportacao, enfileirarRelatorio };// Integrando nos eventos de domínio — container.js
// Quando uma tarefa é criada, enfileira o email automaticamente
// sem que o caso de uso precise saber sobre filas ou email
const { enfileirarEmail } = require('./infrastructure/queue/produtores');
const EVENTOS = require('./domain/events/EventosDominio');
eventEmitter.on(EVENTOS.TAREFA_CRIADA, async ({ tarefa, usuario }) => {
// Invalida cache (síncrono)
cache.del(`stats:${usuario._id}`);
// Enfileira email (assíncrono — não bloqueia a resposta HTTP)
await enfileirarEmail(
'tarefaCriada',
usuario.email,
{ tarefa: tarefa.toJSON(), nome: usuario.nome },
// ID único para evitar email duplicado se o evento for emitido duas vezes
{ jobId: `tarefa-criada:${tarefa.id}` }
);
});// src/interfaces/http/routes/importacao.js
// Rota para upload de CSV — responde imediatamente após enfileirar
const express = require('express');
const multer = require('multer'); // npm install multer
const { autenticar } = require('../middlewares/auth');
const { enfileirarImportacao } = require('../../../infrastructure/queue/produtores');
const { filaImportacao } = require('../../../infrastructure/queue/filas');
const router = express.Router();
// Multer armazena o arquivo em memória — para arquivos grandes, use disco
const upload = multer({
storage: multer.memoryStorage(),
limits: { fileSize: 10 * 1024 * 1024 }, // 10MB máximo
fileFilter: (req, file, cb) => {
if (file.mimetype !== 'text/csv' && !file.originalname.endsWith('.csv')) {
return cb(new Error('Apenas arquivos CSV são aceitos.'));
}
cb(null, true);
},
});
router.post('/produtos', autenticar, upload.single('arquivo'), async (req, res, next) => {
try {
if (!req.file) {
return res.status(400).json({ erro: 'Arquivo CSV não enviado.' });
}
const conteudoCSV = req.file.buffer.toString('utf-8');
const nomeArquivo = req.file.originalname;
// Enfileira e responde IMEDIATAMENTE — não espera o processamento
const job = await enfileirarImportacao(
req.usuario._id.toString(),
conteudoCSV,
nomeArquivo
);
// 202 Accepted — trabalho aceito mas ainda não concluído
res.status(202).json({
mensagem: 'Importação enfileirada com sucesso.',
jobId: job.id,
// O cliente pode usar este endpoint para verificar o progresso
statusUrl: `/importacao/status/${job.id}`,
});
} catch (erro) {
next(erro);
}
});
// Endpoint para verificar o status de um job de importação
router.get('/status/:jobId', autenticar, async (req, res, next) => {
try {
const job = await filaImportacao.getJob(req.params.jobId);
if (!job) {
return res.status(404).json({ erro: 'Job não encontrado.' });
}
// Verifica se o job pertence ao usuário autenticado
if (job.data.usuarioId !== req.usuario._id.toString()) {
return res.status(403).json({ erro: 'Sem acesso a este job.' });
}
const estado = await job.getState();
const progresso = job.progress;
res.json({
jobId: job.id,
estado, // 'waiting', 'active', 'completed', 'failed', 'delayed'
progresso, // 0-100
resultado: estado === 'completed' ? job.returnvalue : null,
erro: estado === 'failed' ? job.failedReason : null,
criadoEm: new Date(job.timestamp).toISOString(),
});
} catch (erro) {
next(erro);
}
});
module.exports = router;Jobs repetidos — agendamento com cron
O BullMQ suporta jobs que se repetem automaticamente em um horário definido, similar ao cron do Unix. Isso é perfeito para relatórios diários, limpeza de dados antigos e alertas periódicos.
// src/infrastructure/queue/agendamentos.js
// Jobs recorrentes — configurados uma vez no startup da aplicação
const { filaEmail, filaRelatorio, filaNotificacao } = require('./filas');
async function configurarAgendamentos() {
// Remove agendamentos antigos antes de recriar
// Evita duplicatas se a aplicação reiniciar
await filaRelatorio.obliterate({ force: true }).catch(() => {});
await filaNotificacao.obliterate({ force: true }).catch(() => {});
// Relatório diário de tarefas — todo dia às 8h (horário de Brasília)
await filaRelatorio.add(
'relatorio-diario',
{ tipo: 'tarefas-do-dia' },
{
repeat: {
pattern: '0 8 * * *', // cron: minuto hora dia mês diaSemana
tz: 'America/Sao_Paulo', // fuso horário
},
jobId: 'relatorio-diario', // ID fixo para evitar duplicatas
}
);
// Verificação de tarefas atrasadas — toda hora
await filaNotificacao.add(
'verificar-tarefas-atrasadas',
{ tipo: 'verificacao-atraso' },
{
repeat: { pattern: '0 * * * *' },
jobId: 'verificar-atraso',
}
);
// Limpeza de tokens expirados — todo domingo à meia-noite
await filaEmail.add(
'limpeza-tokens',
{ tipo: 'limpeza-interna' },
{
repeat: { pattern: '0 0 * * 0' },
jobId: 'limpeza-tokens',
}
);
console.log('[Agendamentos] ✅ Jobs recorrentes configurados.');
}
module.exports = { configurarAgendamentos };// src/workers/notificacaoWorker.js
// Processa jobs de notificação — incluindo o job recorrente de tarefas atrasadas
const { Worker } = require('bullmq');
const { redisConfig } = require('../infrastructure/queue/redisConfig');
const Tarefa = require('../infrastructure/database/models/TarefaModel');
const Usuario = require('../infrastructure/database/models/UsuarioModel');
const { enfileirarEmail } = require('../infrastructure/queue/produtores');
function criarNotificacaoWorker(io) {
return new Worker(
'notificacao',
async (job) => {
const { tipo } = job.data;
if (tipo === 'verificacao-atraso') {
// Busca todas as tarefas pendentes com prazo passado
const atrasadas = await Tarefa.find({
status: { $in: ['pendente', 'em_progresso'] },
prazo: { $lt: new Date() },
// Evita notificar múltiplas vezes — usa um campo de controle
notificacaoAtrasoEnviada: { $ne: true },
})
.populate('usuario', 'email nome')
.lean();
console.log(`[NotificacaoWorker] ${atrasadas.length} tarefas atrasadas encontradas.`);
for (const tarefa of atrasadas) {
if (!tarefa.usuario?.email) continue;
// Enfileira um email para cada tarefa atrasada
await enfileirarEmail(
'tarefaAtrasada',
tarefa.usuario.email,
{ tarefa, nome: tarefa.usuario.nome },
{ jobId: `atraso:${tarefa._id}` }
);
// Marca como notificado para não enviar novamente
await Tarefa.findByIdAndUpdate(tarefa._id, {
notificacaoAtrasoEnviada: true,
});
}
return { tarefasNotificadas: atrasadas.length };
}
},
{ connection: redisConfig, concurrency: 2 }
);
}
module.exports = { criarNotificacaoWorker };Monitoramento — Bull Board
O Bull Board é uma interface web que permite visualizar e gerenciar todas as filas em tempo real — ver jobs pendentes, ativos, concluídos e falhados, inspecionar dados e reintentar jobs manualmente.
// npm install @bull-board/express @bull-board/api
const { createBullBoard } = require('@bull-board/api');
const { BullMQAdapter } = require('@bull-board/api/bullMQAdapter');
const { ExpressAdapter } = require('@bull-board/express');
const { filaEmail, filaImportacao, filaNotificacao, filaRelatorio } = require('./filas');
function configurarBullBoard(app) {
const adaptadorExpress = new ExpressAdapter();
// Rota base do painel — proteja com autenticação em produção!
adaptadorExpress.setBasePath('/admin/filas');
createBullBoard({
queues: [
new BullMQAdapter(filaEmail),
new BullMQAdapter(filaImportacao),
new BullMQAdapter(filaNotificacao),
new BullMQAdapter(filaRelatorio),
],
serverAdapter: adaptadorExpress,
});
// Middleware simples de autenticação para o painel
// Em produção, use autenticação mais robusta
app.use(
'/admin/filas',
(req, res, next) => {
const senhaAdmin = req.headers['x-admin-key'];
if (senhaAdmin !== process.env.ADMIN_KEY) {
return res.status(401).json({ erro: 'Não autorizado.' });
}
next();
},
adaptadorExpress.getRouter()
);
console.log('[BullBoard] 📊 Painel de filas disponível em /admin/filas');
}
module.exports = { configurarBullBoard };Front-end — acompanhando o progresso de jobs
// src/hooks/useProgressoImportacao.js
// Hook para acompanhar o progresso de um job de importação
// Combina polling REST com atualizações via WebSocket
import { useState, useEffect, useCallback } from 'react';
import { useSocket } from './useSocket';
export function useProgressoImportacao(jobId) {
const [estado, setEstado] = useState(null);
const [progresso, setProgresso] = useState(0);
const [resultado, setResultado] = useState(null);
const [erro, setErro] = useState(null);
const { ouvir } = useSocket();
// Polling de fallback — caso o WebSocket não esteja disponível
const verificarStatus = useCallback(async () => {
if (!jobId) return;
try {
const res = await fetch(`/importacao/status/${jobId}`, {
headers: { Authorization: `Bearer ${token}` },
});
const dados = await res.json();
setEstado(dados.estado);
setProgresso(dados.progresso || 0);
if (dados.estado === 'completed') setResultado(dados.resultado);
if (dados.estado === 'failed') setErro(dados.erro);
} catch { /* ignora erros de rede durante polling */ }
}, [jobId]);
useEffect(() => {
if (!jobId) return;
// Ouve eventos WebSocket de progresso — atualização em tempo real
const removerProgresso = ouvir('notificacao:nova', (dados) => {
const notif = dados.notificacao;
if (notif.jobId !== jobId) return;
if (notif.tipo === 'importacao:progresso') {
// Extrai porcentagem da mensagem "Importando... 45%"
const match = notif.mensagem.match(/(d+)%/);
if (match) setProgresso(Number(match[1]));
setEstado('active');
}
if (notif.tipo === 'importacao:concluida') {
setEstado('completed');
setProgresso(100);
setResultado(notif.resultados);
}
if (notif.tipo === 'importacao:falhou') {
setEstado('failed');
setErro(notif.mensagem);
}
});
// Polling a cada 3 segundos como fallback
verificarStatus();
const intervalo = setInterval(verificarStatus, 3000);
return () => {
removerProgresso();
clearInterval(intervalo);
};
}, [jobId, ouvir, verificarStatus]);
const concluido = estado === 'completed';
const falhou = estado === 'failed';
const processando = estado === 'active' || estado === 'waiting';
return { estado, progresso, resultado, erro, concluido, falhou, processando };
}// src/components/ImportacaoProdutos.jsx
// Componente completo de upload e acompanhamento de importação
import { useState } from 'react';
import { useProgressoImportacao } from '../hooks/useProgressoImportacao';
export function ImportacaoProdutos() {
const [jobId, setJobId] = useState(null);
const [enviando, setEnviando] = useState(false);
const { progresso, resultado, erro, concluido, falhou, processando } =
useProgressoImportacao(jobId);
async function handleUpload(e) {
const arquivo = e.target.files[0];
if (!arquivo) return;
setEnviando(true);
const formData = new FormData();
formData.append('arquivo', arquivo);
try {
const res = await fetch('/importacao/produtos', {
method: 'POST',
headers: { Authorization: `Bearer ${token}` },
body: formData,
// Não definir Content-Type — o browser define com o boundary correto para multipart
});
const dados = await res.json();
if (res.ok) {
// Salva o jobId para acompanhar o progresso
setJobId(dados.jobId);
} else {
console.error('Erro ao enfileirar importação:', dados.erro);
}
} finally {
setEnviando(false);
}
}
return (
<div className="importacao">
<h2>Importar Produtos</h2>
<p className="instrucao">
Faça upload de um arquivo CSV com colunas: codigo, nome, preco, estoque, categoria
</p>
{/* Área de upload — só mostra se não há importação em andamento */}
{!jobId && (
<label className="upload-area">
<input
type="file"
accept=".csv"
onChange={handleUpload}
disabled={enviando}
className="upload-input"
/>
<div className="upload-conteudo">
{enviando ? (
<p>Enviando arquivo...</p>
) : (
<>
<span className="upload-icone">📁</span>
<p>Clique ou arraste o CSV aqui</p>
<small>Máximo: 10MB</small>
</>
)}
</div>
</label>
)}
{/* Barra de progresso — exibida durante o processamento */}
{jobId && (
<div className="progresso-container">
<div className="progresso-header">
<p>
{processando && 'Importando produtos...'}
{concluido && '✅ Importação concluída!'}
{falhou && '❌ Importação falhou'}
</p>
<span>{progresso}%</span>
</div>
<div className="barra-progresso">
<div
className={`barra-preenchimento ${concluido ? 'completa' : ''}`}
style={{ width: `${progresso}%` }}
/>
</div>
{/* Resultado detalhado quando concluído */}
{concluido && resultado && (
<div className="resultado-importacao">
<p>✅ {resultado.criados} produtos importados</p>
{resultado.erros?.length > 0 && (
<details className="erros-detalhes">
<summary>⚠️ {resultado.erros.length} erros</summary>
<ul>
{resultado.erros.map((e, i) => (
<li key={i}>
Linha {e.linha} (código: {e.codigo}): {e.erro}
</li>
))}
</ul>
</details>
)}
<button onClick={() => setJobId(null)} className="btn-primario">
Nova importação
</button>
</div>
)}
{falhou && (
<div className="erro-importacao">
<p>{erro}</p>
<button onClick={() => setJobId(null)} className="btn-secundario">
Tentar novamente
</button>
</div>
)}
</div>
)}
</div>
);
}Tarefa para você
Adicione filas de tarefas na aplicação do Módulo 6:
# 1. Configure o Redis localmente para desenvolvimento
# docker run -d -p 6379:6379 redis:7-alpine
# 2. Crie a estrutura de filas:
# src/infrastructure/queue/redisConfig.js
# src/infrastructure/queue/filas.js (filaEmail, filaNotificacao)
# src/infrastructure/queue/produtores.js
# 3. Implemente o emailWorker.js com dois templates:
# - boasVindas: enviado quando um usuário se cadastra
# - tarefaCriada: enviado quando uma tarefa é criada
# Teste com Ethereal (https://ethereal.email) — captura emails sem enviar
# 4. Integre com o EventEmitter de domínio:
# - TAREFA_CRIADA → enfileirarEmail('tarefaCriada', ...)
# - Use jobId único para evitar emails duplicados
# 5. Crie a rota POST /importacao/produtos e o worker de importação
# Teste com um CSV de 100 produtos fictícios
# Observe o progresso no console do worker
# 6. Configure o Bull Board em /admin/filas
# Proteja com a variável de ambiente ADMIN_KEY
# Acesse o painel e observe os jobs sendo processados
# 7. Adicione o job recorrente de verificação de tarefas atrasadas
# Configure para rodar a cada 1 minuto em desenvolvimento:
# pattern: '* * * * *' (toda vez que o minuto muda)
# 8. Implemente o componente ImportacaoProdutos no front-end
# Acompanhe o progresso via polling ou WebSocketConclusão
Neste artigo você aprendeu:
- O problema que filas de tarefas resolvem — desacoplar aceitação de execução
- O padrão produtor-consumidor e por que ele torna sistemas mais resilientes
- BullMQ com Redis — a solução mais robusta para filas em Node.js
- Configuração de filas com retry automático, backoff exponencial e remoção de jobs antigos
- Workers com concorrência configurável, rate limiting e reporte de progresso
- Como integrar filas com o EventEmitter de domínio — emails disparados por eventos, não por controllers
- Upload de CSV e processamento em lotes com progresso reportado ao cliente
- Jobs recorrentes com cron para tarefas agendadas — relatórios, alertas, limpeza
- Bull Board para monitoramento visual das filas em tempo real
- Hook React para acompanhar o progresso de jobs longos via WebSocket e polling
- Componente de importação com barra de progresso, resultado e tratamento de erros
No próximo e último artigo desta série vamos construir o projeto final completo — revisando toda a jornada de 52 semanas e entregando uma aplicação de produção com tudo integrado.
📌 Próximo artigo: Aula 52 — Projeto Final: Revisão Completa da Série e Aplicação de Produção
📚 Fontes e Referências
- BullMQ — Documentação: https://docs.bullmq.io
- BullMQ — Padrões avançados: https://docs.bullmq.io/patterns
- Bull Board: https://github.com/felixmosh/bull-board
- Redis — Documentação: https://redis.io/docs
- ioredis: https://github.com/redis/ioredis
- Nodemailer: https://nodemailer.com/about
- Multer: https://github.com/expressjs/multer
- csv-parse: https://csv.js.org/parse
- Ethereal Email (para testes): https://ethereal.email
- Designing Distributed Systems — Brendan Burns (O'Reilly)
- Enterprise Integration Patterns — Hohpe e Woolf (Addison-Wesley)
- roadmap.sh — Backend: https://roadmap.sh/backend
Prof. Ricardo Matos
Mestre em Programação | Série: Dominando JavaScript em 1 Ano — Artigo 51/52
Quer o Artigo 52 — Projeto Final: Revisão Completa e Aplicação de Produção agora?
# you asked
Sim
# claude response
# 📘 Série: Dominando o JavaScript em 1 Ano