如何让RabbitMQ Worker暂停接收消息且不丢失消息？

基于RabbitMQ构建工作队列：单生产者推送任务到队列，多Worker消费任务。现在需要实现：Worker执行校准任务期间，暂停接收队列消息，让其他Worker处理任务，避免消息滞留在当前Worker，且全程不丢失任何消息。

方案一：动态取消/重新注册消费者 #

核心逻辑：当触发校准需求时，先取消当前Worker的消费者（RabbitMQ会停止向该Worker分发新消息），完成校准后重新注册消费者，恢复任务接收。这种方式能确保校准期间Worker完全不会收到新消息，所有消息都会被其他可用Worker处理。

修改后的worker.js代码：

#!/usr/bin/env node

var amqp = require('amqplib');

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms))
}

let needCalibrate = false;
let consumerTag = null; // 保存消费者标签，用于后续取消消费

async function recalibrate() {
  console.log("Recalibration...");
  await sleep(10000);
  console.log("Done recalibration...");
  needCalibrate = false;
}

async function startConsuming(channel, queue) {
  // 注册消费者并保存标签
  const result = await channel.consume(queue, async function(msg) {
    const msgContent = msg.content.toString();
    const secs = msgContent.split('.').length - 1;
    
    console.log(" [x] Received %s", msgContent);
    await sleep(secs * 1000);
    console.log(" [x] Done");
    channel.ack(msg);

    // 检查是否触发校准
    if (msgContent.includes("b")) {
      needCalibrate = true;
      // 取消当前消费者，停止接收新消息
      await channel.cancel(consumerTag);
      console.log("Paused consuming for calibration");
      // 执行校准
      await recalibrate();
      // 校准完成后重新启动消费
      await startConsuming(channel, queue);
    }
  }, { noAck: false });
  
  consumerTag = result.consumerTag;
  console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", queue);
}

async function run() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  const queue = 'task_queue';

  await channel.assertQueue(queue);
  await channel.prefetch(1); // 确保每次只预取一条消息

  await startConsuming(channel, queue);
}

run();

方案二：Nack重新入队+延迟接收 #

如果不想频繁取消/注册消费者，可在收到触发校准的消息后，将消息nack并重新入队，让其他Worker处理，随后执行校准。需配合prefetch(1)确保不会预取多条消息，避免校准期间积压未处理的消息。

示例代码片段：

channel.consume(queue, async function(msg) {
  if (needCalibrate) {
    // 将消息重新入队，交给其他Worker处理
    channel.nack(msg, false, true);
    await recalibrate();
    return;
  }

  const msgContent = msg.content.toString();
  const secs = msgContent.split('.').length - 1;
  
  console.log(" [x] Received %s", msgContent);
  await sleep(secs * 1000);
  console.log(" [x] Done");
  channel.ack(msg);

  if (msgContent.includes("b")) {
    needCalibrate = true;
  }
}, { noAck: false });

• 方案一：可靠性最高，校准期间Worker完全停止接收消息，消息100%由其他Worker处理，适合对消息分发逻辑有严格要求的场景。
• 方案二：实现简单，但依赖RabbitMQ的重新入队机制，可能存在消息短暂重复分发的情况，适合对一致性要求不高的场景。

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