如何在Node.js中高效处理PostgreSQL大数据集以避免高内存开销？

我完全理解你的顾虑——一次性把10万条用户数据加载到内存里，确实很容易让Node.js进程的内存占用飙升，甚至触发OOM（内存不足）错误。针对这种处理大规模PostgreSQL数据集的场景，我给你分享几个经过验证的最佳实践，直接可以套用到你的代码里：

1. 使用PostgreSQL游标进行流式查询（最推荐） #

PostgreSQL的游标（Cursor）可以让我们逐批从数据库拉取数据，而不是一次性把所有结果加载到客户端内存里。Node.js官方的pg库原生支持游标操作，内存占用会稳定在你设置的批次大小（比如500条），不会出现突然飙升的情况。

修改后的代码示例： #

首先，你需要调整数据查询逻辑为游标模式，同时保留原有的用户处理逻辑：

const { Pool } = require('pg');
const pool = new Pool(/* 你的数据库连接配置 */);

async function processMessageForSubscribers(channelId, channelName, message, addresses) {
  let client;
  try {
    client = await pool.connect();
    const CHUNK_SIZE = 500;
    const notifyTasks = [];
    const autotradeTasks = [];

    // 1. 创建游标：仅定义查询逻辑，不会立即拉取所有数据
    const query = `
      SELECT * FROM users 
      WHERE tracking_config @> $1::jsonb  -- 请根据你的表结构调整查询条件
    `;
    const cursor = client.query(query, [JSON.stringify({ [channelId]: {} })]);

    // 2. 逐批处理游标返回的数据
    let userChunk = [];
    for await (const user of cursor) {
      userChunk.push(user);

      // 当批次达到设定大小，立即处理
      if (userChunk.length >= CHUNK_SIZE) {
        await processUserChunk(userChunk, notifyTasks, autotradeTasks, channelId, addresses, message, channelName);
        userChunk = []; // 清空批次，释放内存
      }
    }

    // 处理最后一批不足CHUNK_SIZE的用户
    if (userChunk.length > 0) {
      await processUserChunk(userChunk, notifyTasks, autotradeTasks, channelId, addresses, message, channelName);
    }

    await queueTasks(notifyTasks, autotradeTasks);
  } catch (error) {
    console.error('Error processing subscribers:', error);
    throw error;
  } finally {
    if (client) client.release(); // 务必释放数据库连接
  }
}

// 抽离批次处理逻辑，复用性更强
async function processUserChunk(userChunk, notifyTasks, autotradeTasks, channelId, addresses, message, channelName) {
  await Promise.all(
    userChunk.map(async (user) => {
      const config = user.trackingConfig[channelId];
      const autotradeAmount = config?.autotradeAmount;
      
      if (config.newPost === 'NOTIFY') {
        createNotificationTask(user, addresses, message, channelId, channelName, autotradeAmount, notifyTasks);
      }
      // 这里可以补充原有的自动交易任务逻辑
    })
  );
}

2. 用键分页替代OFFSET分页（游标兼容备选方案） #

如果因为环境限制无法使用游标，键分页是比OFFSET更高效的分批查询方式。OFFSET在数据量大时会强制数据库跳过前面所有行，性能急剧下降；而键分页利用主键（比如id）的有序性，每次仅查询id > 上一批最大id的指定数量数据，内存占用同样可控。

代码示例： #

async function processMessageForSubscribers(channelId, channelName, message, addresses) {
  try {
    const CHUNK_SIZE = 500;
    const notifyTasks = [];
    const autotradeTasks = [];
    let lastUserId = 0; // 初始值设为小于最小用户ID的数值
    let hasMoreUsers = true;

    while (hasMoreUsers) {
      // 分批查询用户：仅拉取id大于上一批最大id的CHUNK_SIZE条数据
      const users = await getUsersByTrackedChannelWithKeyPagination(channelId, CHUNK_SIZE, lastUserId);
      
      if (users.length === 0) {
        hasMoreUsers = false;
        break;
      }

      // 更新下一批查询的起始id
      lastUserId = Math.max(...users.map(u => u.id));
      // 处理当前批次
      await processUserChunk(users, notifyTasks, autotradeTasks, channelId, addresses, message, channelName);
    }

    await queueTasks(notifyTasks, autotradeTasks);
  } catch (error) {
    console.error('Error processing subscribers:', error);
    throw error;
  }
}

// 对应的分页查询函数
async function getUsersByTrackedChannelWithKeyPagination(channelId, chunkSize, lastUserId) {
  const query = `
    SELECT * FROM users 
    WHERE tracking_config @> $1::jsonb 
      AND id > $2
    ORDER BY id ASC
    LIMIT $3
  `;
  const result = await pool.query(query, [JSON.stringify({ [channelId]: {} }), lastUserId, chunkSize]);
  return result.rows;
}

3. 额外优化建议 #

• 控制并发压力：当前CHUNK_SIZE=500的设置比较合理，但如果下游通知/交易服务承载能力有限，可以适当调小，避免压垮依赖服务。
• 及时清理内存：处理完每个批次后，确保清空临时数组（比如userChunk），让V8垃圾回收器能及时释放无用内存。
• 监控内存状态：可以在代码中加入console.log(process.memoryUsage().heapUsed / 1024 / 1024 + 'MB')，实时查看内存占用情况，验证优化效果。

备注：内容来源于stack exchange，提问作者Nightcrawler