构建基于 Spring Boot 的高可靠 RabbitMQ 消息投递方案
在分布式架构中，确保消息从生产者可靠地送达至消息队列是系统稳健性的核心。以下方案通过引入消息落库记录与异步补偿机制，实现消息投递的闭环保障。
生产者端实现
首先，在项目中引入 spring-boot-starter-amqp 依赖，并在配置中启用发布确认机制。
发布确认与状态追踪
通过实现 ConfirmCallback 接口，可以实时获知消息是否被 Broker 成功接收。结合数据库存储消息投递状态，实现投递链路的闭环监控。
@Component
public class MessageDispatcher {
@Autowired
private RabbitTemplate rabbitTemplate;
public void dispatch(BusinessObject data) {
CorrelationData correlation = new CorrelationData(data.getUniqueId());
rabbitTemplate.setConfirmCallback((cd, ack, cause) -> {
String id = cd.getId();
if (ack) {
// 更新数据库中消息状态为：已送达
updateMessageState(id, MessageStatus.DELIVERED);
} else {
// 记录失败日志以便后续补偿
log.error("消息投递失败，ID: {}, 原因: {}", id, cause);
}
});
rabbitTemplate.convertAndSend("exchange.direct", "routing.key", data, correlation);
}
}定时补偿机制
针对因网络抖动等原因导致的投递失败，需配置定时任务扫描待处理消息，并进行重试。
@Component
public class CompensatoryTask {
@Scheduled(fixedDelay = 15000)
public void retryScheduled() {
List<PendingMessage> pendingList = repository.findStuckMessages();
pendingList.forEach(item -> {
if (item.getRetryCount() < MAX_RETRIES) {
repository.incrementRetryCount(item.getId());
dispatcher.dispatch(item.getPayload());
} else {
repository.markAsFailed(item.getId());
}
});
}
}消费者端实现
消费者端的核心在于手动 ACK，确保消息在业务逻辑完全执行完毕后再从队列中移除。
手动签收处理
在配置文件中设置 spring.rabbitmq.listener.simple.acknowledge-mode=manual，并在业务代码中显式调用应答方法。
@Component
public class MessageProcessor {
@RabbitListener(queues = "order-queue")
public void handle(OrderInfo order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try {
// 执行业务逻辑
processBusinessLogic(order);
// 手动确认消息
channel.basicAck(tag, false);
} catch (Exception e) {
// 拒绝消息并将其重新放回队列，或进行死信处理
channel.basicNack(tag, false, true);
}
}
}