要想保住RabbitMQ消息不丢失，需要从下面几个方面进行完善。
要想做到消息持久化，必须满足以下几点：
new DirectExchange(exchangeName, true, false, new HashMap<String, Object>());
new Queue(name, true, false, false, new HashMap<String, Object>());
发送消息设置发送模式deliveryMode=2代表持久化消息
org.springframework.amqp.rabbit.core.RabbitTemplate默认情况下发送模式为deliveryMode=2
org.springframework.amqp.core.MessageProperties的默认发送模式：
ConfirmCallback
通过实现 ConfirmCallback 接口，消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中。
@Component public class RabbitTemplateConfig implements RabbitTemplate.ConfirmCallback{ @Autowired private RabbitTemplate rabbitTemplate; @PostConstruct public void init(){ rabbitTemplate.setConfirmCallback(this); //指定 ConfirmCallback } @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println("消息唯一标识："+correlationData); System.out.println("确认结果："+ack); System.out.println("失败原因："+cause); }
还需要在配置文件添加配置：
spring: rabbitmq: publisher-confirms: true
ReturnCallback
通过实现 ReturnCallback 接口，启动消息失败返回，比如路由不到队列时触发回调
@Component public class RabbitTemplateConfig implements RabbitTemplate.ReturnCallback{ @Autowired private RabbitTemplate rabbitTemplate; @PostConstruct public void init(){ rabbitTemplate.setReturnCallback(this); //指定 ReturnCallback } @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { System.out.println("消息主体 message : "+message); System.out.println("消息主体 message : "+replyCode); System.out.println("描述："+replyText); System.out.println("消息使用的交换器 exchange : "+exchange); System.out.println("消息使用的路由键 routing : "+routingKey); } }
还需要在配置文件添加配置：
spring: rabbitmq: publisher-returns: true
template:
mandatory: true
1、手动确认消息
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual
@Bean public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){ SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setMessageConverter(new Jackson2JsonMessageConverter()); factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); //开启手动 ack return factory; }
@RabbitHandler public void processMessage2(String message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long tag) { System.out.println(message); try { channel.basicAck(tag,false); // 确认消息 } catch (IOException e) { e.printStackTrace(); } }
2、手动否认、拒绝消息
@RabbitHandler public void processMessage2(String message, Channel channel,@Headers Map<String,Object> map) { System.out.println(message); if (map.get("error")!= null){ System.out.println("错误的消息"); try { channel.basicNack((Long)map.get(AmqpHeaders.DELIVERY_TAG),false,true); //否认消息 return; } catch (IOException e) { e.printStackTrace(); } } try { channel.basicAck((Long)map.get(AmqpHeaders.DELIVERY_TAG),false); //确认消息 } catch (IOException e) { e.printStackTrace(); }
hello
错误的消息
hello
错误的消息
hello
错误的消息
hello
错误的消息
channel.basicReject((Long)map.get(AmqpHeaders.DELIVERY_TAG),false); //拒绝消息
3、手动确认、拒绝消息
@Bean public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setQueueNames("consumer_queue"); // 监听的队列 container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 手动确认 container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> { //消息处理 System.out.println("====接收到消息====="); System.out.println(new String(message.getBody())); if(message.getMessageProperties().getHeaders().get("error") == null){ channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); System.out.println("消息已经确认"); }else { //channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false); channel.basicReject(message.getMessageProperties().getDeliveryTag(),false); System.out.println("消息拒绝"); } }); return container; }
如果消息成功被消费（成功的意思是在消费的过程中没有抛出异常），则自动确认
当抛出 AmqpRejectAndDontRequeueException 异常的时候，则消息会被拒绝，且 requeue = false（不重新入队列）
当抛出 ImmediateAcknowledgeAmqpException 异常，则消费者会被确认
其他的异常，则消息会被拒绝，且 requeue = true（如果此时只有一个消费者监听该队列，则有发生死循环的风险，多消费端也会造成资源的极大浪费，这个在开发过程中一定要避免的）。可以通过 setDefaultRequeueRejected（默认是true）去设置
@Bean public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setQueueNames("consumer_queue"); // 监听的队列 container.setAcknowledgeMode(AcknowledgeMode.AUTO); // 根据情况确认消息 container.setMessageListener((MessageListener) (message) -> { System.out.println("====接收到消息====="); System.out.println(new String(message.getBody())); //抛出NullPointerException异常则重新入队列 //throw new NullPointerException("消息消费失败"); //当抛出的异常是AmqpRejectAndDontRequeueException异常的时候，则消息会被拒绝，且requeue=false //throw new AmqpRejectAndDontRequeueException("消息消费失败"); //当抛出ImmediateAcknowledgeAmqpException异常，则消费者会被确认 throw new ImmediateAcknowledgeAmqpException("消息消费失败"); }); return container; }
引用：
https://www.jianshu.com/p/19e0927315da
https://www.jianshu.com/p/2c5eebfd0e95
原文：https://www.cnblogs.com/caoweixiong/p/12917666.html