RabbitMQ通道关闭重连后Ack/Nack消息报unknown delivery tag错误如何解决

Channel closed by server: 406 (PRECONDITION-FAILED) with message "PRECONDITION_FAILED - unknown delivery tag 80"

根因说明 #

RabbitMQ的delivery tag是通道级的唯一序列号，每个新通道的delivery tag都会从1开始重新计数，规则要求消息只能通过接收它的原通道执行Ack/Nack操作，通道断开重建后，旧通道的delivery tag在新通道中没有任何意义，因此重连后重放缓存的Ack请求必然触发该错误。

现有方案问题 #

你当前使用的Ack请求缓存重放逻辑不符合RabbitMQ的底层规则，跨通道Ack从协议层面就不被支持，属于无效设计。

可行解决方案 #

• 依赖RabbitMQ原生重入队列机制：连接/通道断开时，所有未Ack的消息会被RabbitMQ自动放回原队列头部，后续会被重新投递给新的可用通道的消费者，无需人工尝试跨通道Ack。
• 移除Ack缓存逻辑：该逻辑不仅无法解决问题，还会触发不必要的406错误，直接删除即可。
• 新增通道一致性校验：在消息接收时绑定当前通道的唯一标识，执行Ack前先校验通道是否为接收消息的原通道，不一致直接跳过Ack操作。
• 实现业务幂等性：针对消息自动重投的场景，通过消息ID、业务唯一标识等维度做消费去重，保证重复消费不会产生脏数据或业务异常。

代码修改示例 #

1. 连接方法新增通道唯一标识 #

async connect({ prefetch = 1, queueName }) {
    this.queueName = queueName;
    console.log(`[AMQP][${this.queueName}] | connecting`);
    return queue
        .connect(this.config.rabbitmq.connstring)
        .then(conn => {
            conn.once('error', err => {
                this.channel = null;
                this.currentChannelId = null; // 清空通道ID
                if (err.message !== 'Connection closing') {
                    console.error(
                        `[AMQP][${this.queueName}] (evt:error) | ${err.message}`,
                    );
                }
            });

            conn.once('close', () => {
                this.channel = null;
                this.currentChannelId = null; // 清空通道ID
                console.error(
                    `[AMQP][${this.queueName}] (evt:close) | reconnecting`,
                );
                this.connect({ prefetch, queueName: this.queueName });
            });
            return conn.createChannel();
        })
        .then(ch => {
            console.log(`[AMQP-channel][${this.queueName}] created`);
            ch.on('error', err => {
                console.error(
                    `[AMQP-ch][${this.queueName}] (evt:error) | ${err.message}`,
                );
            });
            ch.on('close', () => {
                console.error(`[AMQP-ch][${this.queueName}] (evt:close)`);
            });
            // 生成通道唯一标识
            this.currentChannelId = Math.random().toString(36).slice(2);
            this.channel = ch;
            return this.channel;
        })
        .then(ch => {
            return this.channel.prefetch(prefetch);
        })
        .then(ch => {
            return this.channel.assertQueue(this.queueName);
        })
        // 移除原缓冲区重放逻辑
        .catch(error => {
            console.error(error);
            console.log(`[AMQP][${this.queueName}] reconnecting in 1s`);
            return this._delay(1000).then(() =>
                this.connect({ prefetch, queueName: this.queueName }),
            );
        });
}

2. 消费消息时绑定通道标识 #

在你的消费回调逻辑中，收到消息时挂载当前通道ID：

this.channel.consume(this.queueName, (msg) => {
  if (!msg) return;
  // 绑定消息接收时的通道ID
  msg.receiveChannelId = this.currentChannelId;
  // 后续你的业务处理逻辑
})

3. 修改Ack逻辑 #

async ack(msg) {
  try {
    // 仅当通道存在且和消息接收通道一致时才执行Ack
    if (this.channel && this.currentChannelId === msg.receiveChannelId) {
      console.log(`[AMQP][${this.queueName}] ack`);
      await this.channel.ack(msg);
    } else {
      console.log(`[AMQP][${this.queueName}] ack skipped: channel has been reset`);
    }
  } catch (e) {
    console.error(`[AMQ][${this.queueName}] ack error: ${e.message}`);
  }
}

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