RabbitMQ消息确认机制

Rabbitmq消息丢失主要涵盖三种情况：

（1）生产者传递到MQ过程中消息丢失
（2）MQ中消息丢失
（3）MQ传递到消费者过程中消息丢失

生产者传递到MQ过程中消息丢失

RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。消息发送到MQ以后，会返回一个结果给发送者，表示消息是否处理成功。结果有两种请求：
publisher-confirm：发送者确认
（1）消息成功投递到交换机，返回ack
（2）消息未投递到交换机，返回nack
publisher-return：发送者回执
（1）消息投递到交换机了，但是没有路由到队列。返回ACK，及路由失败原因。
确认机制发送消息时，需要给每个消息设置一个全局唯一id，以区分不同消息，避免ack冲突。
步骤：
（1）yml中配置

rabbitmq:host: xxxxx
    port:5672username: xxx #用户名password: xxx #密码publisher-confirm-type: correlated #开启消息确认机制publisher-returns:true#开启消息失败机制template:mandatory:true#必须设置成true 消息路由失败通知监听者，而不是将消息丢弃

（2）在项目启动过程中配置

@Slf4j@ConfigurationpublicclassCommonConfigimplementsApplicationContextAware{@OverridepublicvoidsetApplicationContext(ApplicationContext applicationContext)throwsBeansException{RabbitTemplate rabbitTemplate= applicationContext.getBean(RabbitTemplate.class);
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey)->{
            log.info(String.valueOf(replyCode), replyText, exchange, routingKey, message.toString());});}}

（3）发送消息，指定消息ID、消息ConfirmCallback

publicstaticvoidsendDirectMessage(RabbitMQExchanges rabbitMQExchanges,RabbitMQRoutingKeys rabbitMQRoutingKeys,Object content){//消息id封装到CorrelationData中CorrelationData correlationId =newCorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.setMandatory(true);
        correlationId.getFuture().addCallback(result ->{if(result.isAck()){
                log.info("消息发送成功",correlationId.getId());}else{
                log.error("消息发送失败",correlationId.getId(),result.getReason());}},ex -> log.error("消息发送失败",correlationId.getId(),ex.getMessage()));
        rabbitTemplate.convertAndSend(rabbitMQExchanges.getExchange(), rabbitMQRoutingKeys.getRoutingKey(), content, correlationId);}

MQ中消息丢失

MQ默认是内存存储消息，开启持久化功能可以确保缓存在MQ中的消息不丢失。设置持久化有三个步骤：交换机持久化、队列持久化、消息持久化

MQ传递到消费者过程中消息丢失

RabbitMQ支持消费者确认机制，即：消费者处理消息后可以向MQ发送ack回执，MQ收到ack回执后才会删除该消息。而SpringAMQP则允许配置三种确认模式：
manual：手动ack，需要在业务代码结束后，调用api发送ack。
auto：自动ack，由spring监测listener代码是否出现异常，没有异常则返回ack；抛出异常则返回nack
none：关闭ack，MQ假定消费者获取消息后会成功处理，因此消息投递后立即被删除

listener:simple:prefetch:1#消费者每次只能获取一条消息，才处理完才能获取下一条acknowledge-mode: AUTO  #1、NONE：全自动 2、AUTO：半自动 3、MANUAL:手动 开启消费端消息确认机制

当消费者出现异常后，消息会不断requeue队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，给服务器带来压力。
yml中可配置如下：

listener:simple:prefetch:1#消费者每次只能获取一条消息，才处理完才能获取下一条acknowledge-mode: AUTO  #1、NONE：全自动 2、AUTO：半自动 3、MANUAL:手动 开启消费端消息确认机制retry:enabled:true#开启超时重试机制initial-interval: 1000ms  #失败后的初始等待时间multiplier:1#失败后下次的等待时长倍数，下次等待时长= initial-interval * multipliermax-attempts:3#最大重试次数

其次配置失败消息处理策略。重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处理，它包含三种不同的实现：
RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式
ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机

来源：B站黑马