Node.js中RabbitMQ监器重连逻辑失效问题排查与解决

我在Node.js中开发了RabbitMQ监听器，期望实现断开后自动重连的弹性连接机制，但通过RabbitMQ管理界面强制断开连接后，程序并未触发重连操作。不确定是本地环境限制还是代码存在缺失，附上当前代码寻求技术帮助。

const amqp = require('amqplib');

const rabbitmqServerUrl = 'amqp://localhost';
const queueName = 'best_queue';

let connection = null;
let channel = null;

async function setupConnection() {
  try {
    connection = await amqp.connect(rabbitmqServerUrl);
    connection.on('error', (err) => {
      if (err.message.includes('Connection closed')) {
        console.error('Connection closed, reconnecting...');
        setTimeout(setupConnection, 5000); // Retry connection after a delay
      } else {
        console.error('Connection error:', err.message);
      }
    });

    channel = await connection.createChannel();

    // Create a durable queue
    await channel.assertQueue(queueName, { durable: true });

    console.log('Connected to RabbitMQ');

    // Start the consumer
    startConsumer();
  } catch (error) {
    console.error('Error connecting to RabbitMQ:', error.message);

    // Retry connection after a delay
    setTimeout(setupConnection, 5000);
  }
}

function startConsumer() {
  if (!channel) {
    console.error('Channel is not available, skipping consumer start.');
    return;
  }

  channel.consume(queueName, async (msg) => {
    if (msg !== null) {
      try {
        // Process the message
        console.log('Received message:', msg.content.toString());

        // Simulate a processing delay (replace this with your actual processing logic)
        await new Promise((resolve) => setTimeout(resolve, 1000));

        // Acknowledge the message
        channel.ack(msg);
      } catch (err) {
        console.error('Error processing message:', err.message);
        // Handle message processing errors as needed
      }
    }
  });

  channel.on('close', () => {
    console.error('Channel closed, reconnecting...');
    setTimeout(startConsumer, 5000); // Restart the consumer after a delay
  });

  channel.on('error', (error) => {
    console.error('Channel error:', error.message);
    // Handle channel errors as needed
  });

  console.log('Consumer started');
}

setupConnection();

核心问题 #

1. 未监听Connection的close事件：通过RabbitMQ管理界面强制断开连接时，通常只会触发connection的close事件，而非error事件，导致重连逻辑完全未执行。
2. 错误信息匹配过于严格：err.message.includes('Connection closed')无法覆盖所有连接断开场景的错误信息（比如实际可能是"Connection reset by peer"）。
3. Channel关闭后的重连逻辑错误：当前channel关闭后仅调用startConsumer，但此时connection已经失效，channel对象也不可用，必须重建整个连接。
4. 全局变量未重置：重连前没有将connection和channel置为null，可能导致引用旧的失效对象。

修复后的代码 #

const amqp = require('amqplib');

const rabbitmqServerUrl = 'amqp://localhost';
const queueName = 'best_queue';

let connection = null;
let channel = null;

async function setupConnection() {
  // 重连前重置全局变量，避免引用失效对象
  connection = null;
  channel = null;
  
  try {
    connection = await amqp.connect(rabbitmqServerUrl);
    console.log('Connected to RabbitMQ');

    // 监听connection的close事件，覆盖管理界面强制断开场景
    connection.on('close', () => {
      console.error('Connection closed, reconnecting in 5s...');
      setTimeout(setupConnection, 5000);
    });

    // 所有连接错误都触发重连，不再限制错误信息
    connection.on('error', (err) => {
      console.error('Connection error:', err.message);
      console.error('Reconnecting in 5s...');
      setTimeout(setupConnection, 5000);
    });

    channel = await connection.createChannel();
    await channel.assertQueue(queueName, { durable: true });

    startConsumer();
  } catch (error) {
    console.error('Failed to connect to RabbitMQ:', error.message);
    console.error('Retrying in 5s...');
    setTimeout(setupConnection, 5000);
  }
}

function startConsumer() {
  if (!channel) {
    console.error('Channel not available, cannot start consumer');
    return;
  }

  channel.consume(queueName, async (msg) => {
    if (msg === null) return;
    
    try {
      console.log('Received message:', msg.content.toString());
      // 替换为实际业务逻辑
      await new Promise(resolve => setTimeout(resolve, 1000));
      channel.ack(msg);
    } catch (err) {
      console.error('Message processing failed:', err.message);
      // 可选：处理消费失败，比如nack重新入队
      channel.nack(msg, false, true);
    }
  });

  channel.on('error', (err) => {
    console.error('Channel error:', err.message);
  });

  console.log('Consumer started');
}

setupConnection();

关键修改点 #

• 添加connection.on('close')监听，覆盖管理界面强制断开的场景
• 移除connection error的信息判断，所有连接错误都触发重连
• 重连前重置connection和channel变量，避免引用失效对象
• channel关闭时不再单独处理，而是由connection的close/error事件触发完整重连流程
• 优化消费失败的处理逻辑，添加nack重新入队的示例

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