Node.js Stream Deep Dive: From Backpressure to Pipeline Patterns in Production

后端开发

How an OOM Incident Changed Everything

Two years ago I inherited a log processing service. Its job was simple — read Nginx access logs, parse them, and write to ClickHouse. The legacy code:

const fs = require('fs');

async function processLogs(filePath) {
  const content = await fs.promises.readFile(filePath, 'utf-8');
  const lines = content.split('\n');

  for (const line of lines) {
    const parsed = parseLogLine(line);
    await insertToClickHouse(parsed);
  }
}

Daily 200MB log files ran fine. Then ops changed the log rotation cycle from daily to weekly, and a single file swelled to 1.8GB. Three seconds after launch, readFile loaded the entire file into memory — immediate OOM Kill.

This is the canonical use case for Node.js Streams. Streams aren't just a fancy "streaming" buzzword — they're your only option when data exceeds available memory.

Understanding Stream's Four Types

Node.js's Stream module builds on EventEmitter. The four core types each serve a distinct purpose:

Type Role Analogy Key Methods
Readable Produces data Water faucet read(), pipe()
Writable Consumes data Drain write(), end()
Transform Read, transform, write Water filter _transform()
Duplex Both readable and writable Walkie-talkie _read() + _write()

Readable Stream

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

const readable = Readable.from([
  'First line\n',
  'Second line\n',
  'Third line\n',
]);

readable.on('data', (chunk) => {
  console.log('Received:', chunk.toString());
});

readable.on('end', () => {
  console.log('Done reading');
});

The most common usage is fs.createReadStream:

const fs = require('fs');

// No matter how large the file, memory stays bounded — 64KB at a time
const stream = fs.createReadStream('1.8gb-logfile.log', {
  encoding: 'utf-8',
  highWaterMark: 64 * 1024,
});

Writable Stream

const fs = require('fs');

const writable = fs.createWriteStream('output.log', {
  flags: 'a',
  highWaterMark: 16 * 1024,
});

const canContinue = writable.write('A log line\n');
if (!canContinue) {
  writable.once('drain', () => {
    console.log('Buffer drained, ready for more writes');
  });
}

The return value of write() is the key to understanding backpressure — more on this shortly.

Transform Stream

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

const logParser = new Transform({
  readableObjectMode: true,    // Output JS objects, not Buffers
  writableObjectMode: false,   // Input is Buffer/string

  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n').filter(Boolean);

    for (const line of lines) {
      try {
        this.push(JSON.parse(line));
      } catch (err) {
        this.emit('parse-error', { line, error: err.message });
      }
    }

    callback();
  },
});

Duplex Stream

Readable and writable independently. TCP sockets and crypto streams are classic Duplex examples:

const { Duplex } = require('stream');

const echo = new Duplex({
  read(size) { /* read logic */ },
  write(chunk, encoding, callback) {
    console.log('Writing:', chunk.toString());
    this.push(chunk);   // Written data immediately becomes readable
    callback();
  },
});

Backpressure — The Core Stream Concept

Backpressure is the flow control mechanism when consumption can't keep up with production. Without it, data piles up in memory until OOM.

const fs = require('fs');
const zlib = require('zlib');

// ❌ No backpressure protection
const readStream = fs.createReadStream('huge-file.log');
const gzipStream = zlib.createGzip();
const writeStream = fs.createWriteStream('huge-file.log.gz');

readStream.on('data', (chunk) => {
  writeStream.write(chunk);  // Ignores the return value
});
// Problem: if writeStream's internal buffer fills (hits highWaterMark),
// it returns false, but we ignore it. Data keeps reading, memory grows.

Correct backpressure-aware code:

// ✅ Correct: pipe() handles backpressure automatically
readStream.pipe(gzipStream).pipe(writeStream);

// Or manual implementation
readStream.on('data', (chunk) => {
  const canWrite = writeStream.write(chunk);
  if (!canWrite) {
    readStream.pause();    // Pause reading
  }
});

writeStream.on('drain', () => {
  readStream.resume();    // Resume when drained
});

highWaterMark — The Backpressure Threshold

Scenario Default Recommended
File read stream 64 KB (65536) Default or up to 256KB
Network socket 16 KB (16384) Depends on bandwidth
Object mode 16 objects Depends on object size
Compression (Gzip) 16 KB Increase to 64KB to reduce context switches

