内存分配失败 - JavaScript 堆内存不足

Question

我的 Node.js 代码存在内存泄漏问题。我正在尝试流式读取具有 100k 行的 CSV（链接中的示例文件）文件并处理文件中的每个条目。该过程在一段时间后出现内存分配错误。

“致命错误：接近堆限制的无效标记压缩分配失败 - JavaScript 堆内存不足”

sample csv:

我的代码示例

const fs = require('fs');
const config = require('../config/config');
const csv = require('csv-parser');
const tls = require('../services/tls');

processCSV('crawler', 'sample-csv.csv');

包含 100k 条目的流式处理 csv 文件

async function processCSV (jobName, fileName) {
  return new Promise((resolve, reject) => {
    let filePath = config.api.basePath + fileName;
    fs.createReadStream(filePath)
        .on('error', () => {
          // handle error
          console.log('error processing csv');
          reject();

        })
        .pipe(csv())
        .on('data', (row) => {
          createJob(jobName, row);
        })
        .on('end', () => {
          // handle end of CSV
          console.log('Finished processing csv');
          resolve(filePath);
        })
  });
}

验证 csv 文件中的每个 url

async function createJob (name, data) {
  let {hostname, port, ip} = data;
  let protocol = 'https';
  if (port === 80) {
    protocol = 'http';
  }
  let url = protocol + '://' + hostname;
  try {
    await tls.getHostData(url); // call an external api to get details of hostname
    return url;
  } catch (error) {
    return error;
  }
}

我不知道哪个函数导致了内存泄漏。

Answer 1

在我看来，您正在为 CSV 文件中的每一行调用 createJob()，并且您可能会导致这些作业中的每一个都同时在进程中和内存中。这可能会耗尽系统资源，尤其是在文件中有很多行的情况下。

解决这个问题的一个想法是调整代码，使只有 N 个createJob() 操作同时“进行中”。这是一种方法，当您同时达到最大请求数时暂停流，然后在有更多空间时恢复它：

async function processCSV (jobName, fileName) {
  return new Promise((resolve, reject) => {
    let filePath = config.api.basePath + fileName;
    let numConcurrent = 0;
    let paused = false;
    const maxConcurrent = 10;
    let stream = fs.createReadStream(filePath)
        .on('error', (err) => {
          // handle error
          console.log('error processing csv');
          reject(err);

        })
        .pipe(csv())
        .on('data', (row) => {

          function checkResume() {
              --numConcurrent;
              if (paused && numConcurrent < maxConcurrent) {
                  // restart the stream, there's room for more
                  paused = false;
                  stream.resume();
              }
          }
          ++numConcurrent;
          createJob(jobName, row).then(checkResume, checkResume);
          if (numConcurrent >= maxConcurrent) {
              // pause the stream because we have max number of operations going
              stream.pause();
              paused = true;
          }
        })
        .on('end', () => {
          // handle end of CSV
          console.log('Finished processing csv');
          resolve(filePath);
        })
  });
}


async function createJob (name, data) {
  let {hostname, port, ip} = data;
  let protocol = 'https';
  if (port === 80) {
    protocol = 'http';
  }
  let url = protocol + '://' + hostname;
  try {
    await tls.getHostData(url); // call an external api to get details of hostname
    return url;
  } catch (error) {
    // make sure returned promise is rejected
    throw error;
  }
}

注意：如果在处理给定行时出现错误，此实现（就像您在问题中显示的那样）会继续运行。这种行为可以根据需要改变。