Node.js使用stream pipeline批量处理文件夹文件仅最后文件生效问题求助

问题原因 #

• pipeline用法错误：你当前导入的pipeline是回调风格的API，不支持await，会导致循环不会等待上一个文件处理完成就直接进入下一次迭代，所有文件的流同时写入同一个transform实例。
• transform流复用问题：你全局定义的transform流是单例，所有文件的处理流都共用同一个转换实例，数据互相覆盖，最终所有输出都是最后一个文件的处理结果。
• 路径拼接不规范：直接用concat拼接路径在不同系统下可能出现分隔符错误，应该用Node内置的path模块处理路径。

修改方案 #

先修改导入逻辑，使用Promise版本的pipeline，同时每次循环都创建新的transform实例，规范路径拼接：

const fs = require('fs');
const path = require('path');
// 导入Promise版本的pipeline，旧版本Node可写为 const { pipeline } = require('stream').promises
const pipeline = require('stream/promises').pipeline;

// 将原来的transform定义改为工厂函数，每次调用返回全新的转换流实例
function createTransform() {
  // 此处替换为你自己的transform流创建逻辑即可，示例：
  // return new Transform({
  //   transform(chunk, encoding, callback) {
  //     // 你的转换处理逻辑
  //   }
  // })
}

async function processFolder(inPath, outPath) {
  const dir = await fs.promises.opendir(inPath)
  for await (const dirent of dir) {
    // 跳过子目录，仅处理文件
    if (!dirent.isFile()) continue

    // 用path模块拼接路径，兼容不同系统的分隔符
    const inFile = path.join(inPath, dirent.name);
    // 通用的移除原后缀逻辑，无需硬编码截取长度
    const fileNameWithoutExt = path.basename(dirent.name, path.extname(dirent.name))
    const outFile = path.join(outPath, `${fileNameWithoutExt}.json`);
  
    console.log('开始处理：', inFile);
    
    try {
      // 每次处理都传入独立的transform实例，避免数据污染
      await pipeline(
        fs.createReadStream(inFile, { encoding: 'utf8' }),
        createTransform(),
        fs.createWriteStream(outFile)
      );
      console.log('处理完成：', outFile);
    } catch (err) {
      console.error(`处理${inFile}失败`, err);
    }
  }
}

processFolder(inPath, outPath);

核心修改点 #

• 改用Promise版本的pipeline，保证await生效，循环按顺序逐个处理文件
• 将原来的单例transform改为工厂函数，每次处理文件生成独立的转换流实例，避免数据互相覆盖
• 增加文件类型判断，跳过子目录避免运行报错
• 用path模块处理路径和后缀，兼容性更强

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