Patrones de Readable y Writable

Los streams Readable y Writable comparten modos y primitivas de control de flujo. Elige el modo byte u objeto desde el principio y haz coincidir los tamaños de los fragmentos del productor con el rendimiento del consumidor.

Receta

import { Readable, Writable } from 'node:stream';
 
// Modo byte (por defecto)
const byteStream = Readable.from([Buffer.from('ab'), Buffer.from('cd')]);
 
// Modo objeto
const objectStream = Readable.from([{ id: 1 }, { id: 2 }], { objectMode: true });

Cuándo usarlo:

• Analizar objetos de línea CSV/NDJSON desde el disco
• Escribir objetos de métricas agregadas en un escritor por lotes
• Ajustar highWaterMark para la memoria vs la latencia
• Implementar un Readable personalizado desde una fuente de datos asíncrona

Ejemplo de trabajo

import { Readable, Writable } from 'node:stream';
 
class CounterSource extends Readable {
  private i = 0;
  constructor(private max: number, opts?: ConstructorParameters<typeof Readable>[0]) {
    super({ objectMode: true, ...opts });
  }
  _read(): void {
    if (this.i >= this.max) {
      this.push(null);
      return;
    }
    this.push({ n: this.i++ });
  }
}
 
const batchWriter = new Writable({
  objectMode: true,
  write(obj: { n: number }, _enc, cb) {
    // simular inserción por lotes en la base de datos
    setImmediate(cb);
  },
});
 
const source = new CounterSource(1000);
source.pipe(batchWriter);

// Modo pausado - read() explícito
const r = Readable.from(['a', 'b', 'c']);
r.on('readable', () => {
  let chunk;
  while ((chunk = r.read()) !== null) {
    console.log(chunk);
  }
});

Lo que esto demuestra:

• Readable personalizado implementa _read y empuja hasta que null finaliza el stream
• Writable en modo objeto recibe objetos tipados - procesa por lotes en _write o acumula para inserción masiva
• El modo pausado usa el evento readable + read() - útil para analizar el enmarcado
• El modo de flujo (evento data o pipe) es el predeterminado para la mayoría de las E/S

En detalle

Cómo funciona

• Modo de flujo - los datos se emiten automáticamente a través de eventos data o pipe.
• Modo pausado - el consumidor llama a read(n) para extraer fragmentos.
• highWaterMark - umbral del búfer interno; excederlo activa la contrapresión (false desde write).
• cork/uncork en Writable agrupa escrituras pequeñas en menos llamadas al sistema.

Matriz de modos

Notas de TypeScript

import { Readable } from 'node:stream';
 
function linesFromFile(path: string): Readable {
  return Readable.from(
    (async function* () {
      const { createReadStream } = await import('node:fs');
      const { createInterface } = await import('node:readline');
      const rl = createInterface({ input: createReadStream(path) });
      for await (const line of rl) yield line;
    })(),
  );
}

Errores comunes

• Mezclar streams de objetos y bytes en una misma tubería - confusión de tipos. Solución: Transform explícito para convertir.
• Empujar después de null - lanza un error. Solución: rastrear el estado finalizado en un Readable personalizado.
• Ignorar el false devuelto por write - desbordamiento de memoria. Solución: esperar el evento drain.
• highWaterMark pequeño en E/S de alta latencia - sobrecarga de llamadas al sistema. Solución: comparar 16-64 KB para archivos.
• Modo objeto con objetos enormes - cada objeto puede ser de MB. Solución: mantener el modo byte o referencias de stream.

Alternativas

Preguntas frecuentes

Relacionado

• Conceptos básicos de Streams - cuatro tipos de streams
• Contrapresión - detalles de pausa/drenaje
• Transform & Duplex - streams intermedios
• fs y fs/promises - createReadStream

Versiones de la pila: Esta página fue escrita para Node.js 24.18.0 (LTS activa), npm 10+, TypeScript 5.6+, Express 5, Fastify 5 y NestJS 11.