Redis Streams no Node.js: filas com consumer groups

Atualizado em: 27 de outubro de 2026

Rack de servidores processando fluxos de dados no Node.js

Redis Streams no Node.js oferece um log persistente de mensagens com IDs ordenados, grupos de consumidores, confirmação e recuperação de itens pendentes. Diferentemente de Pub/Sub, as mensagens permanecem no Redis até serem removidas pela política de retenção.

Streams são úteis para filas de trabalho, integração entre serviços, processamento assíncrono e eventos que precisam sobreviver a desconexões. Ainda assim, é necessário projetar idempotência, retenção, retries e observabilidade.

Quando usar Redis Streams

Use Streams quando precisar de histórico curto ou médio, consumidores independentes, distribuição de trabalho e possibilidade de reprocessamento. Para volumes enormes, retenção longa ou ecossistema de streaming analítico, Kafka pode ser mais adequado.

Instalação e conexão

npm install redis

import { createClient } from 'redis';

const redis = createClient({ url: process.env.REDIS_URL });
redis.on('error', error => console.error('Redis error', error));
await redis.connect();

Use TLS e ACLs em produção. Consulte Redis com Node.js: Cache e Filas.

Publicando com XADD

const id = await redis.xAdd(
  'orders-stream',
  '*',
  {
    type: 'orders.created.v1',
    eventId: crypto.randomUUID(),
    payload: JSON.stringify({ orderId: 'o-123' })
  }
);

console.log(id);

O asterisco pede ao Redis para gerar um ID baseado em tempo e sequência. O ID ordena mensagens no stream, mas o eventId continua útil para idempotência de negócio.

Lendo sem grupo

const entries = await redis.xRead(
  [{ key: 'orders-stream', id: '0-0' }],
  { COUNT: 100 }
);

Esse modo é adequado para inspeção ou consumidores que precisam de uma cópia completa. Para dividir trabalho entre instâncias, use consumer groups.

Criando um grupo

try {
  await redis.xGroupCreate(
    'orders-stream',
    'billing-workers',
    '0',
    { MKSTREAM: true }
  );
} catch (error) {
  if (!String(error.message).includes('BUSYGROUP')) throw error;
}

O grupo mantém posição própria e uma lista de mensagens entregues, mas ainda não confirmadas.

Consumindo novas mensagens

const response = await redis.xReadGroup(
  'billing-workers',
  workerId,
  [{ key: 'orders-stream', id: '>' }],
  { COUNT: 10, BLOCK: 5000 }
);

O identificador > solicita mensagens que ainda não foram entregues ao grupo. O bloqueio reduz polling agressivo.

Processamento e ACK

for (const stream of response ?? []) {
  for (const message of stream.messages) {
    try {
      await processEvent(message.message);
      await redis.xAck('orders-stream', 'billing-workers', message.id);
    } catch (error) {
      logger.error({ error, messageId: message.id }, 'processing failed');
    }
  }
}

Confirme somente depois do efeito durável. Se o processo cair antes do ACK, a mensagem continuará pendente.

Idempotência

A entrega é normalmente at-least-once. O consumidor pode receber novamente o mesmo evento. Registre o eventId em uma tabela com restrição única dentro da mesma transação do efeito:

BEGIN;
INSERT INTO processed_events (event_id)
VALUES ($1)
ON CONFLICT DO NOTHING;
-- aplique o efeito somente se houve inserção
COMMIT;

Veja Consumer Idempotente no Node.js.

Mensagens pendentes

Use XPENDING para observar itens entregues e não confirmados. Uma quantidade crescente indica consumidores travados, lentos ou falhando.

const pending = await redis.xPending(
  'orders-stream',
  'billing-workers'
);

Recuperando mensagens abandonadas

XAUTOCLAIM transfere mensagens ociosas para outro consumidor:

const claimed = await redis.xAutoClaim(
  'orders-stream',
  'billing-workers',
  workerId,
  60_000,
  '0-0',
  { COUNT: 20 }
);

Defina o tempo mínimo acima da duração normal de processamento. Caso contrário, dois consumidores podem trabalhar simultaneamente no mesmo item.

Retry e Dead Letter Stream

