RabbitMQ消息發(fā)送失敗的重試機制核心流程

更新時間：2025年12月26日 09:06:22 作者：Wang's Blog

本文介紹了基于NestJS和RabbitMQ的高可靠消息重試機制設(shè)計,通過三級保障機制（預(yù)持久化、實時回調(diào)處理、定時任務(wù)補償）和靈活的存儲層（支持MySQL/Redis）,確保消息在發(fā)送失敗時能夠可靠重試,感興趣的朋友跟隨小編一起看看吧

核心問題
重試機制核心流程
關(guān)鍵業(yè)務(wù)流程與 RabbitMQ 交互
NestJS 服務(wù)層實現(xiàn)
消息發(fā)送器（TransMessageSender）
定時任務(wù)與重試控制
NestJS 核心代碼實現(xiàn)
工程示例：1
工程示例：2
工程示例：3
關(guān)鍵配置與命令
RabbitMQ 周邊配置處理
核心優(yōu)化點
補充知識點
關(guān)鍵問題解決方案
注意事項
總結(jié)

核心問題

確保消息在RabbitMQ投遞過程中的可靠性，通過持久化存儲、回調(diào)處理和定時重試解決網(wǎng)絡(luò)異常、服務(wù)宕機等場景下的消息丟失風(fēng)險
解決分布式系統(tǒng)中消息發(fā)送失敗時的可靠重試，確保消息不丟失和最終一致性

重試機制核心流程

1 ）消息發(fā)送前持久化

原因：防止消息在發(fā)送過程中因服務(wù)崩潰或網(wǎng)絡(luò)異常丟失。
實現(xiàn)：調(diào)用發(fā)送接口時，先將消息存入數(shù)據(jù)庫，生成唯一ID（如UUID）記錄交換器、路由鍵、消息體及發(fā)送次數(shù)（初始為0）。

2 ）消息發(fā)送后處理

成功發(fā)送：
- RabbitMQ確認接收（ACK）后，立即刪除數(shù)據(jù)庫中的持久化消息，避免數(shù)據(jù)膨脹。
投遞失?。?ul>
交換機不存在：觸發(fā)ConfirmCallback，標記發(fā)送失敗，等待重試。
隊列不存在：觸發(fā)ReturnCallback，重新持久化消息到數(shù)據(jù)庫（原消息已刪除）。

3 ）定時任務(wù)補償

巡檢未成功消息：定時查詢數(shù)據(jù)庫中狀態(tài)為“待發(fā)送”且未超重試次數(shù)的消息。
重試策略：
- 重試次數(shù)+1，重新投遞消息。
- 若超過最大重試次數(shù)（如5次），標記消息為“死亡” 并觸發(fā)告警（郵件/短信）。
并發(fā)控制：多副本部署時需用分布式鎖（如Redis鎖）避免重復(fù)消費。

關(guān)鍵業(yè)務(wù)流程與 RabbitMQ 交互

Confirm 回調(diào)
- 作用：確認 RabbitMQ 是否接收消息。
- 邏輯：
  - 若收到 ACK → 刪除數(shù)據(jù)庫記錄。
  - 若未收到 ACK → 保留記錄等待重試。
Return 回調(diào)
- 場景：消息被 RabbitMQ 接收但無法路由到隊列（如路由鍵錯誤）。
- 處理：將消息重新持久化至數(shù)據(jù)庫，狀態(tài)保持 READY。

NestJS 服務(wù)層實現(xiàn)

1 ) 消息存儲接口設(shè)計（TransMessageService）

// trans-message.interface.ts
export interface TransMessageService {
  saveForSend(
    exchange: string,
    routingKey: string,
    payload: string
  ): Promise<TransMessagePO>;
  deleteOnAck(id: string): Promise<void>;
  saveOnReturn(
    id: string,
    exchange: string,
    routingKey: string,
    payload: string 
  ): Promise<TransMessagePO>;
  listPendingMessages(): Promise<TransMessagePO[]>;
  incrementRetryCount(id: string): Promise<void>;
  markAsDead(id: string): Promise<void>;
}

2 ) 數(shù)據(jù)庫實現(xiàn)（TypeORM + PostgreSQL）

