大文件处理策略

大文件处理的核心挑战

Node.js处理大文件时面临内存限制和性能瓶颈。单次读取整个文件到内存会导致内存溢出，同步阻塞操作影响应用响应速度。流式处理成为解决这些问题的关键，通过分块读取和写入减少内存占用。

流式处理基础

Node.js内置的fs模块提供createReadStream和createWriteStream方法：

const fs = require('fs');

// 创建读取流
const readStream = fs.createReadStream('largefile.txt', {
  highWaterMark: 16 * 1024 // 每次读取16KB
});

// 创建写入流
const writeStream = fs.createWriteStream('output.txt');

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

highWaterMark参数控制缓冲区大小，需要根据实际内存情况调整。典型值在16KB到64KB之间平衡内存使用和I/O效率。

分块处理技术

对于需要逐行处理的大文件，使用readline模块更高效：

const readline = require('readline');
const fs = require('fs');

const rl = readline.createInterface({
  input: fs.createReadStream('large_log_file.log'),
  crlfDelay: Infinity
});

let lineCount = 0;
rl.on('line', (line) => {
  lineCount++;
  // 处理每行数据
  if(line.includes('ERROR')) {
    console.log(`发现错误日志: ${line}`);
  }
});

rl.on('close', () => {
  console.log(`共处理 ${lineCount} 行日志`);
});

内存优化策略

处理超大JSON文件时，采用流式JSON解析器如JSONStream：

const fs = require('fs');
const JSONStream = require('JSONStream');

const stream = fs.createReadStream('huge_data.json', {
  encoding: 'utf8'
});

const parser = JSONStream.parse('items.*');
stream.pipe(parser);

parser.on('data', (item) => {
  // 逐个处理数组元素
  processItem(item);
});

function processItem(item) {
  // 模拟处理逻辑
  console.log(`处理项目: ${item.id}`);
}

并行处理方案

利用Worker Threads实现CPU密集型任务的并行处理：

const { Worker, isMainThread, parentPort } = require('worker_threads');
const fs = require('fs');

if (isMainThread) {
  // 主线程
  const worker = new Worker(__filename);
  
  fs.createReadStream('large_data.csv')
    .on('data', (chunk) => {
      worker.postMessage(chunk.toString());
    });
} else {
  // 工作线程
  parentPort.on('message', (chunk) => {
    const processed = processChunk(chunk);
    parentPort.postMessage(processed);
  });
  
  function processChunk(data) {
    // 模拟处理逻辑
    return data.split(',').map(Number).filter(n => n > 100);
  }
}

断点续传实现

文件上传场景下的断点续传需要记录处理进度：

const fs = require('fs');
const path = require('path');

class FileProcessor {
  constructor(filePath) {
    this.filePath = filePath;
    this.progressFile = `${filePath}.progress`;
    this.processedBytes = 0;
    this.loadProgress();
  }

  loadProgress() {
    try {
      this.processedBytes = parseInt(
        fs.readFileSync(this.progressFile, 'utf8')
      ) || 0;
    } catch (err) {
      if(err.code !== 'ENOENT') throw err;
    }
  }

  async process() {
    const stat = fs.statSync(this.filePath);
    const readStream = fs.createReadStream(this.filePath, {
      start: this.processedBytes
    });

    readStream.on('data', (chunk) => {
      this.processedBytes += chunk.length;
      fs.writeFileSync(this.progressFile, this.processedBytes.toString());
      // 处理数据块...
    });

    return new Promise((resolve) => {
      readStream.on('end', () => {
        fs.unlinkSync(this.progressFile);
        resolve();
      });
    });
  }
}

实际应用场景

视频转码服务中的分片处理示例：

const ffmpeg = require('fluent-ffmpeg');
const fs = require('fs');

function transcodeLargeVideo(inputPath, outputPath) {
  return new Promise((resolve, reject) => {
    const command = ffmpeg(inputPath)
      .outputOptions([
        '-c:v libx264',
        '-crf 23',
        '-preset fast',
        '-movflags frag_keyframe+empty_moov'
      ])
      .on('progress', (progress) => {
        console.log(`处理进度: ${Math.floor(progress.percent)}%`);
      })
      .on('end', resolve)
      .on('error', reject)
      .save(outputPath);
  });
}

// 使用分片处理
async function processInChunks() {
  const chunkSize = 1024 * 1024 * 50; // 50MB
  const fileSize = fs.statSync('input.mp4').size;
  
  for(let i = 0; i < fileSize; i += chunkSize) {
    await transcodeChunk(i, Math.min(i + chunkSize, fileSize));
  }
}

性能监控与调优

添加处理性能监控指标：

const fs = require('fs');
const { performance } = require('perf_hooks');

async function processWithMetrics() {
  const start = performance.now();
  let bytesProcessed = 0;
  
  const stream = fs.createReadStream('massive_data.bin');
  
  stream.on('data', (chunk) => {
    bytesProcessed += chunk.length;
    const elapsed = (performance.now() - start) / 1000;
    const speed = (bytesProcessed / 1024 / 1024 / elapsed).toFixed(2);
    
    console.log(`处理速度: ${speed} MB/s`);
    // 实际处理逻辑...
  });

  return new Promise((resolve) => {
    stream.on('end', resolve);
  });
}

错误处理机制

健壮的错误处理策略实现：

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

class SafeProcessor extends Transform {
  _transform(chunk, encoding, callback) {
    try {
      // 模拟可能失败的处理
      if(Math.random() < 0.0001) {
        throw new Error('随机处理错误');
      }
      
      const processed = chunk.toString().toUpperCase();
      this.push(processed);
      callback();
    } catch (err) {
      // 记录错误但继续处理
      fs.appendFileSync('errors.log', `${new Date().toISOString()} - ${err.message}\n`);
      callback(null, chunk); // 跳过错误继续
    }
  }
}

fs.createReadStream('input.txt')
  .pipe(new SafeProcessor())
  .pipe(fs.createWriteStream('output.txt'));

内存泄漏预防

处理大文件时的内存泄漏检测：

const fs = require('fs');
const heapdump = require('heapdump');

function monitorMemoryUsage() {
  setInterval(() => {
    const usage = process.memoryUsage();
    console.log(`内存使用: ${Math.round(usage.heapUsed / 1024 / 1024)}MB`);
    
    if(usage.heapUsed > 500 * 1024 * 1024) { // 超过500MB
      heapdump.writeSnapshot(`heapdump-${Date.now()}.heapsnapshot`);
    }
  }, 5000);
}

monitorMemoryUsage();

// 大文件处理逻辑
fs.createReadStream('huge_file.data')
  .on('data', processDataChunk);