Streams não impõem uma política de retry. Registre tentativas em armazenamento durável ou no próprio payload. Depois do limite, envie para outro stream:

await redis.xAdd('orders-dead-letter', '*', {
  originalId: message.id,
  reason: safeErrorCode,
  payload: JSON.stringify(message.message)
});
await redis.xAck(stream, group, message.id);

Não inclua stack traces ou segredos. Crie uma ferramenta controlada de replay.

Retenção com MAXLEN

await redis.xAdd(
  'orders-stream',
  '*',
  fields,
  { TRIM: { strategy: 'MAXLEN', strategyModifier: '~', threshold: 100000 } }
);

O modificador aproximado reduz custo. Defina retenção considerando consumidores lentos; remover mensagens ainda necessárias inviabiliza recuperação.

Particionamento

Um único stream é ordenado, mas pode concentrar carga. Para paralelismo maior, distribua por chave estável:

orders-stream-0
orders-stream-1
orders-stream-2

A mesma entidade deve ir para a mesma partição quando a ordem por entidade importa.

Payloads pequenos

Armazene eventos compactos e versionados. Para arquivos, grave no object storage e publique apenas o identificador. Payloads grandes pressionam memória, replicação e persistência do Redis.

Contratos versionados

{
  "type": "orders.created.v1",
  "eventId": "evt-123",
  "occurredAt": "2026-07-26T12:00:00Z",
  "data": { "orderId": "o-123" }
}

Consumidores devem rejeitar versões desconhecidas de forma observável ou encaminhá-las para análise.

Backpressure

Controle o COUNT e a concorrência. Ler centenas de mensagens e iniciar todas em paralelo pode esgotar banco, APIs e memória. Use um pool limitado:

const concurrency = 10;
await mapWithConcurrency(messages, concurrency, handleMessage);

Bloqueio e shutdown

O BLOCK deve ter timeout finito para que o loop perceba sinais de encerramento. No SIGTERM, pare novas leituras, aguarde tarefas ativas e feche o cliente:

let stopping = false;
process.once('SIGTERM', () => { stopping = true; });

while (!stopping) {
  await pollOnce();
}

await redis.quit();

Consulte Graceful Worker Shutdown no Node.js.

Redis Cluster

Comandos que envolvem múltiplas chaves precisam respeitar hash slots. Use hash tags quando stream e estruturas auxiliares precisarem ficar no mesmo slot:

{orders}:stream
{orders}:dedup

Teste o comportamento no modo de implantação real.

Persistência e disponibilidade

Streams dependem das garantias configuradas para Redis: AOF, snapshots, replicação e failover. Não confunda persistência habilitada com perda zero. Entenda janelas de replicação e durabilidade antes de usar em processos financeiros.

Observabilidade

Monitore:

tamanho do stream;
lag por grupo;
mensagens pendentes;
idade da pendência mais antiga;
taxa de ACK;
retries e dead letters;
tempo de processamento;
erros por versão de evento;
memória e latência do Redis.

Testes

criação idempotente do grupo;
publicação e leitura;
dois consumidores dividindo trabalho;
queda antes do ACK;
claim depois do timeout;
evento duplicado;
DLQ;
retenção;
shutdown com mensagens ativas.

Streams versus Pub/Sub

Pub/Sub envia apenas para inscritos conectados e não mantém backlog. Streams armazenam mensagens, controlam posição e suportam grupos. Leia também Redis Pub/Sub no Node.js.

Erros comuns

confirmar antes do efeito;
não implementar idempotência;
nunca recuperar pendências;
claim cedo demais;
não definir retenção;
usar concorrência ilimitada;
guardar payloads enormes;
ignorar durabilidade do Redis;
não versionar contratos.

Conclusão

Redis Streams no Node.js transforma o Redis em um log de mensagens com grupos, confirmação e recuperação. Ele atende bem workloads assíncronos que precisam de mais garantias que Pub/Sub, sem exigir imediatamente um broker separado.

Projete consumidores idempotentes, retenção, recuperação de pendências e shutdown gracioso. As garantias reais dependem tanto do código quanto da configuração de persistência e failover do Redis.

Consulte a documentação oficial de Redis Streams e os comandos de Streams do Redis.