// trans-message.service.ts 
import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { TransMessagePO } from './trans-message.entity';
@Injectable()
export class DBTransMessageService implements TransMessageService {
  constructor(
    @InjectRepository(TransMessagePO)
    private readonly repo: Repository<TransMessagePO>,
    private readonly serviceName: string 
  ) {}
  async saveForSend(exchange: string, routingKey: string, payload: string) {
    const message = new TransMessagePO();
    message.id = uuidv4();
    message.serviceName = this.serviceName;
    message.exchange = exchange;
    message.routingKey = routingKey;
    message.payload = payload;
    message.retryCount = 0;
    message.status = 'PENDING';
    return this.repo.save(message);
  }
  async deleteOnAck(id: string) {
    await this.repo.delete({ id, serviceName: this.serviceName });
  }
  // 其他方法實現(xiàn)類似，略 
}

消息發(fā)送器（TransMessageSender）

1 ) 發(fā)送消息核心邏輯

// trans-message.sender.ts
import { Injectable, Logger } from '@nestjs/common';
import { RabbitMQService } from '@golevelup/nestjs-rabbitmq';
import { TransMessageService } from './trans-message.interface';
@Injectable()
export class TransMessageSender {
  private readonly logger = new Logger(TransMessageSender.name);
  constructor(
    private readonly rmqService: RabbitMQService,
    private readonly messageService: TransMessageService
  ) {}
  async send(exchange: string, routingKey: string, payload: any) {
    const payloadString = JSON.stringify(payload);
    try {
      // 1. 持久化到數(shù)據(jù)庫
      const messagePO = await this.messageService.saveForSend(
        exchange,
        routingKey,
        payloadString
      );
      // 2. 發(fā)送到RabbitMQ
      await this.rmqService.publish(exchange, routingKey, payload, {
        messageId: messagePO.id, // 關(guān)鍵：設(shè)置消息ID
        persistent: true
      });
      this.logger.log(`Message sent: ${messagePO.id}`);
    } catch (e) {
      this.logger.error(`Send failed: ${e.message}`, e.stack);
    }
  }
}

2 ) 回調(diào)處理（Confirm & Return）

// rabbitmq.config.ts
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
@Module({
  imports: [
    RabbitMQModule.forRoot(RabbitMQModule, {
      exchanges: [{ name: 'orders', type: 'topic' }],
      uri: 'amqp://localhost',
      connectionInitOptions: { wait: false },
      // 注冊回調(diào)函數(shù)
      setupController: (channel) => {
        channel.on('return', (msg) => this.handleReturn(msg));
        channel.on('ack', (msg) => this.handleAck(msg));
      }
    })
  ]
})
export class AppModule {
  handleReturn(msg: ConsumeMessage) {
    const { exchange, routingKey } = msg.fields;
    const payload = msg.content.toString();
    const messageId = msg.properties.messageId; // 關(guān)鍵：獲取消息ID
    this.messageService.saveOnReturn(messageId, exchange, routingKey, payload);
  }
  handleAck(msg: ConsumeMessage) {
    const messageId = msg.properties.messageId;
    this.messageService.deleteOnAck(messageId);
  }
}

定時任務(wù)與重試控制

// retry.task.ts
import { Injectable, Logger } from '@nestjs/common';
import { SchedulerRegistry } from '@nestjs/schedule';
@Injectable()
export class RetryTask {
  private readonly logger = new Logger(RetryTask.name);
  private readonly MAX_RETRY = 5;
  constructor(
    private messageService: TransMessageService,
    private scheduler: SchedulerRegistry
  ) {
    this.startRetryCycle();
  }
  startRetryCycle() {
    const interval = setInterval(async () => {
      const messages = await this.messageService.listPendingMessages();
      for (const msg of messages) {
        if (msg.retryCount >= this.MAX_RETRY) {
          await this.messageService.markAsDead(msg.id);
          this.triggerAlert(`Message dead: ${msg.id}`);
          continue;
        }
        await this.messageService.incrementRetryCount(msg.id);
        await this.resendMessage(msg);
      }
    }, 60_000); // 每分鐘執(zhí)行 
    this.scheduler.addInterval('retry-job', interval);
  }
  private async resendMessage(msg: TransMessagePO) {
    // 調(diào)用發(fā)送器重新發(fā)送（略）
  }
}

NestJS 核心代碼實現(xiàn)

消息持久化服務(wù)層（TransMessageService）