Increasing highWaterMark improves throughput at the cost of memory. A rule of thumb:

function createOptimalReadStream(filePath) {
  const stat = fs.statSync(filePath);
  const size = stat.size / (1024 * 1024);  // MB

  let highWaterMark;
  if (size < 10) {
    highWaterMark = 64 * 1024;       // Small files: 64KB
  } else if (size < 500) {
    highWaterMark = 256 * 1024;      // Medium: 256KB
  } else {
    highWaterMark = 1024 * 1024;     // Large: 1MB
  }

  return fs.createReadStream(filePath, { highWaterMark });
}

pipe vs pipeline

pipe() has existed since Node.js's birth but has a fatal flaw — it doesn't propagate errors:

// ❌ pipe() error trap
fs.createReadStream('nonexistent.log')
  .pipe(transformStream)
  .pipe(writeStream);

// If any intermediate stream errors, the error won't propagate along the pipe chain
// Unhandled error → process crash

pipeline(), introduced in Node.js 10, fixes this:

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

async function processFile(inputPath, outputPath) {
  try {
    await pipeline(
      fs.createReadStream(inputPath),
      zlib.createGzip(),
      fs.createWriteStream(outputPath),
    );
    console.log('Compression complete');
  } catch (err) {
    console.error('Compression failed:', err.message);
    // pipeline automatically destroys all intermediate streams
  }
}

Three Benefits of pipeline

  1. Automatic error propagation — any stream error → entire pipeline errors
  2. Automatic destruction — cleans up all streams on error, no resource leaks
  3. Dual callback/Promise support — callbacks for legacy code, async/await for new

Custom Stream Implementations

Pattern 1: CSV Line-by-Line Transform

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

class CsvTransformer extends Transform {
  constructor(delimiter = ',') {
    super({ readableObjectMode: true });
    this.delimiter = delimiter;
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop();  // Last line may be incomplete

    for (const line of lines) {
      if (line.trim()) {
        this.push(line.split(this.delimiter));
      }
    }

    callback();
  }

  _flush(callback) {
    if (this.buffer.trim()) {
      this.push(this.buffer.split(this.delimiter));
    }
    callback();
  }
}

Pattern 2: Rate-Limited Writable

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

class RateLimitedWritable extends Writable {
  constructor(options, maxOpsPerSecond) {
    super(options);
    this.interval = 1000 / maxOpsPerSecond;
    this.lastWrite = 0;
  }

  _write(chunk, encoding, callback) {
    const now = Date.now();
    const elapsed = now - this.lastWrite;
    const delay = Math.max(0, this.interval - elapsed);

    setTimeout(() => {
      this.lastWrite = Date.now();
      console.log('Write:', chunk.toString().trim());
      callback();
    }, delay);
  }
}

Pattern 3: Batch Aggregation Transform

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

class BatchTransform extends Transform {
  constructor(batchSize = 100) {
    super({ readableObjectMode: true, writableObjectMode: true });
    this.batchSize = batchSize;
    this.buffer = [];
  }

  _transform(item, encoding, callback) {
    this.buffer.push(item);

    if (this.buffer.length >= this.batchSize) {
      this.push([...this.buffer]);
      this.buffer = [];
    }

    callback();
  }

  _flush(callback) {
    if (this.buffer.length > 0) {
      this.push([...this.buffer]);
    }
    callback();
  }
}

// Batch-insert 100 records at a time into the database
await pipeline(
  incomingDataStream,
  new BatchTransform(100),
  async function* (source) {
    for await (const batch of source) {
      await db.bulkInsert(batch);
    }
  },
);

Performance Comparison: Stream vs Full Load

Benchmark comparing three approaches on a 500MB file:

// Approach 1: Full load
async function fullLoad(path) {
  const data = await fs.promises.readFile(path, 'utf-8');
  const lines = data.split('\n');
  let count = 0;
  for (const line of lines) {
    if (line.includes('ERROR')) count++;
  }
  return count;
}

// Approach 2: readline
const readline = require('readline');
async function readlineMethod(path) {
  const rl = readline.createInterface({
    input: fs.createReadStream(path),
    crlfDelay: Infinity,
  });
  let count = 0;
  for await (const line of rl) {
    if (line.includes('ERROR')) count++;
  }
  return count;
}

