Streams no Node.js: Guia Prático

Processar arquivos, respostas HTTP, uploads, logs e grandes volumes de dados pode consumir muita memória quando a aplicação tenta carregar tudo de uma só vez. Os Streams no Node.js resolvem esse problema permitindo que os dados sejam lidos, transformados e gravados em pequenas partes, chamadas de chunks.

Em vez de esperar um arquivo inteiro chegar à memória, a aplicação começa a trabalhar assim que recebe o primeiro bloco. Essa abordagem reduz o consumo de recursos, diminui o tempo até o primeiro resultado e ajuda o sistema a lidar com cargas maiores. Neste guia, você aprenderá os tipos de stream, como usar pipe() e pipeline(), como funciona o backpressure, como criar transformações e quais cuidados adotar em produção.

O que são Streams no Node.js?

Uma stream representa um fluxo de dados que pode ser consumido progressivamente. Esse fluxo pode vir de um arquivo, uma conexão de rede, uma requisição HTTP, um processo externo ou qualquer fonte capaz de produzir dados ao longo do tempo.

O Node.js usa streams em várias APIs centrais. Objetos como process.stdin, process.stdout, requisições HTTP e métodos do módulo fs trabalham com esse modelo. Por isso, entender streams melhora não apenas o processamento de arquivos, mas também o desenvolvimento de servidores e APIs.

Antes de avançar, vale revisar o que é Node.js, o que é JavaScript e o que é uma API.

Por que não carregar tudo na memória?

Imagine um arquivo de vídeo com 4 GB. Uma implementação baseada em leitura completa precisaria reservar memória suficiente para armazenar o arquivo antes de iniciar a próxima etapa. Em ambientes com vários usuários, algumas operações simultâneas poderiam esgotar a memória do servidor.

Com streams, o Node.js lê um bloco, envia esse bloco para o próximo componente e continua o processo. A aplicação não precisa manter todo o conteúdo ao mesmo tempo. O consumo de memória tende a ficar mais previsível, mesmo quando o tamanho total dos dados aumenta.

Streams também melhoram a latência percebida. Em um download, o cliente começa a receber bytes imediatamente. Em uma importação, as primeiras linhas podem ser processadas enquanto o restante do arquivo ainda está sendo lido.

Os quatro tipos principais de stream

O módulo stream organiza os fluxos em quatro categorias:

• Readable: produz dados que podem ser lidos.
• Writable: recebe dados para gravação.
• Duplex: permite leitura e escrita de forma independente.
• Transform: recebe dados, modifica o conteúdo e produz uma nova saída.

Um arquivo aberto com createReadStream() é Readable. Um arquivo aberto com createWriteStream() é Writable. Uma conexão TCP é Duplex. Já uma stream de compressão é Transform, pois recebe dados sem compressão e entrega dados comprimidos.

A documentação oficial de streams do Node.js detalha as classes, eventos e opções disponíveis.

Lendo um arquivo com createReadStream

O módulo fs oferece uma maneira simples de criar uma stream de leitura:

const fs = require('node:fs');

const source = fs.createReadStream('./dados.csv', {
  encoding: 'utf8'
});

source.on('data', chunk => {
  console.log('Bloco recebido:', chunk.length);
});

source.on('end', () => {
  console.log('Leitura concluída');
});

source.on('error', error => {
  console.error('Falha na leitura:', error.message);
});

O evento data é emitido sempre que um novo bloco está disponível. O evento end indica que não haverá mais dados. O evento error precisa ser tratado, pois o arquivo pode não existir, as permissões podem ser insuficientes ou o dispositivo pode apresentar uma falha.

A opção encoding faz os chunks chegarem como texto. Sem ela, os dados normalmente são entregues como objetos Buffer.

Gravando dados com createWriteStream

Uma stream Writable recebe blocos por meio do método write() e é finalizada com end():

const fs = require('node:fs');

const destination = fs.createWriteStream('./resultado.txt');

destination.write('Primeira linha');
destination.write(String.fromCharCode(10));
destination.write('Segunda linha');
destination.end();