// trans-message.service.ts
import { Injectable } from '@nestjs/common';
import { TransMessage } from './entity/trans-message.entity';
import { Repository } from 'typeorm';
import { InjectRepository } from '@nestjs/typeorm';
import { MessageType } from './enums/message-type.enum';
@Injectable()
export class TransMessageService {
  constructor(
    @InjectRepository(TransMessage)
    private readonly messageRepo: Repository<TransMessage>,
    private readonly serviceName: string, // 注入服務(wù)標識
  ) {}
  // 發(fā)送前持久化
  async createReadyMessage(
    exchange: string,
    routingKey: string,
    payload: string,
  ): Promise<TransMessage> {
    const message = this.messageRepo.create({
      id: uuidv4(),
      service: this.serviceName,
      exchange,
      routingKey,
      payload,
      sendDate: new Date(),
      sequence: 0,
      type: MessageType.READY,
    });
    return this.messageRepo.save(message);
  }
  // 發(fā)送成功刪除記錄 
  async markMessageSuccess(id: string): Promise<void> {
    await this.messageRepo.delete({ id, service: this.serviceName });
  }
  // 消息路由失敗后重新持久化
  async recreateReturnedMessage(
    exchange: string,
    routingKey: string,
    payload: string,
  ): Promise<TransMessage> {
    return this.createReadyMessage(exchange, routingKey, payload);
  }
  // 查詢待重試消息
  async listReadyMessages(): Promise<TransMessage[]> {
    return this.messageRepo.find({
      where: { type: MessageType.READY, service: this.serviceName },
    });
  }
  // 遞增重試次數(shù)
  async incrementRetryCount(id: string): Promise<void> {
    const message = await this.messageRepo.findOneBy({ id });
    if (message) {
      message.sequence += 1;
      await this.messageRepo.save(message);
    }
  }
  // 標記消息為死亡狀態(tài)
  async markMessageDead(id: string): Promise<void> {
    const message = await this.messageRepo.findOneBy({ id });
    if (message) {
      message.type = MessageType.DEAD;
      await this.messageRepo.save(message);
    }
  }
}

消息發(fā)送器（TransMessageSender）

// trans-message.sender.ts
import { Injectable, Logger } from '@nestjs/common';
import { RabbitMQService } from '@golevelup/nestjs-rabbitmq';
import { TransMessageService } from './trans-message.service';
@Injectable()
export class TransMessageSender {
  private readonly logger = new Logger(TransMessageSender.name);
  constructor(
    private readonly rmqService: RabbitMQService,
    private readonly transMessageService: TransMessageService,
  ) {}
  async send(exchange: string, routingKey: string, payload: object): Promise<void> {
    try {
      // 1. 序列化消息
      const payloadString = JSON.stringify(payload);
      // 2. 持久化到數(shù)據(jù)庫
      const message = await this.transMessageService.createReadyMessage(
        exchange,
        routingKey,
        payloadString,
      );
      // 3. 發(fā)送至 RabbitMQ
      await this.rmqService.publish(exchange, routingKey, payload, {
        messageId: message.id,
        contentType: 'application/json',
      });
      this.logger.log(`Message sent: ${message.id}`);
    } catch (e) {
      this.logger.error(`Send failed: ${e.message}`, e.stack);
    }
  }
}

RabbitMQ 回調(diào)處理

// rmq.config.ts
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { TransMessageService } from './trans-message.service';
@Module({
  imports: [
    RabbitMQModule.forRoot(RabbitMQModule, {
      exchanges: [{ name: 'orders', type: 'topic' }],
      uri: 'amqp://localhost:5672',
      connectionInitOptions: { wait: false },
      // 注冊回調(diào)處理器 
      setupController: (channel) => {
        // 確認回調(diào) 
        channel.on('ack', (msg) => {
          const messageId = msg.properties.messageId;
          this.transMessageService.markMessageSuccess(messageId);
        });
        // 返回回調(diào)
        channel.on('return', (msg) => {
          const { exchange, routingKey } = msg.fields;
          const payload = msg.content.toString();
          this.transMessageService.recreateReturnedMessage(
            exchange,
            routingKey,
            payload,
          );
        });
      },
    }),
  ],
})
export class RmqConfigModule {}

消息預(yù)持久化機制
- 發(fā)送前必須將消息持久化到數(shù)據(jù)庫，防止發(fā)送過程中因網(wǎng)絡(luò)中斷或服務(wù)崩潰導(dǎo)致消息丟失。業(yè)務(wù)系統(tǒng)調(diào)用發(fā)送接口時：
- 發(fā)送成功后的清理邏輯
當 RabbitMQ 確認（ACK）接收消息后，立即刪除數(shù)據(jù)庫中的持久化記錄，避免無效數(shù)據(jù)堆積引發(fā)存儲壓力。
- 定時任務(wù)重試策略
- 獨立進程定時掃描狀態(tài)為 PENDING 的消息：
- 檢查重試次數(shù)是否超限（如 ≤5次）
每次重試增加 retryCount 值
超限則標記為 DEAD 并觸發(fā)告警

工程示例：1

1 ) 方案1：基礎(chǔ)持久化+重試（推薦）

