RabbitMq保证消息不丢失

RabbitMQ如何保证消息的可靠性：
1.从生产者到消息队列，congfirm模式（与事务相比confirm模式最大的优势是异步）通过消息确认机制来保证，通过给每个指派唯一标志，完成消费后返回ack确认， 2 消息队列 消息队列的持久化，包括交换机持久化，消息队列持久化以及消息持久化 3.消费者 消费完返回完后手动ack确认，开启RabbitMQ会等待消费者显式发回ack信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息。否则，RabbitMQ会在队列中消息被消费后立即删除它。 若长时间没有ack确认会像下一个消费者发送任务，同时需要做等幂性处理。

RabbitMQ消息丢失的情况

RabbitMQ 消息丢失的源头主要有以下三个：

• 生产者丢失消息
• RabbitMQ 丢失消息
• 消费者丢失消息

生产者丢失消息解决方案

事务消息机制：在生产者发送消息之前，通过 channel.txSelect 开启一个事务，接着发送消息，如果消息没有成功被 RabbitMQ 接收到，生产者会收到异常，此时就可以进行事务回滚 channel.txRollback 然后重新发送；假如 RabbitMQ 收到了这个消息，就可以提交事务channel.txCommit；但是这样一来，生产者的吞吐量和性能都会降低很多，现在一般不这么干。

// 开启事务
channel.txSelect
try {// 这里发送消息
} catch (Exception e) {channel.txRollback
// 这里再次重发这条消息
}
// 提交事务
channel.txCommit

confirm 机制：就是生产端投递的消息一旦投递到RabbitMQ后，RabbitMQ就会发送一个确认消息给生产端，让生产端知道我已经收到消息了，否则这条消息就可能已经丢失了，需要生产端重新发送消息了。

一般采用异步confirm，所以发送消息之前，需要把消息存起来。所以我们需要为消息分配全局唯一的Id，与消息内容一一对应。

生产者也需要监听Broker发送的通知，根据ack / nack 进行确认。

而具体实现时，需要注意：

• 限制重发次数：超限应标志为任务异常，通过人工处理，避免一直重发浪费资源
• 定时扫描未确认消息，因为broker的响应可能会丢失

rabbitmq:# 开启发送确认publisher-confirm-type: correlated

@Service
@Slf4j
public class MqSender {@Autowiredprivate RabbitTemplate rabbitTemplate;//发送秒杀信息public void sendSeckillMessage(String msg){log.info("发送消息：" + msg);rabbitTemplate.convertAndSend("seckillExchange", "seckill.message", msg);}
}

//连接工厂@Resourceprivate CachingConnectionFactory cachingConnectionFactory;
//向spring容器中注入RabbitTemplate对象，并配置开启确认消息回调和消息失败回调@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);/*** 消息确认回调,确认消息是否到达broker* correlationData:消息唯一标识* ack：确认结果,消息是否发送成功* cause：失败原因*/rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {String msgId = correlationData.getId();if (ack) {LOGGER.info("消息发送成功============>" + msgId);//监听消息是否发送成功，如果成功将状态改为1
//                mailLogService.update(new UpdateWrapper().set("status", 1).eq("msgId", msgId));} else {LOGGER.error("消息发送到queue失败============>" + msgId);}});return rabbitTemplate;}

RabbitMQ 丢失消息解决方案

消息持久化

RabbitMQ收到消息后将这个消息暂时存在了内存中，如果RabbitMQ挂了，那重启后数据就丢失了，所以相关的数据应该持久化到硬盘中，需要给exchange、queue和message都进行持久化。

设置持久化，两个步骤：必须都设置。

• 创建 exchange、queue 时，设置为持久化 durable = true：(其实默认是持久化)：会持久化 queue 的元数据，但不会持久化 queue 里的消息
• 因此需要单独设置消息的持久化：发送消息时,将消息Properties的 deliveryMode=PERSISTENT：

无法保障100%不丢失，因为可能持久化完成前就宕机。

（若设置持久化，会在持久化完成后再发出ack）

@Beanpublic Queue queue(){// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效// exclusive:该队列是否只供一个消费者进行消费是否进行消息共享，true可以多个消费者消费，false:只能-一个消费者消费// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。return new Queue(QUEUE);}

//绑定correlationId、以及消息持久化rabbitTemplate.setBeforePublishPostProcessors(new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message, Correlation correlation) {MessageProperties messageProperties = message.getMessageProperties();if (correlation instanceof CorrelationData) {String correlationId = ((CorrelationData) correlation).getId();messageProperties.setCorrelationId(correlationId);}// 持久化处理messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);return message;}@Overridepublic Message postProcessMessage(Message message) throws AmqpException {MessageProperties messageProperties = message.getMessageProperties();messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);return message;}

消息入库

消息入库，顾名思义就是将要发送的消息保存到数据库中。

• 发送消息前先将消息保存到数据库中，有一个状态字段status=0，表示消息发送了但还没收到确认；收到确认后将status设为1，表示RabbitMQ已收到消息。
• 生产端这边还需要开一个定时器，定时检索消息表，将status=0并且超过固定时间后（可能消息刚发出去还没来得及确认，这边定时器刚好检索到这条status=0的消息，所以要设置时间）
• 还没收到确认的消息取出重发 （消息重发可能会造成幂等性问题，这里消费端要做幂等性处理），可能重发还会失败，所以还要添加一个最大重发次数字段retry_count，超过就做另外的处理。
  

消息延迟投递，做二次确认，回调检查

消费者消息丢失

如何保证消费成功（或限流处理）：对应消费端丢数据，即消费完成前，宕机

ACK确认机制

RabbitMQ默认是自动ack的，缺点：

• 消息失败，则重新入队，但入队还是在队列首部，很容易造成死循环(可通过 default-requeue-rejected覆盖)
• 只要有消息，就会源源不断地发送给客户端，而不管客户端能否消费的完（即无法限流）。注意：不要搞混了none和auto，none才是发送了就不管，auto只是自动回发ack。

因此使用手动ack确认

• application.yml配置

spring:rabbitmq:listener:simple:# 消费者最小数量concurrency: 10# 消费者最大数量max-concurrency: 10# 限制消费者每次处理消息的数量(预抓取:等价于缓冲区大小,限流关键)prefetch: 1# manual:手动确认acknowledge-mode: manual

@RabbitListener(queues = "seckillQueue")@RabbitHandlerpublic void receiveDeadMsg(String msg, Channel channel, Message messages) throws Exception {try {log.info("接收消息：" + msg);VoucherOrder voucherOrder = JsonUtil.jsonStr2Object(msg, VoucherOrder.class);voucherOrderService.createVoucherOrder(voucherOrder);// 回发ack，第二个参数multiple，表示是否批量确认channel.basicAck(messages.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 回发Nack，且重回队列为false，防死循环channel.basicNack(messages.getMessageProperties().getDeliveryTag(), false, false);}}