Node.js Stream 流处理与背压

const { createReadStream, createWriteStream } = require('node:fs');
const { createGzip } = require('node:zlib');

createReadStream('./app.log')   // Readable：文件变成块流
  .pipe(createGzip())           // Transform：每块压一块
  .pipe(createWriteStream('./app.log.gz')); // Writable：落盘

const dst = createWriteStream('./out.txt');
const src = createReadStream('./in.txt', { highWaterMark: 64 * 1024 });

src.on('data', (chunk) => {
  const ok = dst.write(chunk);        // 返回 false = 下游缓冲已到 highWaterMark
  if (!ok) {
    src.pause();                      // 闸门：暂停上游读取
    dst.once('drain', () => src.resume()); // drain = 缓冲排空，恢复
  }
});
src.on('end', () => dst.end());

// pipe：错误不传播——中间流出错，下游不知道，进程可能挂起
src.pipe(transform).pipe(dst);
src.on('error', (e) => { /* 只管到 src */ });
transform.on('error', (e) => { /* 还要单独管 transform */ });
dst.on('error', (e) => { /* 以及 dst */ });

// pipeline：一条语句管全链的错误与清理
const { pipeline } = require('node:stream/promises');
await pipeline(src, transform, dst); // 任何一环出错：reject + 全链自动 destroy

// ============================================================
// 大文件逐行处理：正确处理背压
// ============================================================
const { createReadStream, createWriteStream } = require('node:fs');
const { createInterface } = require('node:readline');
const { Transform, pipeline } = require('node:stream');

// 错误示范：直接 readline + 处理函数，无背压
async function badPattern(inputPath, outputPath) {
  const rl = createInterface({
    input: createReadStream(inputPath),
    crlfDelay: Infinity
  });

  const output = createWriteStream(outputPath);
  for await (const line of rl) {
    // 若处理慢于读取速度，readline 内部缓冲会无限增长
    const processed = await expensiveProcess(line);
    output.write(processed + '\n'); // 不检查返回值
  }
  output.end();
}

// 正确示范：使用 pipeline + Transform
function goodPattern(inputPath, outputPath) {
  const transform = new Transform({
    // highWaterMark 控制内部缓冲上限
    highWaterMark: 1024 * 64,

    async transform(chunk, encoding, callback) {
      const lines = chunk.toString().split('\n');
      for (const line of lines) {
        if (!line) continue;
        try {
          const processed = await expensiveProcess(line);
          this.push(processed + '\n');
        } catch (err) {
          callback(err);
          return;
        }
      }
      callback();
    }
  });

  // pipeline 自动处理背压与错误传播
  pipeline(
    createReadStream(inputPath),
    transform,
    createWriteStream(outputPath),
    (err) => {
      if (err) console.error('管线失败:', err);
      else console.log('处理完成');
    }
  );
}

async function expensiveProcess(line) {
  // 模拟耗时处理
  return new Promise(resolve => {
    setTimeout(() => resolve(line.toUpperCase()), 1);
  });
}

goodPattern('./input.txt', './output.txt');

const { createReadStream } = require('node:fs');
const { createInterface } = require('node:readline');

async function topIps(path) {
  const counts = new Map(); // 键控集合的用武之地，见 javascript/075
  const rl = createInterface({
    input: createReadStream(path, { encoding: 'utf8' }),
    crlfDelay: Infinity,   // 兼容 Windows 的 \r\n 换行
  });
  for await (const line of rl) {          // readline 实现了异步迭代器
    const ip = line.split(' ')[0];
    counts.set(ip, (counts.get(ip) ?? 0) + 1);
  }
  return [...counts.entries()]
    .sort((a, b) => b[1] - a[1])
    .slice(0, 10);
}

// 反模式：未处理 'error' 事件，导致进程崩溃
const stream = fs.createReadStream('large.txt');
stream.pipe(transformStream).pipe(fs.createWriteStream('out.txt'));
// 任一流抛出 error 事件，整个进程崩溃

// 正确：使用 pipeline 自动处理错误
const { pipeline } = require('node:stream/promises');

async function copy() {
  try {
    await pipeline(
      fs.createReadStream('large.txt'),
      transformStream,
      fs.createWriteStream('out.txt')
    );
    console.log('完成');
  } catch (err) {
    console.error('管线失败:', err);
  }
}

// 反模式：无背压的 Kafka 生产
const producer = kafka.producer();

app.post('/log', (req, res) => {
  producer.send({
    topic: 'logs',
    messages: [{ value: JSON.stringify(req.body) }]
  });
  res.json({ ok: true });
});

// 改造：基于 Stream 的背压
const { Transform, pipeline } = require('node:stream');

class KafkaSink extends Transform {
  constructor(producer, options = {}) {
    super({ ...options, highWaterMark: 1000 }); // 1000 条缓冲
    this.producer = producer;
  }

  async _transform(chunk, encoding, callback) {
    try {
      await this.producer.send({
        topic: 'logs',
        messages: [{ value: chunk }]
      });
      callback();
    } catch (err) {
      callback(err);
    }
  }
}

const sink = new KafkaSink(producer);

app.post('/log', (req, res) => {
  const canWrite = sink.write(JSON.stringify(req.body) + '\n');
  if (!canWrite) {
    // 背压：返回 429，让客户端重试
    return res.status(429).json({ error: '系统繁忙，请稍后重试' });
  }
  res.json({ ok: true });
});

sink.on('error', (err) => {
  console.error('Kafka Sink 错误:', err);
  // 触发熔断
});

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

// Promise 风格的 pipeline
await pipeline(
  fs.createReadStream('input.txt'),
  gzip,
  fs.createWriteStream('output.txt.gz')
);