发送大量消息时RabbitMQ连接失败，少量消息正常的原因排查

你的代码核心问题是每次发送一条消息就新建一个RabbitMQ连接。当发送2000条消息时，会瞬间创建2000个连接，这直接触发了CloudAMQP的连接数限制（CloudAMQP的免费/基础套餐通常有严格的连接数上限，比如最多10-20个并发连接），同时短时间内大量TCP连接请求也会被服务器或网络层面拦截，导致ECONNREFUSED和连接被主动关闭的错误。

少量消息时，连接数还没达到限制，所以能正常工作；但消息量上去后，连接数超限就会被拒绝。

1. 复用连接与通道 #

RabbitMQ的设计是一个连接可以创建多个通道，通道是轻量级的，适合复用。你应该在应用启动时创建一次连接和通道，之后所有消息都通过这个通道发送，而不是每次新建连接。

修改后的代码示例：

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

const app = express();
const port = process.env.PORT || 3300;

// 全局保存连接和通道
let rabbitConnection = null;
let rabbitChannel = null;
const QUEUE_NAME = "review_data";

// 启动时初始化RabbitMQ连接和通道
function initRabbitMQ() {
  amqp.connect(`amqps://nuogafvveJoldl_RhEyfs7qJzX0BIRep6J8rFIN-0@puffin.rmq2.cloudamqp.com/xxxxxxx`, (err, connection) => {
    if (err) {
      console.error("RabbitMQ连接失败:", err);
      // 可选：添加重连逻辑，间隔几秒重试
      setTimeout(initRabbitMQ, 5000);
      return;
    }
    rabbitConnection = connection;
    console.log("RabbitMQ连接成功");

    connection.createChannel((err, channel) => {
      if (err) {
        console.error("创建通道失败:", err);
        return;
      }
      rabbitChannel = channel;
      // 预先声明队列（只需要做一次）
      channel.assertQueue(QUEUE_NAME, { durable: false });
      console.log("RabbitMQ通道初始化完成");
    });

    // 监听连接关闭事件，自动重连
    connection.on('close', () => {
      console.log("RabbitMQ连接已关闭，准备重连");
      rabbitConnection = null;
      rabbitChannel = null;
      setTimeout(initRabbitMQ, 5000);
    });
  });
}

// 初始化RabbitMQ
initRabbitMQ();

// 读取Excel文件
const file = XLSX.readFile('./limit9.xlsx')

app.get('/', (req, res) => {
  if (!rabbitChannel) {
    res.status(503).send("RabbitMQ通道未就绪，请稍后重试");
    return;
  }

  const sheets = file.SheetNames;
  let messageCount = 0;
  for (let i = 0; i < sheets.length; i++) {
    const temp = XLSX.utils.sheet_to_json(file.Sheets[file.SheetNames[i]]);
    temp.forEach((rowData) => {
      const json = JSON.stringify(rowData);
      rabbitChannel.sendToQueue(QUEUE_NAME, Buffer.from(json));
      messageCount++;
    });
  }
  res.send(`成功发送 ${messageCount} 条消息到队列`);
});

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

2. 额外优化建议 #

• 添加消息确认：如果需要确保消息被队列接收，可以使用sendToQueue的回调参数，或者开启通道的confirm模式，处理消息发送失败的情况。
• 批量处理：如果Excel数据量极大，可以分批次发送，避免一次性占用过多内存。
• 连接池（可选）：如果确实需要多通道，可以使用连接池，但对于大多数场景，单连接+多通道已经足够。

关键注意点 #

• 永远不要在循环或请求处理中重复创建RabbitMQ连接，这是RabbitMQ客户端开发的基本规范。
• CloudAMQP的连接数、通道数、消息速率都有套餐限制，如果你需要更高的配额，可以升级套餐。

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