// Approach 3: Transform Stream
async function streamMethod(path) {
  let count = 0;
  const counter = new (require('stream').Transform)({
    transform(chunk, encoding, callback) {
      const lines = chunk.toString().split('\n');
      for (const line of lines) {
        if (line.includes('ERROR')) count++;
      }
      callback();
    },
  });

  await pipeline(fs.createReadStream(path), counter);
  return count;
}

Results (500MB log file, MacBook Pro M1, Node.js 22):

Approach Time Peak Memory Notes
Full load 0.8s 580 MB File + string overhead
readline 1.5s 45 MB Line-by-line parsing overhead
Transform Stream 1.2s 42 MB Block-level, balanced perf/memory

Stream approach is 0.4s slower than full load but uses only 7% of the memory. For larger files, full load crashes; Stream keeps running.

Production Patterns

Pattern A: Large File Upload (Frontend + Backend)

// Node.js server: receive chunked upload
const http = require('http');
const { pipeline } = require('stream/promises');
const fs = require('fs');
const crypto = require('crypto');

http.createServer(async (req, res) => {
  if (req.method !== 'PUT') {
    res.writeHead(405);
    return res.end();
  }

  const fileId = crypto.randomUUID();
  const filePath = `./uploads/${fileId}`;

  try {
    await pipeline(
      req,                                    // Read directly from HTTP
      fs.createWriteStream(filePath),         // Write to disk as data arrives
    );
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ fileId }));
  } catch (err) {
    res.writeHead(500);
    res.end('Upload failed');
  }
}).listen(3000);

Pattern B: Real-Time Log Cleaning Pipeline

const { pipeline } = require('stream/promises');
const { createReadStream, createWriteStream } = require('fs');
const { Transform } = require('stream');
const { createGzip } = require('zlib');

// Complete ETL pipeline
async function etlPipeline() {
  // 1. Filter: keep only ERROR and WARN lines
  const filterStream = new Transform({
    transform(chunk, encoding, callback) {
      const lines = chunk.toString().split('\n')
        .filter(line => line.includes('ERROR') || line.includes('WARN'))
        .join('\n');

      if (lines) this.push(lines + '\n');
      callback();
    },
  });

  // 2. Mask: replace IP addresses
  const maskStream = new Transform({
    transform(chunk, encoding, callback) {
      const masked = chunk.toString()
        .replace(/\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}/g, '[MASKED_IP]');
      this.push(masked);
      callback();
    },
  });

  // 3. Assemble pipeline
  await pipeline(
    createReadStream('/var/log/nginx/access.log'),
    filterStream,      // Filter
    maskStream,        // Mask
    createGzip(),      // Compress
    createWriteStream('/tmp/filtered-log.gz'),
  );
}

Pattern C: Paginated API Data Export

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

function createOrderExportStream(userIds) {
  return new Readable({
    objectMode: true,

    async read() {
      if (userIds.length === 0) {
        this.push(null);  // End stream
        return;
      }

      const userId = userIds.shift();

      try {
        let page = 1;
        let hasMore = true;

        while (hasMore) {
          const response = await fetch(
            `/api/users/${userId}/orders?page=${page}&size=100`
          );
          const { data, total } = await response.json();

          for (const order of data) {
            if (!this.push(order)) {
              // Backpressure: save remaining work for next round
              userIds.unshift(userId);
              return;
            }
          }

          hasMore = (page * 100) < total;
          page++;
        }
      } catch (err) {
        this.destroy(err);
      }
    },
  });
}

Common Pitfalls

Pitfall 1: Forgetting error Event Handlers

// ❌ Unhandled error crashes the process
const stream = fs.createReadStream('maybe-not-exist.txt');
stream.pipe(process.stdout);

// ✅ Correct
stream.on('error', (err) => {
  console.error('Read failed:', err.message);
});

Pitfall 2: Forgetting callback() in Transform._flush

// ❌ Stream never ends
_flush(callback) {
  this.push('remaining data');
  // Forgot callback()!
}

// ✅ Correct
_flush(callback) {
  this.push('remaining data');
  callback();
}

Pitfall 3: Mixing Object Mode and Buffer Mode

// ❌ Input is objects but objectMode not declared
const t = new Transform({
  transform(chunk, encoding, callback) {
    // chunk is Buffer, not object!
    callback();
  },
});

// ✅ Correct
const t = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    // chunk is now a JS object
    callback();
  },
});

Try these browser-local tools — no sign-up required →

#Node.js#Stream#后端#性能优化#教程