Node.js中AMQP库如何在消费完所有队列消息后关闭消费者连接

你当前使用的ch.consume是常驻推送消费模式，启动后会持续监听队列新消息，不会自动终止，要实现「消费完队列所有消息后关闭连接」，可以根据场景选择以下两种实现方式：

方案1：主动拉取模式（逻辑最简洁，推荐） #

使用通道的get方法主动轮询拉取消息，当接口返回空值时代表队列已无待消费消息，直接执行资源关闭逻辑即可，不会残留常驻监听。

const amqp = require('amqplib');

async function consumeAllMessages() {
  let connection = null;
  try {
    connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queueName = 'hello';

    // 声明队列确保存在
    await channel.assertQueue(queueName, { durable: false });
    console.log('开始消费队列存量消息');

    // 循环拉取直到队列为空
    while (true) {
      const message = await channel.get(queueName, { noAck: false });
      // 无消息时退出循环
      if (!message) {
        console.log('所有消息消费完成');
        break;
      }
      console.log(`接收到消息: ${message.content.toString()}`);
      // 手动确认消息消费成功
      channel.ack(message);
    }

    // 依次关闭通道、连接
    await channel.close();
    await connection.close();
    console.log('AMQP连接已正常关闭');
  } catch (error) {
    console.warn('消费异常:', error);
    // 异常场景兜底关闭连接
    if (connection) await connection.close();
    process.exit(1);
  }
}

consumeAllMessages();

方案2：推送消费模式下自动终止 #

如果你需要保留推送消费的性能优势，可以在每次消费消息后检查队列剩余消息量，当剩余消息数为0时主动取消消费者、关闭连接。

const amqp = require('amqplib');

amqp.connect('amqp://localhost').then(async function(conn) {
  const ch = await conn.createChannel();
  const queueName = 'hello';
  let consumerTag = null;

  await ch.assertQueue(queueName, { durable: false });
  // 设置预取数为1，避免本地缓存消息导致队列长度判断不准
  await ch.prefetch(1);

  const consumeResult = await ch.consume(queueName, async function(msg) {
    console.log(` [x] Received '%s'`, msg.content.toString());
    ch.ack(msg);

    // 消费后检查队列状态
    const queueStatus = await ch.checkQueue(queueName);
    // 队列剩余消息数为0时终止消费
    if (queueStatus.messageCount === 0) {
      await ch.cancel(consumerTag);
      await ch.close();
      await conn.close();
      console.log('消息消费完成，连接已关闭');
    }
  }, { noAck: false });

  consumerTag = consumeResult.consumerTag;
  console.log(' [*] 等待消费消息...');

  // 兜底处理：启动时队列已经为空的场景
  const initQueueStatus = await ch.checkQueue(queueName);
  if (initQueueStatus.messageCount === 0) {
    await ch.close();
    await conn.close();
    console.log('队列为空，连接已关闭');
  }
}).catch(console.warn);

注意事项 #

• 如果队列有多个消费者同时消费，建议将预取数prefetch设为1，避免本地预缓存消息导致队列长度判断偏差
• 如果消费过程中持续有生产者向队列发送新消息，上述逻辑会持续消费直到队列完全为空才退出；如果仅需要消费程序启动时已经存在的存量消息，可以在启动时先记录队列初始消息总数，累计消费达到对应数值时直接退出即可
• 手动确认模式下一定要执行ack操作，否则消息会重新入队导致逻辑异常

内容的提问来源于stack exchange，提问作者Võ Khắc Bảo