Node.js异步调用如何设置优先级？解决Node-RED节点I/O堆积内存溢出问题

首先得明确：Node.js的事件循环本身没有原生的回调优先级机制，但我们可以通过控制任务的执行节奏、自定义优先级队列来避免回调堆积，解决内存溢出和进程停滞的问题。下面是几个实用的方案：

1. 实现请求限流（最直接有效） #

与其让所有数据库请求同时涌入回调队列，不如限制并发执行的I/O请求数量，从根源上避免队列爆炸。你可以用p-limit这类库（或者自己实现简单的限流逻辑）来控制并发数：

// 先安装p-limit：npm install p-limit
const pLimit = require('p-limit');
// 根据数据库承载能力调整并发数，比如设为100
const limit = pLimit(100);

// 在Node-RED节点的input事件中使用
node.on('input', async (msg) => {
  // 把数据库调用包装成限流任务
  const processTask = limit(async () => {
    try {
      // 你的数据库I/O操作，比如查询/插入
      const result = await db.query('SELECT * FROM table WHERE id = ?', [msg.payload.id]);
      // 处理结果后发送到下一个节点
      node.send({ payload: result });
    } catch (err) {
      node.error('数据库操作失败', err);
    }
  });

  // 执行限流任务
  await processTask();
});

这样一来，最多只有100个数据库请求同时处于等待执行状态，剩下的请求会排队等待前面的任务完成，不会一次性把回调队列塞满，内存占用就能稳定在可控范围。

2. 自定义带优先级的任务队列 #

如果确实需要给不同请求设置优先级（比如紧急业务请求优先处理），可以自己实现一个优先级队列，每次从队列中取出最高优先级的任务执行：

// 简易优先级队列实现
class PriorityQueue {
  constructor() {
    this.queue = [];
  }

  add(task, priority = 0) {
    // 优先级数字越小，优先级越高（比如0>1>2）
    this.queue.push({ task, priority });
    // 按优先级排序
    this.queue.sort((a, b) => a.priority - b.priority);
  }

  async processNext() {
    if (this.queue.length === 0) return;
    const { task } = this.queue.shift();
    try {
      await task();
    } catch (err) {
      node.error('任务执行失败', err);
    }
    // 递归处理下一个任务
    this.processNext();
  }
}

// 初始化队列，启动自动处理
const taskQueue = new PriorityQueue();
taskQueue.processNext();

// 在input事件中添加任务
node.on('input', (msg) => {
  // 根据业务逻辑设置优先级，比如紧急请求设为0，普通请求设为1
  const priority = msg.payload.isEmergency ? 0 : 1;

  const task = async () => {
    const result = await db.query('SELECT * FROM table WHERE id = ?', [msg.payload.id]);
    node.send({ payload: result });
  };

  taskQueue.add(task, priority);
});

这个方案既能让高优先级请求优先得到处理，又能控制任务执行节奏，避免队列无限膨胀。

3. 优化数据库连接池配置 #

内存溢出也可能和数据库连接池的不合理配置有关——如果连接池设置过大，每个连接都会占用内存，加上大量等待的回调，内存压力会陡增。调整连接池参数：

// 以mysql2为例，调整连接池配置
const mysql = require('mysql2/promise');
const pool = mysql.createPool({
  host: 'localhost',
  user: 'user',
  password: 'password',
  database: 'db',
  // 限制连接池大小，根据数据库性能设为50-100
  connectionLimit: 50,
  // 闲置连接超时自动释放，减少无用内存占用
  idleTimeout: 60000
});

合理的连接池大小能避免数据库资源耗尽，同时降低内存占用。

4. 批量处理请求 #

如果业务场景允许，把多个输入事件合并成一个批量数据库请求，能大幅减少I/O次数和回调数量：

let batch = [];
const BATCH_SIZE = 100; // 每100个请求批量处理一次
let batchTimer = null;

node.on('input', (msg) => {
  // 保存完整msg对象，方便后续对应结果
  batch.push(msg);

  // 达到批量大小或者超时（比如1秒）就执行批量查询
  if (batch.length >= BATCH_SIZE) {
    processBatch();
  } else if (!batchTimer) {
    batchTimer = setTimeout(processBatch, 1000);
  }
});

async function processBatch() {
  clearTimeout(batchTimer);
  batchTimer = null;

  if (batch.length === 0) return;

  try {
    // 提取批量ID，构造IN语句
    const ids = batch.map(item => item.payload.id);
    const result = await db.query(`SELECT * FROM table WHERE id IN (${ids.map(() => '?').join(',')})`, ids);
    
    // 将结果对应到原始请求，发送到下一个节点
    result.forEach((row, index) => {
      node.send({ 
        payload: row, 
        originalMsg: batch[index] 
      });
    });
  } catch (err) {
    node.error('批量数据库操作失败', err);
  }

  batch = [];
}

批量处理能把100个请求压缩成1个数据库调用，回调数量直接减少99%，内存占用会显著降低。

最后提醒下：在Node-RED中，千万别在input事件里直接做无控制的耗时I/O操作，一定要注意并发量和任务队列的管理，不然很容易触发内存问题。

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