NodeJS大文本文件高效处理优化求助

代码核心问题分析 #

1. 异步回调脱节：lineReader.eachLine不支持async回调的Promise等待，读取流程会持续推进，完全不等待聚合任务完成，导致并发失控，可能引发CPU/内存过载。
2. 无意义异步包装：aggregate内部是纯同步逻辑，却用Promise包裹，增加不必要的异步调度开销。
3. 内存累积风险：所有聚合结果存入result数组，3亿行的聚合数据会占用大量内存，最终一次性写入也会拖慢速度。
4. 无并发控制：每到chunk边界就启动聚合任务，若处理速度跟不上读取速度，会同时运行大量任务，引发资源竞争。
5. 工具库性能瓶颈：line-reader是上层封装库，性能不如Node.js原生流方案。

优化步骤与代码实现 #

1. 替换为原生readline流处理 #

原生readline基于Node.js流实现，性能更优，且能更好控制读取节奏。

2. 同步聚合+并发控制 #

去掉aggregate的Promise包装，改为同步函数；通过手动实现的并发控制器限制同时运行的聚合任务数（建议值为CPU核心数-1），避免资源过载。

3. 增量合并聚合结果 #

维护全局聚合Map，每个chunk处理完成后直接合并结果，无需存储所有chunk的聚合数据，大幅降低内存占用。

4. 逐行写入输出 #

避免一次性写入大文件，改为逐行写入，减少内存压力。

优化后代码 #

const fs = require('fs');
const readline = require('readline');
const { promisify } = require('util');
const writeFile = promisify(fs.writeFile);
const appendFile = promisify(fs.appendFile);

// 并发控制：根据CPU核心数调整，示例为2
const MAX_CONCURRENT_TASKS = 2;
let activeTasks = 0;
let taskQueue = [];

async function parseFileFast(fileName_) {
    const start = Date.now();
    // 全局聚合结果容器
    const totalAggregate = new Map();
    const chunkSize = 3000000;
    let chunkData = [];
    let lineCounter = 0;

    console.log(`processing .. ${fileName_}`);

    // 创建原生readline流接口
    const rl = readline.createInterface({
        input: fs.createReadStream(fileName_),
        crlfDelay: Infinity // 兼容所有换行符格式
    });

    // 处理每行数据
    rl.on('line', (line) => {
        chunkData.push(line);
        lineCounter++;

        if (chunkData.length >= chunkSize) {
            // 提交聚合任务（拷贝数组避免后续修改影响）
            submitAggregateTask([...chunkData]);
            chunkData = [];
            console.log(`processed lines so far .. ${lineCounter}`);
        }
    });

    // 流结束处理剩余数据
    rl.on('close', async () => {
        if (chunkData.length > 0) {
            await submitAggregateTask(chunkData);
        }
        // 等待所有聚合任务完成
        while (activeTasks > 0) {
            await new Promise(resolve => setTimeout(resolve, 100));
        }
        // 写入输出文件
        await writeOutput(totalAggregate, fileName_.replace('.csv', '_3.csv'));
        const end = Date.now();
        console.log(`finished processing in sec: ${(end - start) / 1000}`);
    });

    // 提交任务并控制并发
    function submitAggregateTask(data) {
        return new Promise(resolve => {
            taskQueue.push({ data, resolve });
            runNextTask();
        });
    }

    // 执行下一个队列任务
    function runNextTask() {
        if (activeTasks >= MAX_CONCURRENT_TASKS || taskQueue.length === 0) return;
        
        const { data, resolve } = taskQueue.shift();
        activeTasks++;
        
        // 同步聚合处理
        const chunkAggregate = aggregate(data);
        // 合并到全局结果
        mergeAggregate(totalAggregate, chunkAggregate);
        
        activeTasks--;
        resolve();
        runNextTask();
    }
}

// 同步聚合函数
function aggregate(dataSet) {
    const aggregateMap = new Map();
    dataSet.forEach(line => {
        // 替换为你的实际processFrames逻辑
        processFrames(line, aggregateMap);
    });
    return aggregateMap;
}

// 合并聚合结果（根据你的求和逻辑调整）
function mergeAggregate(total, chunk) {
    for (const [key, value] of chunk.entries()) {
        if (total.has(key)) {
            // 示例：数值累加，根据实际需求修改
            total.set(key, total.get(key) + value);
        } else {
            total.set(key, value);
        }
    }
}

// 逐行写入输出，避免内存过载
async function writeOutput(aggregateMap, outputPath) {
    // 清空目标文件（若存在）
    await writeFile(outputPath, '', 'utf8');
    for (const [key, value] of aggregateMap.entries()) {
        // 替换为你的实际输出格式
        const line = `${key},${value}\n`;
        await appendFile(outputPath, line, 'utf8');
    }
}

// 示例processFrames函数（替换为你的实际逻辑）
function processFrames(line, aggregateMap) {
    const [key, numStr] = line.split(',');
    const num = parseFloat(numStr);
    if (!isNaN(num)) {
        aggregateMap.set(key, (aggregateMap.get(key) || 0) + num);
    }
    return aggregateMap;
}

额外优化建议 #

• 调整chunkSize：若内存占用过高，减小chunkSize；若CPU利用率不足，可适当增大（需控制在内存承受范围内）。
• CPU密集型任务用Worker线程：如果processFrames是纯CPU密集型，可将聚合任务拆分到Worker线程。注意不要传递大数组，而是让Worker直接读取文件的字节片段，最后合并各Worker的结果，避免线程间大数据传输开销。
• 启用更大堆内存：启动Node.js时添加--max-old-space-size=8192（单位MB，根据机器内存调整），缓解内存压力。
• 使用更快的流工具：如split2库，处理大文件换行分割的性能优于原生readline。

内容的提问来源于stack exchange，提问作者k.honsali