Express.js后端每秒Socket数据的PostgreSQL存储优化方案咨询

当前实现的潜在问题 #

你的现有代码存在两个关键问题：

1. 使用Array.map配合async/await时，map不会等待异步操作完成，会同时发起所有插入请求，且在最后一条数据的回调中清空数组——这会导致插入未完成就丢失未成功写入的数据。
2. 循环执行单条INSERT语句，没有利用PostgreSQL的批量写入能力，数据库请求次数过多，效率低下。

以下是几种更优的替代方案，均支持服务器故障时丢失临时数据的需求：

方案1：用PostgreSQL批量INSERT优化现有逻辑 #

这是最直接的优化，无需引入第三方库，仅通过构造批量INSERT语句提升效率，同时修复异步逻辑问题。

let tempValues = [];

setInterval(async () => {
  if (tempValues.length === 0) return;

  // 构造批量插入的占位符和参数数组
  const placeholders = tempValues.map((_, idx) => 
    `($${idx * 3 + 1}, $${idx * 3 + 2}, $${idx * 3 + 3})`
  ).join(', ');
  const insertValues = tempValues.flatMap(item => 
    [item.name, item.value, item.timestamp]
  );

  try {
    // 单次请求插入所有数据
    await pool.query(
      `INSERT INTO plc_options (name, value, date) VALUES ${placeholders}`,
      insertValues
    );
    tempValues = []; // 确保所有数据写入成功后再清空
  } catch (err) {
    console.error('批量插入失败:', err);
    // 可选：若允许丢失数据，此处可直接清空；若需重试，可保留数据等待下次执行
  }
}, 60000);

socket.onAny((eventName, ...args) => {
  tempValues.push({ name: eventName, value: args[0], timestamp: new Date() });
});

优势 #

• 无需额外依赖，代码改动小
• 大幅减少数据库请求次数，提升写入效率
• 修复异步逻辑问题，确保数据写入成功后再清空缓存

方案2：用轻量异步队列库实现多条件触发（数量+时间） #

如果希望同时支持「攒够N条数据立即写入」和「到时间强制写入」两种触发逻辑，可以用fastq这类轻量异步队列库，避免内存数据积压过多。

首先安装依赖：

npm install fastq

实现代码：

const fastq = require('fastq');
let tempValues = [];

// 定义批量处理函数，复用方案1的批量INSERT逻辑
const batchInsert = async (batchData) => {
  const placeholders = batchData.map((_, idx) => 
    `($${idx * 3 + 1}, $${idx * 3 + 2}, $${idx * 3 + 3})`
  ).join(', ');
  const insertValues = batchData.flatMap(item => 
    [item.name, item.value, item.timestamp]
  );
  await pool.query(
    `INSERT INTO plc_options (name, value, date) VALUES ${placeholders}`,
    insertValues
  );
};

// 创建异步队列，限制同时处理的批次数量
const queue = fastq.promise(batchInsert, 1);

// 定时触发批量写入（每分钟一次）
setInterval(async () => {
  if (tempValues.length > 0) {
    await queue.push([...tempValues]);
    tempValues = [];
  }
}, 60000);

// 当数据攒够100条时，立即触发批量写入
socket.onAny((eventName, ...args) => {
  tempValues.push({ name: eventName, value: args[0], timestamp: new Date() });
  
  if (tempValues.length >= 100) {
    queue.push([...tempValues]).then(() => {
      tempValues = [];
    }).catch(err => {
      console.error('批量插入失败:', err);
    });
  }
});

优势 #

• 支持双条件触发，平衡实时性和写入效率
• 队列机制确保异步任务有序执行，避免并发写入冲突
• 轻量库，性能开销极低

方案3：用PostgreSQL COPY命令实现超高效率写入 #

如果每秒产生的数据量极大（比如上千条），PostgreSQL的COPY命令是效率最高的批量写入方式，比批量INSERT快数倍。

const { Readable } = require('stream');
let tempValues = [];

setInterval(async () => {
  if (tempValues.length === 0) return;

  // 将缓存数据转换成CSV格式（注意处理特殊字符，比如逗号、引号）
  const csvContent = tempValues.map(item => {
    // 简单转义处理，避免CSV格式错误
    const escapedName = item.name.replace(/"/g, '""');
    const escapedValue = item.value.replace(/"/g, '""');
    return `"${escapedName}","${escapedValue}","${item.timestamp.toISOString()}"`;
  }).join('\n');

  try {
    // 创建可读流并通过COPY命令导入数据
    const stream = Readable.from([csvContent]);
    await new Promise((resolve, reject) => {
      stream.pipe(pool.query.stream(
        'COPY plc_options (name, value, date) FROM STDIN WITH (FORMAT csv, HEADER false, DELIMITER ",")'
      ))
      .on('finish', resolve)
      .on('error', reject);
    });
    tempValues = [];
  } catch (err) {
    console.error('COPY批量写入失败:', err);
  }
}, 60000);

socket.onAny((eventName, ...args) => {
  tempValues.push({ name: eventName, value: args[0], timestamp: new Date() });
});

优势 #

• 数据库层面的批量导入，效率远高于INSERT语句
• 适合超大规模数据的批量写入场景

方案选择建议 #

• 数据量较小（每秒几十条）：优先选方案1，简单高效无依赖
• 需要灵活触发逻辑：选方案2，平衡实时性和性能
• 数据量极大（每秒数百上千条）：选方案3，极致写入效率

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