適用場景：中小型系統(tǒng)
技術(shù)棧：
- @golevelup/nestjs-rabbitmq + TypeORM + PostgreSQL
- 消息表字段：id, serviceName, exchange, routingKey, payload, retryCount, status
優(yōu)勢：實現(xiàn)簡單，依賴少

2 ) 方案2：Redis 高性能存儲

// redis-trans-message.service.ts
import { RedisService } from '@liaoliaots/nestjs-redis';
@Injectable()
export class RedisTransMessageService implements TransMessageService {
  private readonly KEY_PREFIX = 'msg:';
  constructor(private redisService: RedisService) {}
  async saveForSend(exchange: string, routingKey: string, payload: string) {
    const id = uuidv4();
    const client = this.redisService.getClient();
    await client.hset(`${this.KEY_PREFIX}${id}`, {
      exchange,
      routingKey,
      payload,
      retryCount: '0',
      status: 'PENDING'
    });
    return { id } as TransMessagePO;
  }
  // 其他方法通過Redis HSET/HGET/DEL實現(xiàn) 
}

3 ) 方案3：分布式事務(wù)框架集成

技術(shù)棧：NestJS + RabbitMQ + Transactional Outbox 模式
核心邏輯：
- 業(yè)務(wù)操作與消息持久化在同一個數(shù)據(jù)庫事務(wù)中提交
- 獨立進程輪詢Outbox表并投遞消息
優(yōu)勢：強一致性，避免業(yè)務(wù)成功但消息未存儲

工程示例：2

1 ) 方案 1：基礎(chǔ)數(shù)據(jù)庫重試

// task.service.ts 
@Injectable()
export class RetryTaskService {
  constructor(private readonly transMessageService: TransMessageService) {}
  @Cron('*/5 * * * * *') // 每5秒執(zhí)行
  async handleRetry() {
    const messages = await this.transMessageService.listReadyMessages();
    messages.forEach(async (msg) => {
      if (msg.sequence < 5) {
        await this.transMessageService.incrementRetryCount(msg.id);
        // 重新發(fā)送邏輯（略）
      } else {
        await this.transMessageService.markMessageDead(msg.id);
        // 觸發(fā)告警（略）
      }
    });
  }
}

2 ) 方案 2：Redis 高性能暫存

優(yōu)勢：

讀寫速度比 MySQL 快 10 倍以上
支持分布式鎖解決多副本并發(fā)問題

// 使用 Redis 存儲消息
import { Redis } from 'ioredis';
@Injectable()
export class RedisTransMessageService implements TransMessageService {
  private readonly redis = new Redis();
  async createReadyMessage(
    exchange: string,
    routingKey: string,
    payload: string,
  ): Promise<TransMessage> {
    const id = uuidv4();
    await this.redis.hset(
      `msg:${id}`,
      'payload', payload,
      'exchange', exchange,
      'routingKey', routingKey,
      'sequence', '0',
    );
    return { id, exchange, routingKey, payload, sequence: 0 };
  }
}

3 ) 方案 3：死信隊列（DLX）自動重試

RabbitMQ 配置命令：

創(chuàng)建死信交換機和隊列
rabbitmqadmin declare exchange name=dlx type=direct
rabbitmqadmin declare queue name=dead_messages
rabbitmqadmin declare binding source=dlx destination=dead_messages routing_key=dead
主隊列綁定死信路由
rabbitmqadmin declare queue name=orders \
  arguments='{"x-dead-letter-exchange":"dlx", "x-dead-letter-routing-key":"dead"}'

NestJS 消費死信消息：

@RabbitSubscribe({
  exchange: 'dlx',
  routingKey: 'dead',
  queue: 'dead_messages',
})
async handleDeadMessage(msg: any) {
  // 解析原始消息ID并重新持久化
  const originalMsgId = msg.properties.headers['x-original-message-id'];
  await this.transMessageService.recreateReturnedMessage(
    msg.fields.exchange,
    msg.fields.routingKey,
    msg.content.toString(),
  );
}

工程示例：3

1 ) 方案1：數(shù)據(jù)庫存儲方案（TypeORM）

// trans-message.service.ts
@Injectable()
export class TransMessageService {
  constructor(
    @InjectRepository(TransMessage)
    private readonly messageRepo: Repository<TransMessage>
  ) {}
  async prePersist(exchange: string, routingKey: string, payload: string) {
    const message = this.messageRepo.create({
      exchange,
      routingKey,
      payload,
      status: 'PENDING'
    });
    return this.messageRepo.save(message);
  }
  async markAsSuccess(id: string) {
    await this.messageRepo.delete(id);
  }
}

