如何在Node.js中通过Passthrough与Pipe实现流排序？

Passthrough流本身仅做数据透传，要实现排序逻辑，必须先缓存所有流数据（排序依赖完整数据集），完成排序后再将数据写入Passthrough，最终管道到写入流。以下是两种常见场景的实现示例：

场景1：文本文件按行排序 #

适用于日志、纯文本等按行存储的文件，用readline处理行分割，避免流分块导致的不完整行问题：

const fs = require('fs');
const { PassThrough } = require('stream');
const { createInterface } = require('readline');

// 初始化流实例
const readStream = fs.createReadStream('./input.txt', 'utf8');
const writeStream = fs.createWriteStream('./output.txt');
const passThrough = new PassThrough({ encoding: 'utf8' });

const lines = [];

// 按行读取并缓存数据
const rl = createInterface({
  input: readStream,
  crlfDelay: Infinity // 兼容所有换行格式
});

rl.on('line', line => lines.push(line));

rl.on('close', () => {
  // 执行排序（可自定义排序逻辑，比如按数字、日期等）
  const sortedLines = lines.sort((a, b) => a.localeCompare(b));
  
  // 将排序后的内容写入Passthrough
  sortedLines.forEach(line => passThrough.write(`${line}\n`));
  passThrough.end();
});

// 管道传输
passThrough.pipe(writeStream);

writeStream.on('finish', () => console.log('排序写入完成'));

场景2：JSON数组文件排序 #

适用于存储数组的JSON文件，比如[5,2,7]或[{"name":"B"},{"name":"A"}]：

const fs = require('fs');
const { PassThrough } = require('stream');

const readStream = fs.createReadStream('./input.json', 'utf8');
const writeStream = fs.createWriteStream('./output.json');
const passThrough = new PassThrough({ encoding: 'utf8' });

let rawContent = '';

readStream.on('data', chunk => rawContent += chunk);

readStream.on('end', () => {
  try {
    const dataArray = JSON.parse(rawContent);
    // 自定义排序逻辑：示例为数字排序，对象可改为a.id - b.id等
    const sortedArray = dataArray.sort((a, b) => a - b);
    
    passThrough.write(JSON.stringify(sortedArray, null, 2));
    passThrough.end();
  } catch (err) {
    console.error('JSON解析失败:', err);
    passThrough.destroy(err);
  }
});

passThrough.pipe(writeStream);

writeStream.on('finish', () => console.log('JSON数组排序写入完成'));

注意事项 #

• 全量缓存数据会占用内存，处理超大文件时需考虑外部排序（将数据分块排序后合并），避免内存溢出
• Passthrough的作用是作为中间流载体，若无需额外流处理，也可直接将排序后的数据写入写入流，但保留Passthrough便于后续扩展其他流操作

内容的提问来源于stack exchange，提问作者ryokan