流的背压问题

流的背压问题

Node.js中的流（Stream）是处理大量数据的核心机制，但数据生产速度超过消费速度时，背压（Backpressure）问题就会出现。这个问题不解决可能导致内存暴涨或进程崩溃。

背压的产生原理

当可读流（Readable）推送数据的速度超过可写流（Writable）处理速度时，未处理的数据会在缓冲区堆积。Node.js的流默认有16KB高水位线（highWaterMark），超过这个限制时：

const fs = require('fs');

// 快速读取大文件
const readStream = fs.createReadStream('huge.mov');
const writeStream = fs.createWriteStream('copy.mov');

// 当写入速度跟不上读取速度时
readStream.on('data', (chunk) => {
  const canContinue = writeStream.write(chunk);
  if (!canContinue) {
    readStream.pause();  // 关键点：需要手动暂停
  }
});

writeStream.on('drain', () => {
  readStream.resume();  // 关键点：需要手动恢复
});

背压的自动处理

现代Node.js的pipe()方法已经内置背压处理：

// 最佳实践：使用自动背压控制的pipe
fs.createReadStream('input.mp4')
  .pipe(fs.createWriteStream('output.mp4'))
  .on('error', console.error);

但某些场景需要更精细控制：

let bytesWritten = 0;
const monitorStream = new PassThrough();

monitorStream.on('data', (chunk) => {
  bytesWritten += chunk.length;
  console.log(`已写入: ${bytesWritten} bytes`);
});

readStream
  .pipe(monitorStream)
  .pipe(writeStream);

自定义流的背压实现

创建自定义转换流时需要显式处理背压：

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

class ThrottleStream extends Transform {
  constructor(ms) {
    super();
    this.delay = ms;
  }

  _transform(chunk, encoding, callback) {
    this.push(chunk);
    // 主动延迟模拟处理耗时
    setTimeout(callback, this.delay);
  }
}

// 使用示例
const throttle = new ThrottleStream(100);
fastReadStream
  .pipe(throttle)
  .pipe(slowWriteStream);

高并发场景的特殊处理

HTTP服务器同时处理多个上传请求时：

const http = require('http');
const server = http.createServer((req, res) => {
  // 每个请求单独控制
  let uploaded = 0;
  req.on('data', (chunk) => {
    uploaded += chunk.length;
    if (uploaded > 100 * 1024 * 1024) {
      req.pause(); // 超过100MB暂停接收
      processAndClearBuffer(() => {
        req.resume();
      });
    }
  });
});

function processAndClearBuffer(done) {
  // 模拟异步处理
  setTimeout(done, 1000);
}

性能监控与调试

通过事件监听诊断背压问题：

const stream = fs.createReadStream('data.bin');

stream.on('data', (chunk) => {
  console.log('Buffer size:', stream._readableState.length);
});

stream.on('end', () => {
  console.log('Peak memory:', process.memoryUsage().rss);
});

// 或者使用第三方模块
const probe = require('stream-meter')();
stream.pipe(probe).pipe(writeStream);

不同流类型的处理差异

对象模式流

const objectStream = new Transform({
  objectMode: true,
  transform(obj, _, cb) {
    // 对象流的高水位线计数不同
    if (this._writableState.length > 100) {
      process.nextTick(cb);
    } else {
      this.push(processObject(obj));
      cb();
    }
  }
});

双工流

const { Duplex } = require('stream');
class EchoStream extends Duplex {
  _write(chunk, _, callback) {
    this.push(chunk);  // 写入时同时读取
    callback();
  }

  _read() {
    // 实现读取逻辑
  }
}

常见误区与解决方案

1. 错误示例：忽略drain事件

// 反模式：可能导致内存泄漏
readStream.on('data', (chunk) => {
  writeStream.write(chunk); // 没有检查返回值
});

2. 正确做法：

function copyStream(source, target) {
  return new Promise((resolve, reject) => {
    source.on('data', onData);
    target.on('drain', onDrain);
    target.on('finish', resolve);
    target.on('error', reject);

    let isPaused = false;
    function onData(chunk) {
      const canContinue = target.write(chunk);
      if (!canContinue && !isPaused) {
        isPaused = true;
        source.pause();
      }
    }

    function onDrain() {
      if (isPaused) {
        isPaused = false;
        source.resume();
      }
    }
  });
}

高级应用场景

动态速率控制

class DynamicThrottle extends Transform {
  constructor() {
    super({ highWaterMark: 1 }); // 严格控制缓冲区
    this.speed = 1;
  }

  _transform(chunk, _, cb) {
    this.push(chunk);
    // 根据系统负载动态调整
    const delay = 1000 / this.speed;
    setTimeout(cb, delay);
  }

  adjustSpeed(newSpeed) {
    this.speed = Math.max(1, newSpeed);
  }
}

// 使用示例
const throttle = new DynamicThrottle();
setInterval(() => {
  const load = process.cpuUsage().user;
  throttle.adjustSpeed(load > 50 ? 0.5 : 2);
}, 1000);

多级流水线控制

const pipeline = util.promisify(stream.pipeline);

async function complexProcessing(input) {
  try {
    await pipeline(
      input,
      createDecryptStream(),
      createDecompressStream(),
      createValidationStream(),
      output
    );
  } catch (err) {
    console.error('Pipeline failed:', err);
  }
}