destination.on('finish', () => {
  console.log('Arquivo gravado');
});

destination.on('error', error => {
  console.error('Falha na gravação:', error.message);
});

O evento finish acontece depois que end() é chamado e todos os dados pendentes são enviados ao destino. Ele não deve ser confundido com o evento close, que indica o encerramento do recurso subjacente.

Consulte também a documentação oficial do módulo fs para conhecer opções como permissões, flags, intervalos de leitura e tamanhos de buffer.

Conectando streams com pipe

O método pipe() conecta uma Readable a uma Writable. O exemplo abaixo copia um arquivo sem carregá-lo completamente:

const fs = require('node:fs');

const source = fs.createReadStream('./entrada.zip');
const destination = fs.createWriteStream('./copia.zip');

source.pipe(destination);

O fluxo de dados passa automaticamente da origem para o destino. O método também ajuda a controlar a velocidade entre as duas pontas. Mesmo assim, a propagação de erros exige cuidado. Um erro na origem não encerra necessariamente todos os componentes de uma cadeia complexa.

Por que preferir pipeline?

A função pipeline() conecta streams e coordena o encerramento quando ocorre uma falha. Ela reduz o risco de arquivos abertos, sockets pendentes e streams que continuam executando depois de um erro.

const fs = require('node:fs');
const { pipeline } = require('node:stream/promises');
const { createGzip } = require('node:zlib');

async function compressFile() {
  await pipeline(
    fs.createReadStream('./relatorio.csv'),
    createGzip(),
    fs.createWriteStream('./relatorio.csv.gz')
  );

  console.log('Compressão concluída');
}

compressFile().catch(error => {
  console.error('Falha no pipeline:', error.message);
});

Nesse exemplo, o arquivo é lido, comprimido e gravado progressivamente. A versão baseada em Promises funciona bem com async e await. Quando uma etapa falha, o pipeline rejeita a Promise e tenta destruir as streams envolvidas.

Entendendo backpressure

Backpressure é o mecanismo usado quando o produtor envia dados mais rapidamente do que o consumidor consegue processar. Sem controle, os blocos se acumulam na memória até provocar lentidão ou falha.

O método write() de uma Writable retorna um valor booleano. Quando retorna false, o buffer interno atingiu o limite configurado e o produtor deve aguardar o evento drain antes de continuar.

async function writeMany(destination, items) {
  for (const item of items) {
    const canContinue = destination.write(item);

    if (!canContinue) {
      await new Promise(resolve => {
        destination.once('drain', resolve);
      });
    }
  }

  destination.end();
}

O uso de pipe() e pipeline() já administra esse comportamento em grande parte dos cenários. O controle manual é necessário quando a aplicação escreve diretamente em uma stream ou implementa um produtor personalizado.

O papel de highWaterMark

highWaterMark define um limite aproximado para o buffer interno. Ele não representa o tamanho total do fluxo nem um limite rígido de memória. É um sinal usado pelo sistema para decidir quando deve reduzir a leitura ou solicitar que o produtor aguarde.

const fs = require('node:fs');

const source = fs.createReadStream('./grande.log', {
  highWaterMark: 64 * 1024
});

Aumentar o valor pode melhorar a vazão em determinadas cargas, mas também aumenta o uso de memória por conexão. Reduzir demais o valor gera mais operações e pode diminuir o desempenho. A configuração deve ser testada com arquivos e concorrência próximos do ambiente real.

Criando uma Transform Stream

Uma Transform recebe chunks, aplica uma regra e envia novos chunks para a próxima etapa. O exemplo converte texto para letras maiúsculas:

const fs = require('node:fs');
const { Transform } = require('node:stream');
const { pipeline } = require('node:stream/promises');

const uppercase = new Transform({
  transform(chunk, encoding, callback) {
    const output = chunk.toString().toUpperCase();
    callback(null, output);
  }
});

async function run() {
  await pipeline(
    fs.createReadStream('./entrada.txt'),
    uppercase,
    fs.createWriteStream('./saida.txt')
  );
}

run().catch(console.error);

