发送大量消息时RabbitMQ出现ECONNREFUSED错误的原因排查

我在应用中采用RabbitMQ作为队列解决方案，使用CloudAMQP免费版（支持队列容量10000条消息）。测试10条消息时一切正常，但尝试发送9000条消息到队列时，出现以下错误：

Error: connect ECONNREFUSED 15.206.75.114:5671
at TCPConnectWrap.afterConnect [as oncomplete] (node:net:1247:16) {
  errno: -111,
  code: 'ECONNREFUSED',
  syscall: 'connect',
  address: '15.206.75.114',
  port: 5671
}

以下是从含约9000行数据的Excel表（使用XLSX库读取）发送消息到队列的代码：

const express = require('express');
const XLSX = require("xlsx");
const amqp = require('amqplib/callback_api');
require('dotenv').config();

const app = express();
const port = process.env.PORT || 3300;
const file = XLSX.readFile('./limit9.xlsx')

app.get('/', (req, res) => {
  const sheets = file.SheetNames;
  for (let i = 0; i < sheets.length; i++) {
    const temp = XLSX.utils.sheet_to_json(
      file.Sheets[file.SheetNames[i]]
    )
    temp.forEach((res) => {
      sendData(res);  //This function will send data to RabbitMQ
    })
   }
});

function sendData(res) {
  amqp.connect(process.env.RABBITMQ_URL, (err, connection) => {
    if (err) {
      throw err;
    } else {
      connection.createChannel((err, channel) => {
        if (err) {
          throw err;
        } else {
          let queu = "review_data";
          channel.assertQueue(queu, {
            durable: false
          });
          var json = JSON.stringify(res);
          channel.sendToQueue(queu, Buffer.from(json));
        }
      });
    }
  });
}

为何上述代码处理少量数据正常，处理大量数据时却失败？

核心原因 #

你的代码在每条消息发送时都新建一个RabbitMQ连接，处理9000条消息会瞬间发起9000次连接请求。CloudAMQP免费版有严格的并发连接数限制（通常仅支持10-20个并发连接），同时操作系统的临时端口资源会被快速耗尽，导致后续连接请求被拒绝，出现ECONNREFUSED错误。

修复方案 #

只创建一个RabbitMQ连接和一个通道，复用它们发送所有消息，而非每条消息新建连接：

修改后的代码示例：

const express = require('express');
const XLSX = require("xlsx");
const amqp = require('amqplib/callback_api');
require('dotenv').config();

const app = express();
const port = process.env.PORT || 3300;
const file = XLSX.readFile('./limit9.xlsx');

// 全局复用连接和通道
let rabbitConnection = null;
let rabbitChannel = null;

// 初始化RabbitMQ连接与通道
function initRabbitMQ() {
  return new Promise((resolve, reject) => {
    amqp.connect(process.env.RABBITMQ_URL, (err, connection) => {
      if (err) {
        reject(err);
        return;
      }
      rabbitConnection = connection;
      connection.createChannel((err, channel) => {
        if (err) {
          reject(err);
          return;
        }
        rabbitChannel = channel;
        const queue = "review_data";
        channel.assertQueue(queue, { durable: false });
        resolve();
      });
    });
  });
}

app.get('/', async (req, res) => {
  try {
    // 确保连接已初始化
    if (!rabbitConnection || !rabbitChannel) {
      await initRabbitMQ();
    }

    const sheets = file.SheetNames;
    for (let i = 0; i < sheets.length; i++) {
      const temp = XLSX.utils.sheet_to_json(
        file.Sheets[file.SheetNames[i]]
      );
      temp.forEach((data) => {
        const json = JSON.stringify(data);
        rabbitChannel.sendToQueue("review_data", Buffer.from(json));
      });
    }
    res.send("消息发送完成");
  } catch (err) {
    console.error("发送失败:", err);
    res.status(500).send("发送出错");
  }
});

// 服务器关闭时清理连接
process.on('exit', () => {
  if (rabbitChannel) rabbitChannel.close();
  if (rabbitConnection) rabbitConnection.close();
});

app.listen(port, () => {
  console.log(`服务器运行在端口 ${port}`);
});

额外优化建议 #

1. 添加消息确认：用sendToQueue的回调函数确认消息被RabbitMQ接收，避免丢失：

   rabbitChannel.sendToQueue("review_data", Buffer.from(json), (err) => {
     if (err) console.error("消息发送失败:", data);
   });
   

2. 限流发送：若CloudAMQP有速率限制，可添加批量延迟，比如每发100条暂停100ms，避免触发限制。
3. 错误重试：捕获发送错误，对失败消息进行重试，保证数据完整性。

内容的提问来源于stack exchange，提问作者Digvijay Singh Thakur