如何用Node.js向SQL Server插入千条以上CSV记录（批量插入异常）

问题概述 #

通过EJS表单获取table_name和csv_file，用Node.js解析CSV并批量插入SQL Server。处理少于100条记录的CSV正常，但千条以上记录时无法完成上传，尝试50条批量插入仍无效，且无报错信息。

核心问题分析 #

1. 内存过载：将整个CSV文件的所有记录一次性存入results数组，大文件会耗尽Node.js进程内存，导致进程无响应甚至静默崩溃，无错误抛出。
2. SQL语句拼接风险：直接拼接字符串生成INSERT语句，若CSV中包含单引号、换行等特殊字符，会导致SQL语句语法错误，部分批次执行失败但未被正确捕获；同时存在严重SQL注入风险。
3. 低效的批量插入方式：循环中每次创建新的request并执行单条INSERT语句，频繁的数据库请求会导致性能瓶颈，大文件处理时超时概率极高。
4. 潜在的超时问题：HTTP请求默认存在超时限制，大文件解析+插入耗时过长会导致客户端断开连接，服务器端可能仍在处理但用户感知为上传失败。

解决方案 #

1. 流式处理CSV，降低内存占用 #

不要一次性加载所有数据到内存，而是边读取CSV行边累积批次，达到批次大小后立即插入数据库，避免内存溢出。

2. 使用参数化查询+批量插入API #

利用mssql的Table类进行批量插入，既避免SQL注入，又提升插入效率，同时自动处理特殊字符转义。

3. 优化连接与错误处理 #

• 确保数据库连接池正确复用，避免频繁创建/关闭连接。
• 增加关键步骤的日志输出，便于定位卡住的环节。
• 延长HTTP请求超时时间，避免客户端提前断开。

修改后的完整代码 #

const express = require('express');
const multer = require('multer');
const csv = require('csv-parser');
const fs = require('fs').promises;
const mssql = require('mssql');

const app = express();
const upload = multer({ dest: 'uploads/' });
const config = { /* 你的SQL Server配置 */ };

// 延长请求超时与数据大小限制，根据实际情况调整
app.use(express.json({ limit: '50mb' }));
app.use(express.urlencoded({ extended: true, limit: '50mb' }));

app.post('/', upload.single('csvFile'), async (req, res) => {
  let pool;
  let table;
  const batchSize = 100; // 可根据数据库性能调整批次大小
  let currentBatch = [];
  let columns = [];
  const tableName = req.body.tableName;

  try {
    // 1. 检查目标表是否已存在
    pool = await mssql.connect(config);
    const tableExists = await pool.request()
      .input('tableName', mssql.NVarChar, tableName)
      .query(`SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = @tableName`);
    
    if (tableExists.recordset.length > 0) {
      await fs.unlink(req.file.path);
      res.send(`Table ${tableName} already exists.`);
      console.log(`Table ${tableName} already exists.`);
      return;
    }

    // 2. 流式解析CSV并批量插入
    await new Promise((resolve, reject) => {
      const stream = fs.createReadStream(req.file.path)
        .pipe(csv())
        .on('data', async (row) => {
          // 初始化列名与批量插入Table实例
          if (columns.length === 0) {
            columns = Object.keys(row);
            table = new mssql.Table(tableName);
            table.create = true; // 自动创建表
            table.columns.add('Id', mssql.Int, { identity: true, primary: true });
            columns.forEach(col => {
              table.columns.add(col, mssql.VarChar(mssql.MAX));
            });
          }

          currentBatch.push(row);
          // 达到批次大小则执行插入
          if (currentBatch.length >= batchSize) {
            stream.pause(); // 暂停流避免数据堆积
            // 将批次数据加入Table
            currentBatch.forEach(row => {
              const values = columns.map(col => row[col]);
              table.rows.add(...values);
            });
            // 执行批量插入
            await new mssql.Request(pool).bulk(table);
            // 清空批次并恢复流
            currentBatch = [];
            table.rows.clear();
            stream.resume();
          }
        })
        .on('end', async () => {
          // 处理剩余不足一个批次的数据
          if (currentBatch.length > 0) {
            currentBatch.forEach(row => {
              const values = columns.map(col => row[col]);
              table.rows.add(...values);
            });
            await new mssql.Request(pool).bulk(table);
          }
          // 删除临时上传文件
          await fs.unlink(req.file.path);
          resolve();
        })
        .on('error', async (err) => {
          await fs.unlink(req.file.path).catch(() => {});
          reject(err);
        });
    });

    // 关闭数据库连接池
    await pool.close();
    res.send('success');
    console.log('数据插入完成');
  } catch (error) {
    console.error('处理失败:', error);
    // 异常时清理临时文件与数据库连接
    if (req.file?.path) {
      await fs.unlink(req.file.path).catch(() => {});
    }
    if (pool) await pool.close().catch(() => {});
    res.send('error');
  }
});

额外优化建议 #

• 增加文件大小限制：在multer配置中设置limits: { fileSize: 50 * 1024 * 1024 }（限制50MB），避免超大文件拖垮服务器。
• 进度反馈：通过WebSocket或SSE向客户端返回实时处理进度，提升用户体验。
• 索引优化：数据插入完成后，根据业务查询需求添加合适的索引，避免后续查询性能问题。

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