O callback recebe o erro como primeiro argumento e os dados transformados como segundo. Para uma falha de validação, chame callback(error). Nunca deixe o callback sem execução, pois o pipeline ficará aguardando indefinidamente.

Transformações assíncronas

Uma transformação pode depender de banco, API externa ou outra operação assíncrona. Nesse caso, só chame o callback quando o trabalho terminar:

const { Transform } = require('node:stream');

const enrich = new Transform({
  objectMode: true,
  async transform(item, encoding, callback) {
    try {
      const enriched = await findExtraData(item);
      callback(null, { ...item, ...enriched });
    } catch (error) {
      callback(error);
    }
  }
});

Esse código processa um item de cada vez. Para aumentar a concorrência, não basta chamar o callback antecipadamente, pois isso pode desorganizar a sequência e remover o backpressure. Quando a paralelização é necessária, use uma estratégia com limite explícito de tarefas simultâneas.

Streams em object mode

Por padrão, streams trabalham com bytes, strings e Buffers. Com objectMode: true, cada chunk pode ser um objeto JavaScript. Essa opção é útil em pipelines internos, como importação de registros, transformação de eventos e processamento de filas.

const { Readable } = require('node:stream');

const users = Readable.from([
  { id: 1, name: 'Ana' },
  { id: 2, name: 'Bruno' }
], { objectMode: true });

users.on('data', user => {
  console.log(user.name);
});

No modo de objetos, highWaterMark normalmente representa quantidade de itens, não bytes. Ainda assim, um item pode ser muito grande. Evite objetos com payloads enormes e monitore o consumo de memória.

Consumindo uma Readable com for await

Streams Readable podem ser consumidas como iteradores assíncronos. Essa sintaxe facilita a leitura sequencial e o uso de await dentro do loop:

const fs = require('node:fs');

async function countBytes() {
  const source = fs.createReadStream('./arquivo.bin');
  let total = 0;

  for await (const chunk of source) {
    total += chunk.length;
  }

  return total;
}

countBytes()
  .then(total => console.log('Bytes:', total))
  .catch(console.error);

O loop aguarda novos chunks e respeita o ritmo do consumidor. Ele é uma boa alternativa aos eventos data quando o processamento precisa ser escrito de maneira sequencial e legível.

Servindo downloads com streams

Em uma API, um arquivo pode ser enviado diretamente para a resposta HTTP:

const express = require('express');
const fs = require('node:fs');
const { pipeline } = require('node:stream/promises');

const app = express();

app.get('/downloads/:file', async (req, res, next) => {
  try {
    const safeFile = validateFileName(req.params.file);
    res.setHeader('Content-Type', 'application/octet-stream');

    await pipeline(
      fs.createReadStream(`./downloads/${safeFile}`),
      res
    );
  } catch (error) {
    next(error);
  }
});

O nome do arquivo deve ser validado para impedir path traversal. Também é importante tratar o caso em que parte da resposta já foi enviada antes do erro. O middleware de erros não deve tentar escrever um novo JSON quando os cabeçalhos já foram enviados.

Para uma estrutura mais ampla de rotas e middlewares, consulte o guia sobre como criar uma API com Node.js.

Uploads e processamento progressivo

Uploads também podem ser processados por streams. Em vez de manter todo o arquivo em memória, a aplicação encaminha os chunks para armazenamento, antivírus, compactação ou serviço de objetos.

Defina limites de tamanho, tempo e taxa. Uma stream eficiente não elimina riscos de abuso. Um cliente pode manter a conexão aberta por muito tempo, enviar dados lentamente ou iniciar várias transferências simultâneas.

Ao construir endpoints de alto volume, combine streams com as técnicas apresentadas no artigo sobre performance de APIs Node.js.

Cancelamento com AbortController

Operações demoradas precisam poder ser canceladas quando o usuário fecha a conexão, o tempo limite é atingido ou a aplicação inicia um desligamento controlado.

const fs = require('node:fs');
const { pipeline } = require('node:stream/promises');

const controller = new AbortController();

setTimeout(() => {
  controller.abort();
}, 10000);

