使用pg-promise批量插入80万条PostgreSQL数据时连接意外终止

从PostgreSQL日志中的signal 9: Killed可以确定，问题本质是系统OOM（内存不足）杀手终止了PostgreSQL进程——批量插入大量数据时，内存占用超过系统可用阈值，系统强制杀死相关进程，导致数据库连接意外中断。30万条能成功是因为内存占用未触达临界值，和连接超时参数无关。

1. 拆分批量插入为小批次提交 #

不要一次性将80万条数据塞进单个INSERT语句，拆分多个小批次（比如每1万-5万条为一批），每批次提交一次，大幅降低单次操作的内存占用。

示例代码（使用pg-promise的helpers生成插入语句）：

async function batchInsert(data) {
  const batchSize = 10000; // 根据系统内存调整批次大小
  const tableName = 'your_table';
  const columns = ['col1', 'col2', 'col3']; // 替换为你的表字段

  for (let i = 0; i < data.length; i += batchSize) {
    const batch = data.slice(i, i + batchSize);
    const insertQuery = pgp.helpers.insert(batch, columns, tableName);
    await db.none(insertQuery);
  }
}

2. 使用PostgreSQL COPY命令替代普通INSERT #

COPY是PostgreSQL专为批量数据导入设计的命令，直接写入数据文件，内存开销远低于普通INSERT，效率也更高。

示例代码（结合pg-copy-streams实现流式COPY）：

const copyFrom = require('pg-copy-streams').from;

async function copyInsert(data) {
  const tableName = 'your_table';
  const columns = ['col1', 'col2', 'col3'];
  
  // 将数据转换为CSV格式（注意处理特殊字符转义）
  const csvLines = data.map(row => {
    // 对包含逗号、引号的字段做转义
    return columns.map(col => `"${String(row[col]).replace(/"/g, '""')}"`).join(',');
  });
  const csvContent = csvLines.join('\n');

  return new Promise((resolve, reject) => {
    const stream = db.client.query(copyFrom(`COPY ${tableName}(${columns.join(',')}) FROM STDIN WITH (FORMAT csv, HEADER false)`));
    stream.write(csvContent);
    stream.end();
    stream.on('finish', resolve);
    stream.on('error', reject);
  });
}

3. 调整PostgreSQL内存配置 #

修改postgresql.conf中的参数，降低批量操作的内存占用：

• work_mem: 降低单操作内存限制，比如从默认4MB改为2MB，避免排序、哈希操作占用过多内存
• shared_buffers: 若系统内存紧张，调小该值（建议为系统内存的1/4以内）
• maintenance_work_mem: 若涉及索引维护，适当降低该值

修改后重启PostgreSQL服务生效。

4. 优化Node.js端内存使用 #

避免一次性加载所有80万条数据到内存，改用流式读取/处理：

• 若数据来自文件，用readline或stream模块逐行读取，积累到指定批次后再插入
• 若数据来自API/数据库，分页获取后分批插入

示例（流式读取CSV文件并分批插入）：

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

async function streamInsertFromCsv(filePath) {
  const rl = readline.createInterface({
    input: fs.createReadStream(filePath),
    crlfDelay: Infinity
  });

  let batch = [];
  const batchSize = 10000;
  const tableName = 'your_table';
  const columns = ['col1', 'col2', 'col3'];

  for await (const line of rl) {
    const rowData = line.split(',').map(item => item.replace(/^"|"$/g, '')); // 解析CSV字段
    batch.push({
      col1: rowData[0],
      col2: rowData[1],
      col3: rowData[2]
    });

    if (batch.length >= batchSize) {
      const insertQuery = pgp.helpers.insert(batch, columns, tableName);
      await db.none(insertQuery);
      batch = [];
    }
  }

  // 插入剩余的最后一批数据
  if (batch.length > 0) {
    const insertQuery = pgp.helpers.insert(batch, columns, tableName);
    await db.none(insertQuery);
  }
}

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