2 ) 方案2：Redis 高性能存儲方案

// redis-trans.service.ts
import { RedisService } from '@liaoliaots/nestjs-redis';
@Injectable()
export class RedisTransService {
  constructor(private readonly redisService: RedisService) {}
  async prePersist(exchange: string, routingKey: string, payload: string) {
    const client = this.redisService.getClient();
    const id = uuidv4();
    await client.hset(
      `msg:${id}`,
      'exchange', exchange,
      'routingKey', routingKey,
      'payload', payload,
      'status', 'PENDING'
    );
    return id;
  }
}

3 ）方案3：混合存儲方案（數(shù)據(jù)庫+Redis）

熱數(shù)據(jù)：高頻重試消息存 Redis
冷數(shù)據(jù)：超限消息轉(zhuǎn)存數(shù)據(jù)庫供審計
一致性保障：通過 Redis 事務(wù)確保狀態(tài)同步

關(guān)鍵配置與命令

RabbitMQ 必要設(shè)置

啟用持久化交換機和隊列
rabbitmqctl set_policy HA ".*" '{"ha-mode":"all"}' --apply-to all
監(jiān)控命令
rabbitmqctl list_queues name messages_ready messages_unacknowledged

NestJS 模塊配置

// app.module.ts
@Module({
  imports: [
    TypeOrmModule.forFeature([TransMessagePO]),
    RabbitMQModule.forRootAsync(RabbitMQModule, {
      useFactory: () => ({
        uri: process.env.RABBITMQ_URI,
        exchanges: [{ name: 'orders', type: 'topic', durable: true }]
      })
    })
  ],
  providers: [
    { provide: TransMessageService, useClass: DBTransMessageService },
    TransMessageSender,
    RetryTask
  ]
})
export class AppModule {}

RabbitMQ 周邊配置處理

1 ) NestJS 連接配置

// rabbitmq.module.ts
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
@Module({
  imports: [
    RabbitMQModule.forRoot(RabbitMQModule, {
      exchanges: [{ name: 'orders', type: 'topic' }],
      uri: 'amqp://user:pass@localhost:5672',
      connectionInitOptions: { wait: false }
    })
  ]
})
export class RabbitModule {}

2 ) 關(guān)鍵 Shell 命令

創(chuàng)建帶持久化的交換機
rabbitmqadmin declare exchange name=orders type=topic durable=true
監(jiān)控未路由消息
rabbitmqctl list_queues name messages_unroutable

3 ) 告警機制實現(xiàn)

// alert.service.ts
@Injectable()
export class AlertService {
  async triggerAlert(messageId: string) {
    // 集成郵件/Slack/Webhook
    await slackService.send(`消息 ${messageId} 重試超限！`);
  }
}

核心優(yōu)化點

消息軌跡追蹤
通過 properties.messageId 實現(xiàn)全鏈路消息關(guān)聯(lián)

并發(fā)控制
使用 Redis 分布式鎖防止多副本重試沖突：

import Redlock from 'redlock';
const lock = await redlock.acquire([`lock:${messageId}`], 5000);

退避策略
指數(shù)級延長重試間隔：delay = Math.min(2 retryCount * 1000, 60000)

關(guān)鍵設(shè)計原則：

所有消息操作必須冪等
持久化存儲與 MQ 狀態(tài)需原子性同步
死信消息必須提供人工干預(yù)接口

補充知識點

RabbitMQ 持久化機制
- 消息需設(shè)置 deliveryMode: 2
- 隊列聲明時添加 durable: true

NestJS 微服務(wù)模式

// main.ts 啟用混合模式
app.connectMicroservice<RabbitMQTransportOptions>({
  transport: Transport.RMQ,
  options: { urls: ['amqp://...'], queue: 'retry_queue' }
});

監(jiān)控指標
- 重試成功率
- 平均重試耗時
- 死信隊列堆積量

關(guān)鍵問題解決方案

并發(fā)重試控制

const lockKey = `lock:msg:${msg.id}`;
const lock = await redis.set(lockKey, '1', 'EX', 30, 'NX');
if (lock) { /* 執(zhí)行重試 */ }

使用 Redis 分布式鎖確保多副本服務(wù)不會重復(fù)處理同一條消息：

消息冪等性設(shè)計

@RabbitSubscribe({ exchange: 'orders', routingKey: 'order.create' })
async handleOrderEvent(msg: any, @Message() amqpMsg) {
  const messageId = amqpMsg.properties.messageId;
  if (await this.redis.exists(`processed:${messageId}`)) return;
  // 處理業(yè)務(wù)...
}