RabbitMQ消费者重试的两种方案

直接重试

优点：

缺点：

保存数据库的重试方案

优点：

缺点：

选择建议

适合直接重试的场景

适合数据库记录的场景

这篇文章总结一下消费者消费重试的方案

直接重试

一种是消息消费失败然后消费者直接重试，这需要配置消费者重试机制

java">@Component
public class DirectRetryConsumer {@RabbitListener(queues = "myQueue")@RabbitListener(queues = "myQueue",containerFactory = "retryContainerFactory")public void processMessage(Message message, Channel channel) {try {// 处理消息processBusinessLogic(message);// 确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 重试处理handleRetry(message, channel, e);}}private void handleRetry(Message message, Channel channel, Exception e) {MessageProperties props = message.getMessageProperties();Long retryCount = props.getHeader("retry-count");if (retryCount == null) {retryCount = 0L;}if (retryCount < maxRetryCount) {// 重新入队，等待重试channel.basicNack(props.getDeliveryTag(), false, true);} else {// 超过重试次数，进入死信队列channel.basicNack(props.getDeliveryTag(), false, false);}}
}@Configuration
public class RetryConfig {@Beanpublic SimpleRabbitListenerContainerFactory retryContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 配置重试策略RetryTemplate retryTemplate = new RetryTemplate();ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();backOffPolicy.setInitialInterval(1000);backOffPolicy.setMultiplier(2.0);backOffPolicy.setMaxInterval(10000);retryTemplate.setBackOffPolicy(backOffPolicy);factory.setRetryTemplate(retryTemplate);return factory;}
}

优点：

实现简单，开发成本低
消息处理实时性高
系统复杂度低
资源消耗相对较少

缺点：

重试策略不够灵活
无法保存失败原因和重试历史
难以进行人工干预
监控和统计困难

保存数据库的重试方案

java">@Entity
public class MessageRecord {@Idprivate Long id;private String messageId;private String payload;private String status; // PENDING, PROCESSING, FAILED, SUCCESSprivate Integer retryCount;private String errorMessage;private LocalDateTime createTime;private LocalDateTime updateTime;
}@Service
@Transactional
public class MessageProcessService {@Autowiredprivate MessageRecordRepository recordRepository;@RabbitListener(queues = "myQueue")public void processMessage(Message message, Channel channel) {MessageRecord record = null;try {// 1. 保存消息记录record = saveMessageRecord(message);// 2. 处理业务逻辑processBusinessLogic(message);// 3. 更新状态为成功record.setStatus("SUCCESS");recordRepository.save(record);// 4. 确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 5. 记录失败信息handleFailure(record, e);// 6. 拒绝消息channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);}}
}@Component
@Slf4j
public class FailedMessageRetryJob {@Autowiredprivate MessageRecordRepository recordRepository;@Scheduled(fixedDelay = 300000) // 5分钟public void retryFailedMessages() {List<MessageRecord> failedRecords = recordRepository.findByStatusAndRetryCountLessThan("FAILED", maxRetryCount);for (MessageRecord record : failedRecords) {try {// 重试处理processMessage(record);// 更新状态record.setStatus("SUCCESS");recordRepository.save(record);} catch (Exception e) {// 更新重试次数和错误信息record.setRetryCount(record.getRetryCount() + 1);record.setErrorMessage(e.getMessage());recordRepository.save(record);log.error("Retry failed for message: {}", record.getMessageId(), e);}}}
}

将所有的消息都记录在数据库中进行保存，并以消息是否消费成功来更改数据库中消息的状态值，可以开启一个定时任务执行从数据库取出失败的消息进行重新消费，因为可能取出消费的时候还可能会失败，可以设置一次任务取出数据消费条数，若是超出条数则等到下次定时任务再进行消费。

当然也可以通过定时任务给另一个重试队列投递消息，然后消费者收到消息就从数据库中取出失败的记录进行重试。这样的方法其实和刚刚讲的原理差不多。