pipeline(
  fs.createReadStream('./origem.bin'),
  fs.createWriteStream('./destino.bin'),
  { signal: controller.signal }
).catch(error => {
  console.error(error.name, error.message);
});

Ao abortar, remova arquivos parciais quando eles não forem úteis. Caso o destino suporte retomada, registre a posição de maneira segura para continuar depois.

Tratamento correto de erros

Erros podem acontecer na leitura, transformação, compressão, rede ou gravação. Em uma cadeia manual de pipe(), cada stream pode exigir um listener. Com pipeline(), a falha é centralizada e as etapas são encerradas de forma coordenada.

Evite ignorar Promises rejeitadas. Registre contexto suficiente para investigar o problema, como operação, arquivo, duração e quantidade processada. Não inclua tokens, conteúdo confidencial ou dados pessoais nos logs.

Uma operação pode falhar depois de escrever parte do destino. Use arquivos temporários e renomeação atômica quando o consumidor não deve enxergar resultados incompletos.

Streams e filas de tarefas

Streams controlam o fluxo de dados dentro de uma operação, enquanto filas coordenam tarefas entre processos e ao longo do tempo. Uma importação grande pode ser registrada em uma fila e, quando executada pelo worker, usar streams para ler o arquivo sem estourar a memória.

As duas técnicas são complementares. Para conhecer cache, comunicação e filas simples, veja Redis com Node.js.

Como testar pipelines

Não teste apenas o resultado ideal. Inclua cenários com origem inexistente, erro no meio da transformação, destino sem espaço, cancelamento e consumidor lento.

Para testes unitários, use Readable.from() como origem e uma Writable personalizada para capturar a saída. Um Transform deve ser verificado com vários chunks, pois dados textuais podem ser divididos em posições inesperadas.

const { Writable } = require('node:stream');

function createCollector() {
  const chunks = [];

  const writable = new Writable({
    write(chunk, encoding, callback) {
      chunks.push(Buffer.from(chunk));
      callback();
    }
  });

  writable.result = () => Buffer.concat(chunks);
  return writable;
}

Faça também testes de carga com concorrência realista. Uma stream que funciona com um arquivo pode consumir recursos demais quando centenas de conexões são abertas ao mesmo tempo.

Erros comuns

• Carregar o arquivo antes de criar a stream: isso elimina o principal benefício do processamento progressivo.
• Ignorar o retorno de write(): o produtor continua enviando dados mesmo com o buffer cheio.
• Usar pipe() sem estratégia de erros: componentes podem permanecer abertos depois de uma falha.
• Transformar texto sem considerar chunks: uma linha ou caractere pode ser dividido entre dois blocos.
• Aumentar highWaterMark sem medir: o consumo de memória cresce por stream ativa.
• Esquecer timeouts e cancelamento: conexões lentas mantêm recursos ocupados.
• Não limpar arquivos parciais: falhas deixam dados incompletos no armazenamento.
• Compartilhar uma stream entre operações independentes: eventos e encerramentos ficam difíceis de controlar.

Boas práticas para produção

• Prefira pipeline() em cadeias com múltiplas etapas.
• Use stream/promises para integrar com async e await.
• Respeite backpressure ao escrever manualmente.
• Defina limites de tamanho, tempo e concorrência.
• Valide caminhos e nomes de arquivos.
• Monitore bytes processados, duração, erros e streams abertas.
• Teste consumidores lentos e interrupções de rede.
• Remova destinos parciais ou grave primeiro em arquivos temporários.
• Ajuste highWaterMark somente com medições.
• Não registre o conteúdo completo dos dados em logs.

Conclusão

Streams no Node.js permitem processar grandes volumes de dados com memória previsível e menor latência. Readable, Writable, Duplex e Transform formam uma base reutilizada por arquivos, HTTP, compressão, processos e conexões de rede.

Comece usando createReadStream(), createWriteStream() e pipeline(). Depois, aprofunde-se em backpressure, object mode, iteração assíncrona e cancelamento. Com tratamento de erros, limites e testes adequados, streams tornam downloads, uploads, importações e transformações muito mais eficientes e seguros.