Node.js异步流：按2015年后日期过滤CSV时多文件输出内容重复问题

核心问题根因 #

• 作用域缺陷：循环内使用var list = []存储过滤后的数据，var不具备块级作用域，所有异步CSV解析回调共享同一个list引用。由于文件读取、CSV解析是异步非阻塞操作，循环会同步发起所有文件的读取流后立刻结束，等各个流的end事件触发时，list早已被后续循环的赋值覆盖为最后一个文件的解析结果，最终所有输出文件写入的都是同一份最后处理的文件数据。
• API误用：fs.createWriteStream()返回可写流实例，并非Promise对象，不存在.then()方法，原代码中绑定的写文件完成回调逻辑完全无法按预期执行，整个写入流程没有正确的时序控制。
• 缺少逻辑隔离：未将单文件的处理流程做独立的作用域封装，所有异步操作的临时变量互相串扰。

修复实现 #

1. 替换循环内的变量声明：将var改为let/const，让每次循环迭代拥有独立的块级作用域，每个文件的解析数据存入单独的数组，从根源上避免变量互相覆盖。
2. 抽离单文件处理逻辑为独立的异步函数，将「读取文件-解析CSV-过滤早于2015-01-01的行-写入结果」的完整流程封装，每个文件的处理状态完全独立。
3. 调整写流逻辑顺序：等当前文件所有数据解析、过滤完成后，再创建写流执行写入，删除无效的.then调用。

修复后参考代码 #

const fastcsv = require('fast-csv');
const csv = require('csv-parser');
const fs = require('fs');
const thresholdDate = new Date('2015-01-01');

// 封装单文件全流程处理逻辑，返回Promise方便时序控制
function processSingleFile(inputPath, outputPath) {
    return new Promise((resolve, reject) => {
        const filteredRows = [];
        fs.createReadStream(inputPath)
            .pipe(csv())
            .on('data', (row) => {
                const rowDate = new Date(row.Date);
                if (rowDate >= thresholdDate) {
                    filteredRows.push(row);
                }
            })
            .on('end', () => {
                // 解析完成后再初始化写流，避免时序错位
                const writeStream = fs.createWriteStream(outputPath);
                fastcsv.write(filteredRows, { headers: true })
                    .pipe(writeStream)
                    .on('finish', () => {
                        console.log(`文件处理完成：${inputPath}`);
                        resolve();
                    });
            })
            .on('error', reject);
    });
}

// 批量串行处理所有文件
async function batchProcess() {
    const files = fs.readdirSync("dir");
    for (const file of files) {
        const inputPath = `<inputPath>${file}`;
        const outputPath = `<outputPath>${file}`;
        await processSingleFile(inputPath, outputPath);
    }
    console.log("全部文件处理完成");
}

batchProcess().catch(err => console.error("处理异常：", err));

补充说明 #

• 无需额外使用setTimeout手动控制异步时序，上述方案通过独立的异步函数作用域隔离每个文件的状态，结合Promise和async/await保证流程顺序，即可从根源解决输出内容重复的问题。
• 如果需要提升批量处理速度，可以在batchProcess中调整并发数，批量调度多文件的并行处理，无需修改单文件的处理逻辑。

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