Excel导入MongoDB最后一行未插入问题修复及代码优化咨询

一、核心问题原因 #

你的代码中fast-csv的data事件回调使用了async/await，但fast-csv的事件触发是同步逻辑——它不会等待data事件里的异步操作（比如OurMedicine.findOne）完成，就会继续触发下一个data事件，甚至可能提前触发end事件。这就导致最后一行数据的异步检查还没完成，end事件已经执行，最后一行的数据可能还没被加入insertDocuments数组，从而无法插入数据库。

二、修复代码（解决最后一行丢失问题） #

修改思路：

• 在data事件触发时暂停解析器，避免后续数据涌入导致异步任务堆积
• 收集所有异步任务的Promise，确保在end事件前所有异步操作都完成
• 调整end事件逻辑，等待所有异步任务完成后再执行批量插入

const { OurMedicine } = require('../../models/pharamacists/ourMedicine');
const XLSX = require('xlsx');
const fastcsv = require('fast-csv');

async function ourMedicine(req, res) {
  try {
    if (!req.file) {
      return res.status(400).json({ error: 'Error: No file uploaded' });
    }

    const fileExtension = req.file.originalname.split('.').pop();
    if (fileExtension !== 'xlsx') {
      return res.status(400).json({ error: 'Error: File must be in .xlsx format' });
    }

    // 关闭不必要的解析选项，减少内存占用
    const workbook = XLSX.readFile(req.file.path, { cellStyles: false, cellHTML: false });
    const sheetName = workbook.SheetNames[0];
    const sheet = workbook.Sheets[sheetName];

    const expectedHeaders = {
      'name': 'A1',
      'age': 'B1'
    };

    for (const [headerName, cell] of Object.entries(expectedHeaders)) {
      const sheetCell = sheet[cell];
      const trimmedHeaderName = headerName.trim();
      if (!sheetCell || typeof sheetCell.v !== 'string') {
        return res.status(400).json({
          message: `请在单元格${cell}中填入${trimmedHeaderName}`
        });
      } else if (sheetCell.v.trim().toLowerCase() !== trimmedHeaderName) {
        return res.status(400).json({
          message: `请将单元格${cell}重命名为${trimmedHeaderName}`
        });
      }
    }

    for (const cellAddress in sheet) {
      const sheetCell = sheet[cellAddress];
      if (sheetCell.v && typeof sheetCell.v === 'string' && sheetCell.v.length > 600) {
        return res.status(400).json({
          message: `单元格${cellAddress}的内容长度请控制在600字符以内`
        });
      }
    }

    const csvStream = XLSX.stream.to_csv(sheet);

    const newMedicines = [];
    const alreadyExist = [];
    const insertDocuments = [];
    // 收集所有异步任务的Promise，确保全部完成
    const asyncTasks = [];

    const parser = fastcsv.parse({ headers: true })
      .on('data', (data) => {
        // 暂停解析，避免异步任务堆积
        parser.pause();

        // 处理数据格式转换
        const keys = Object.keys(data);
        const lowerCaseKeys = keys.map(key => key.toLowerCase().trim());
        const newData = {};
        for (let i = 0; i < keys.length; i++) {
          newData[lowerCaseKeys[i]] = data[keys[i]];
        }

        // 将异步操作包装成Promise，加入任务列表
        const task = (async () => {
          const existingDoc = await OurMedicine.findOne({ name: newData.name });
          if (existingDoc) {
            alreadyExist.push({ bran: existingDoc.name, rigistration_number: existingDoc.age });
          } else {
            newMedicines.push(newData.name);
            insertDocuments.push(newData);
          }
        })();

        asyncTasks.push(task);
        // 恢复解析
        parser.resume();
      })
      .on('end', async () => {
        // 等待所有异步任务完成
        await Promise.all(asyncTasks);

        if (insertDocuments.length > 0) {
          const result = await OurMedicine.bulkWrite(
            insertDocuments.map(doc => ({
              insertOne: { document: doc }
            }))
          );
          console.log(`${result.insertedCount} 条文档插入MongoDB`);
        }

        res.json({ newMedicines, alreadyExist });
      });

    csvStream.pipe(parser);

  } catch (ex) {
    console.error(ex);
    res.status(500).json({ error: 'Internal server error' });
  }
}

module.exports = { ourMedicine };

三、大数据量场景优化建议 #

针对大量数据导入，以下几点可以显著提升性能和稳定性：

• 批量查询重复数据，减少数据库请求：不要每条数据都单独调用findOne，每积累100-500条待检查数据，用$in批量查询已存在的文档，大幅降低数据库IO次数。
• 分批批量插入，避免内存溢出：不要将所有待插入数据存在数组中，每积累500-1000条就执行一次bulkWrite，然后清空数组，避免内存占用过高。
• 流式读取Excel文件：对于超大Excel文件，改用XLSX.stream.to_json配合文件流读取，避免一次性加载整个文件到内存。
• 关闭不必要的Excel解析选项：读取Excel时关闭cellStyles、cellHTML等不需要的解析项，减少内存占用。
• 设置批量插入无序模式：如果不关心插入顺序，在bulkWrite中添加{ ordered: false }，这样单条插入失败不会终止整个批量操作，提升容错性。

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