消息丢失场景

MQ消息丢失场景主要有三个：
1.消息生产者，发送消息后，rabbitmq服务器没有收到；导致消息丢失
2.rabbitmq收到消息后，没有持久化保存，导致消息丢失
3.消费者收到消息后，没来得及处理，消费者宕机，导致消息丢失

1、生产者发送消息没有发送到rabbit交换机

解决方案：消息异步确认机制（confirm机制）
在集成springboot项目中；通过配置文件开启confirm机制

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=test
spring.rabbitmq.password=123123
spring.rabbitmq.virtual-host=/test
spring.rabbitmq.connection-timeout=15000#开启 confirm 确认机制
spring.rabbitmq.publisher-confirms=true
#开启 return 确认机制
spring.rabbitmq.publisher-returns=true
#设置为 true 后 消费者在消息没有被路由到合适队列情况下会被return监听，而不会自动删除
spring.rabbitmq.template.mandatory=true

开启confirm机制后，在生产者每次发送消息，都会调用回调代码；开发人员，需要写回调函数的逻辑，处理发送失败的消息

@Component
@Slf4j
public class RabbitMQConfirmAndReturn implements RabbitTemplate.ConfirmCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init() {rabbitTemplate.setConfirmCallback(this);}/*** confirm机制只保证消息到达exchange，不保证消息可以路由到正确的queue* @param correlationData 发送的消息的信息（交换机，路由，消息体等）* @param ack true成功，false失败* @param cause 发生错误的信息*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {// 失败，一般解决方案，是将发送失败消息，存入定时任务队列；尝试重新发送消息；再多次失败，// 就不再发送，转为人工处理if (!ack) {log.error("rabbitmq confirm fail,cause:{}", cause);// ...... 失败处理逻辑}}
}

2、交换机没有发送到队列

解决方案：Return模式，确保消息从交换机发送到队列。
1.开启return模式

#开启 return 机制
spring.rabbitmq.publisher-returns=true

2.开发回调函数代码

@Component
public class Sender implements RabbitTemplate.ReturnCallback {@Autowiredprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init() {rabbitTemplate.setReturnCallback(this);}//通过实现ReturnCallback接口，如果消息从交换器发送到对应队列失败时触发（比如根据发送消息时指定的routingKey找不到队列时会触发）@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("消息主体message: " + message);System.out.println("消息replyCode: " + replyCode);System.out.println("描述: " + replyText);System.out.println("消息使用的交换器exchange: " + exchange);System.out.println("消息使用的路由键routing: " + routingKey);}
}

3、交换机、队列、消息没有设置持久化

交换机、队列、消息没有持久化，当rabbitmq的服务重启之后，这些信息就会丢失。

交换机持久化
在声明交换机的时候，设置持久化属性

	/*** 构造参数说明：* 参数1：交换机名称* 参数2：durable：true表示持久化，false表示不持久化* 参数3：autoDelete：true自动删除，false不自动删除*/@Beanpublic TopicExchange exchange() {return new TopicExchange("exchangeName", true, false);}

队列持久化
在声明队列的时候，设置持久化属性

    public Queue queue() {/*** @param queueName 队列名称* @param durable 队列持久化，true持久化，false不持久化* @param exclusive 是否排他， true不排他，false排他；此处配置一般false* @param autoDelete 是否自动删除，无生产者，队列自动删除* @param args 队列参数*/return new Queue("queueName", true, false, false, args);}

消息持久化

消息的持久化是默认持久的。无需配置

4、消费者接收到消息没有执行业务逻辑，导致消息丢失

解决方案：手动确认消息机制
配置文件配置

spring.rabbitmq.listener.simple.acknowledge-mode=manual

spring:rabbitmq:host: localhostport: 5672username: rootpassword: 123123virtual-host: /testpublisher-confirms: true   # 开启发送确认publisher-returns: true  # 开启发送失败回退#开启acklistener:direct:acknowledge-mode: manualsimple:acknowledge-mode: manual #采取手动应答#concurrency: 1 # 指定最小的消费者数量#max-concurrency: 1 #指定最大的消费者数量retry:enabled: true # 是否支持重试

消费者代码

@Component
public class Consumer {@RabbitHandlerpublic void consumeMsg(String msg, Channel channel, Message message) throws IOException {//拿到消息延迟消费try {// .... 消费消息业务逻辑/*** deliveryTag	消息的随机标签信息* multiple	是否批量；true表示一次性的将小于deliveryTag的值进行ack*/channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (InterruptedException e) {e.printStackTrace();/*** deliveryTag	消息的随机标签信息* multiple	是否批量；true表示一次性的将小于deliveryTag的值进行ack* requeue	被拒绝的消息是否重新入队列*/channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);}}
}

当业务出现意料之外的一场；消息就会重新回到队列中；会分发到其他正常consumer中进行消费