Node.js API内存及CPU占用过高，请求代码优化支持

问题背景 #

开发了一个Node.js API，用于从包含1000条记录的集合中读取数据，每条记录包含300个子负载数据点。执行时出现内存与CPU利用率过高的问题，POD资源配置如下：

• requests: 内存250Mi，CPU50m
• limits: 内存8000Mi，CPU4000m

原代码实现 #

主逻辑代码 #

const data = await csvModel['csvFiles'].find({ "created_datetime": today_datetime });
if (data.length > 0) {
    for (let mappData of data) {
        await payloadMapping(mappData.payload);
    }
}

payloadMapping 函数 #

const payloadMapping = async (jsonPayload) => {
  try {
      let categoryId = await getCategoryID(jsonPayload["category"][0].parent_category);
      let totalEvent = jsonPayload["category"][0].event.length;
      if (categoryId) {
          let allExistingEvent = await getAllEvents(categoryId);
          let eventInsertData = [];
          for (let i = 0; i < totalEvent; i++) {
              let duplicate_event = false;
              let eventPayload = {};
              eventPayload["categoryId"] = categoryId;
              eventPayload["start"] = jsonPayload["category"][0].event[i].start;
              /*Check no two event at same times */
              if (allExistingEvent.length > 0) {
                  allExistingEvent.forEach(item => {
                      let result = item.categoryId.every((element, index) => element === categoryId[index]);
                      if (result && item.start === eventPayload.start) {
                          duplicate_event = true;
                      }
                  });
              }
              /*Check ends */
              if (!duplicate_event) {
                  eventPayload["title"] = jsonPayload["category"][0].event[i].title;
                  eventPayload["description"] = jsonPayload["category"][0].event[i].description;
                  eventPayload["event_location"] = jsonPayload["category"][0].event[i].location;
                  eventPayload["duration"] = jsonPayload["category"][0].event[i].duration;
                  eventPayload["status"] = 1;
                  eventInsertData.push(eventPayload);
              }
          }
          if (eventInsertData.length > 0) {
            try {
                await Events.insertMany(eventInsertData);
            } catch (mongoErr) {
              console.log(mongoErr);
            }
          }
      } else {
        console.log("Category collection not exist" + jsonPayload["category"][0].name);
      }
    } catch (error) {
    console.log(error, "catch error")
  }
}

getCategoryID 函数 #

const getCategoryID = (parent_category) => {
  return new Promise( (resolve) => {
    (async () => {
      const query = { parent_category: parent_category};
      let categoryArray = [];
      let categoryResult = await categoryModel.find(query);
      if (categoryResult.length > 0) {
        categoryResult.forEach((item) => {
          categoryArray.push(item._id.toString());
        })
      }
      resolve(categoryArray);
    });
  });
};

getAllEvents 函数 #

const getAllEvents = (categoryId) => {
  return new Promise( (resolve) => {
    (async () => {
      try {
        const eventLists = await Events.find({ 'categoryId': categoryId }, { "categoryId": 1, "start": 1 });
        if (eventLists.length > 0) {
          resolve(eventLists);
        } else {
          resolve([]);
        }
      } catch (dbErr) {
        console.log("dbErr", dbErr);
        resolve([]);
      }
    });
  });
};

优化方案 #

1. 批量查询分类ID，避免循环单查 #

原代码每条payload都单独调用getCategoryID，产生1000次数据库查询。改为先收集所有需要的parent_category，一次性批量查询：

// 主逻辑修改
const data = await csvModel['csvFiles'].find({ "created_datetime": today_datetime });
if (data.length > 0) {
    // 收集所有parent_category
    const parentCategories = data.map(item => item.payload.category[0].parent_category);
    // 批量查询分类ID并构建映射表
    const categoryMap = await getCategoryIDBatch(parentCategories);
    // 并行处理payload（后续可控制并发数）
    await Promise.all(data.map(mappData => payloadMappingOptimized(mappData.payload, categoryMap)));
}

// 新增批量查询函数
const getCategoryIDBatch = async (parentCategories) => {
    const categoryResult = await categoryModel.find({ parent_category: { $in: parentCategories } });
    return categoryResult.reduce((map, item) => {
        map[item.parent_category] = item._id.toString();
        return map;
    }, {});
};

2. 数据库层面去重，替代内存遍历检查 #

原代码先拉取所有同类事件到内存再遍历判断重复，数据量大时CPU和内存占用极高。改用MongoDB的distinct查询已存在的时间，直接过滤重复事件：

const payloadMappingOptimized = async (jsonPayload, categoryMap) => {
  try {
      const parentCategory = jsonPayload["category"][0].parent_category;
      const categoryId = categoryMap[parentCategory];
      if (!categoryId) {
          console.log("Category collection not exist" + jsonPayload["category"][0].name);
          return;
      }
      
      const events = jsonPayload["category"][0].event;
      // 查询当前分类下已存在的start时间
      const existingStarts = await Events.distinct('start', { categoryId });
      // 过滤掉重复事件并组装插入数据
      const eventInsertData = events.filter(e => !existingStarts.includes(e.start)).map(e => ({
          categoryId,
          start: e.start,
          title: e.title,
          description: e.description,
          event_location: e.location,
          duration: e.duration,
          status: 1
      }));

      if (eventInsertData.length > 0) {
          await Events.insertMany(eventInsertData);
      }
  } catch (error) {
      console.log(error, "catch error");
  }
};

3. 优化Promise写法，消除冗余嵌套 #

原getCategoryID和getAllEvents存在不必要的Promise嵌套，直接改为async函数简化逻辑：

// 优化后的getCategoryID
const getCategoryID = async (parent_category) => {
    const categoryResult = await categoryModel.find({ parent_category });
    return categoryResult.map(item => item._id.toString());
};

// 优化后的getAllEvents
const getAllEvents = async (categoryId) => {
    try {
        return await Events.find({ categoryId }, { categoryId: 1, start: 1 });
    } catch (dbErr) {
        console.log("dbErr", dbErr);
        return [];
    }
};

4. 分页读取源数据，降低内存占用 #

原代码一次性将1000条记录加载到内存，每条包含300个数据点，内存压力大。改用分页分批读取：

const batchSize = 100; // 每次读取100条
let page = 0;
let hasMore = true;

while (hasMore) {
    const data = await csvModel['csvFiles']
        .find({ "created_datetime": today_datetime })
        .skip(page * batchSize)
        .limit(batchSize);
    
    if (data.length === 0) {
        hasMore = false;
        break;
    }
    
    // 处理当前批次数据
    const parentCategories = data.map(item => item.payload.category[0].parent_category);
    const categoryMap = await getCategoryIDBatch(parentCategories);
    await Promise.all(data.map(mappData => payloadMappingOptimized(mappData.payload, categoryMap)));
    
    page++;
}

5. 添加数据库索引，加速查询 #

为高频查询字段创建索引，减少数据库查询时间和CPU消耗：

// 在API启动时执行一次索引创建
await csvModel['csvFiles'].createIndex({ created_datetime: 1 });
await categoryModel.createIndex({ parent_category: 1 });
await Events.createIndex({ categoryId: 1, start: 1 });

6. 控制并发数，避免数据库连接过载 #

使用p-limit库控制并行处理的payload数量，防止同时发起过多数据库请求：

const pLimit = require('p-limit');
const limit = pLimit(10); // 同时处理10个payload

// 主逻辑中修改
await Promise.all(data.map(mappData => limit(() => payloadMappingOptimized(mappData.payload, categoryMap))));

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