RabbitMQ RPC流程中如何实现消息确认与错误通知（Node.js）

方案一：主动返回错误响应 #

消费者处理出错时，不直接仅执行nack，而是构造包含错误信息的响应发送回客户端的回复队列，客户端通过响应中的状态字段判断请求结果。

消费者端代码 #

const amqp = require('amqplib');

async function consumeRpc() {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();
  const rpcQueue = 'rpc_pdf_queue';
  await ch.assertQueue(rpcQueue, { durable: false });
  ch.prefetch(1);

  ch.consume(rpcQueue, async (msg) => {
    const request = JSON.parse(msg.content.toString());
    let response;

    try {
      // 模拟业务校验：数据缺失抛出异常
      if (!request.templateId || !request.data) {
        throw new Error('生成PDF失败：缺少模板ID或业务数据');
      }
      // 正常生成PDF逻辑
      response = { status: 'success', pdfUrl: '/generated/pdf/xxx.pdf' };
      ch.ack(msg);
    } catch (err) {
      // 构造错误响应
      response = { status: 'error', message: err.message };
      // 若无需重试，直接ack避免消息重复积压；若需重试则改为nack(msg, false, true)
      ch.ack(msg);
    }

    // 发送响应到客户端指定的回复队列
    ch.sendToQueue(msg.properties.replyTo, Buffer.from(JSON.stringify(response)), {
      correlationId: msg.properties.correlationId
    });
  });
}

consumeRpc();

客户端端代码 #

const amqp = require('amqplib');

async function requestPdfGeneration(params) {
  const conn = await amqp.connect('amqp://localhost');
  const ch = await conn.createChannel();
  // 声明独占回复队列，用于接收响应
  const replyQueue = await ch.assertQueue('', { exclusive: true });

  const correlationId = generateUniqueId();
  // 设置超时兜底，避免无限等待
  const timeoutTimer = setTimeout(() => {
    ch.close();
    throw new Error('请求超时：PDF生成耗时过长');
  }, 60000); // 60秒超时，根据业务调整

  return new Promise((resolve, reject) => {
    ch.consume(replyQueue.queue, (msg) => {
      if (msg.properties.correlationId === correlationId) {
        clearTimeout(timeoutTimer);
        const response = JSON.parse(msg.content.toString());
        ch.ack(msg);
        ch.close();
        
        if (response.status === 'error') {
          reject(new Error(response.message));
        } else {
          resolve(response.pdfUrl);
        }
      }
    }, { noAck: false });

    // 发送RPC请求
    ch.sendToQueue('rpc_pdf_queue', Buffer.from(JSON.stringify(params)), {
      correlationId,
      replyTo: replyQueue.queue,
      persistent: false
    });
  });
}

// 生成唯一ID的辅助函数
function generateUniqueId() {
  return Date.now().toString(36) + Math.random().toString(36).slice(2);
}

// 调用示例
requestPdfGeneration({})
  .then(result => console.log('PDF生成成功:', result))
  .catch(err => console.error('PDF生成失败:', err.message));

方案二：结合死信队列与超时机制 #

如果业务需要对失败请求进行重试或归档，可以给RPC请求队列配置死信队列，同时给客户端回复队列设置TTL，客户端通过超时感知错误，失败消息进入死信队列后可后续处理。

关键配置 #

1. 客户端回复队列加TTL：

// 客户端声明带自动过期的回复队列，超时后队列自动销毁
const replyQueue = await ch.assertQueue('', { 
  exclusive: true,
  expires: 60000 // 60秒后队列自动删除
});

2. 消费者端nack消息（需重试场景）：

catch (err) {
  // nack时设置requeue为false，让消息进入死信队列
  ch.nack(msg, false, false);
}

3. 声明死信队列（提前配置）：

// 死信交换机与队列声明
const dlxExchange = 'rpc_pdf_dlx';
const dlQueue = 'rpc_pdf_dead_letter';
await ch.assertExchange(dlxExchange, 'direct', { durable: true });
await ch.assertQueue(dlQueue, { durable: true });
await ch.bindQueue(dlQueue, dlxExchange, 'pdf_failure');

// 给RPC请求队列绑定死信配置
await ch.assertQueue('rpc_pdf_queue', {
  durable: false,
  deadLetterExchange: dlxExchange,
  deadLetterRoutingKey: 'pdf_failure'
});

方案三：Confirm Channel 兜底说明 #

Confirm Channel 的 sendToQueue 回调仅能确认消息是否成功投递到RabbitMQ服务器，无法感知消费者的业务处理结果，所以必须结合上述错误响应或超时机制，不要依赖它来捕获业务错误。

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