企业官网建设流程全解析

RabbitMQ 保证消息不丢失需要从‌生产者、Broker、消费者‌三个核心环节同时配置，缺一不可，核心是开启持久化、生产者确认和消费者手动ACK机制。

一，RabbitMQ如何保证消息不丢失

一、生产者端：确保消息成功送达Broker

‌开启生产者确认机制‌：发送消息后等待Broker返回ACK确认，收到NACK或超时未回调时自动重试。
‌设置消息回退回调‌：当消息无法路由到队列时触发通知，执行补偿处理避免静默丢失。

二、Broker端：确保消息持久化存储不丢失

三重持久化配置‌：交换机设置durable=true、队列设置durable=true、消息设置deliveryMode=2，将消息写入磁盘。
‌高可用部署‌：使用镜像队列或Quorum队列，将消息同步到多节点，避免单节点故障导致永久丢失。

三、消费者端：确保消息处理完成再确认

‌关闭自动ACK‌：改为手动确认模式，业务逻辑处理成功后再调用basicAck发送确认信号。
‌失败异常处理‌：处理失败时调用basicNack将消息重新入队，或转入死信队列单独处理，避免消息直接丢弃。

RabbitMQ消息不丢失的三个核心环节（生产者确认、Broker持久化、消费者手动ACK），

二，基于 ‌Spring Boot‌ 环境的完整配置代码与实现方案。

一、application.yml 核心配置

这是实现消息可靠性的基础，必须开启生产者确认和消费者手动ACK。

spring:rabbitmq:host:localhostport:5672username:guestpassword:guest# 1. 生产者确认机制：correlated 表示异步回调确认publisher-confirm-type:correlated# 2. 消息回退机制：当消息无法路由到队列时触发publisher-returns:truelistener:simple:# 3. 消费者手动ACK模式acknowledge-mode:manual# 4. 消费失败重试策略（可选）retry:enabled:truemax-attempts:3

二、生产者端：发送确认与回退处理

通过实现 RabbitTemplate.ConfirmCallback 和 ReturnCallback 接口，确保消息成功到达交换机并路由到队列。

importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.amqp.support.CorrelationData;importorg.springframework.stereotype.Component;importjavax.annotation.PostConstruct;importjavax.annotation.Resource;@ComponentpublicclassReliableProducer{@ResourceprivateRabbitTemplaterabbitTemplate;@PostConstructpublicvoidinit(){// 设置确认回调rabbitTemplate.setConfirmCallback((correlation,ack,cause)->{if(ack){System.out.println("消息成功送达Broker: "+correlation);}else{System.err.println("消息送达Broker失败，原因: "+cause);// 执行重试或记录日志}});// 设置回退回调（仅当消息无法路由到队列时触发）rabbitTemplate.setReturnsCallback(returned->{System.err.println("消息路由失败，退回消息: "+returned.getMessage());// 执行补偿逻辑，如存入数据库或死信队列});}publicvoidsendMessage(Stringexchange,StringroutingKey,Objectmessage){rabbitTemplate.convertAndSend(exchange,routingKey,message);}}

三、消费者端：手动ACK与异常处理

在监听器中通过 Channel 手动发送ACK或NACK，确保业务逻辑执行成功后才确认消息

importcom.rabbitmq.client.Channel;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;importjava.io.IOException;@ComponentpublicclassReliableConsumer{@RabbitListener(queues="my_queue")publicvoidhandleMessage(Messagemessage,Channelchannel)throwsIOException{longdeliveryTag=message.getMessageProperties().getDeliveryTag();try{// 1. 执行业务逻辑System.out.println("收到消息: "+newString(message.getBody()));// 模拟业务处理...// 2. 业务成功，手动ACKchannel.basicAck(deliveryTag,false);}catch(Exceptione){// 3. 业务失败，手动NACK并重新入队（requeue=true）// 注意：若一直失败会导致死循环，建议配合重试次数或转入死信队列channel.basicNack(deliveryTag,false,true);System.err.println("消息处理失败，重新入队: "+e.getMessage());}}}

四、Broker端：队列与消息持久化配置

在创建队列和发送消息时，必须显式声明持久化属性。

importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassRabbitConfig{// 1. 定义持久化交换机@BeanpublicDirectExchangedirectExchange(){returnnewDirectExchange("my_exchange",true,false);}// 2. 定义持久化队列@BeanpublicQueuequeue(){returnnewQueue("my_queue",true);// durable=true}// 3. 绑定关系@BeanpublicBindingbinding(DirectExchangeexchange,Queuequeue){returnBindingBuilder.bind(queue).to(exchange).with("my_routing_key");}// 4. 发送消息时设置持久化模式 (Spring Boot 默认即为 PERSISTENT)// 若需自定义，可在 convertAndSend 时传入 MessagePostProcessor}

五、关键注意事项

性能权衡‌：开启生产者确认和持久化会略微降低吞吐量，但在金融、订单等核心场景中是必须的。
死信队列‌：对于多次重试仍失败的消息，应配置 TTL 和死信交换机（DLX），避免阻塞正常业务。
幂等性‌：由于网络抖动可能导致消息重复投递，消费者端必须结合之前讨论的‌幂等性设计‌（如Redis去重或数据库唯一索引）来处理重复消息。