RabbitMQ消息一致性及失败重试_rabbitmq失败消息处理策略

RabbitMQ消息的一致性保证，通常从MQ配置、生产者和消费者3个角度：

1. 生产者：采用confirm消息确认机制及return监听回调机制，确保消息发送成功

2. MQ：将交换机、队列及消息设置持久化，保证rabbitmq宕机后消息不丢失

3. 消费者：手动确认接收消息方式，消息处理失败拒收或重回队列

生产者开启消息确认及return监听回调配置

spring:
  rabbitmq:
    host: ip
    port: 5672
    username: guest
    password: guest
    ##消息发送确认回调
    publisher-confirms: true
    ##采用confirm以及return机制发送返回监听回调
    publisher-confirm-type: correlated
    ##Return机制确保消息从交换机发送到指定的队列
    publisher-returns: true

MQ配置

 
/**
 * RabbitMQ配置信息，绑定交换器、队列、路由键设置
 *
 * <p>
 * 如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，可以将交换机、队列、消息都进行持久化，这样可以保证绝大部分情况下消息不会丢失。
 * 但还是会有小概率发生消息丢失的情况（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），
 * 如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。(transaction/confirm机制)
 *
 * <p>
 * 说明：
 * 1. 队列持久化：需要在声明队列的时候设置durable=true，如果只对队列进行持久化，那么mq重启之后队列里面的消息不会保存
 * 如果需要队列里面的消息也保存下来，那么还需要对消息进行持久化；
 * <p>
 * 2. 消息持久化：设置消息的deliveryMode = 2，消费者重启之后还能够继续消费持久化之后的消息；
 * 使用convertAndSend方式发送消息，消息默认就是持久化的，下面是源码：
 * new MessageProperties() --> DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT --> deliveryMode = 2;
 * <p>
 * 3.重启mq: CMD命令行下执行 net stop RabbitMQ && net start RabbitMQ
 */
@Component
public class RabbitMQConfig {
    private static final String DURABLE_QUEUE_NAME = "durable_queue_name";
    private static final String DURABLE_EXCHANGE_NAME = "durable_exchange_name";
    private static final String ROUTING_KEY = "user.#";
 
    private static final String QUEUE_NAME = "not_durable_queue_name";
    private static final String EXCHANGE_NAME = "not_durable_exchange_name";
 
    @Bean
    public Queue durableQueue() {
//        public Queue(String name) {
//            this(name, true, false, false);
//        }
//        public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete)
        //不指定durable的话默认好像也是true
 
        //public Queue(String name, boolean durable)
        //durable：是否将队列持久化 true表示需要持久化 false表示不需要持久化
        return new Queue(DURABLE_QUEUE_NAME, true);
    }
 
    @Bean
    public TopicExchange durableExchange() {
//        public AbstractExchange(String name) {
//            this(name, true, false);
//        }
//        public AbstractExchange(String name, boolean durable, boolean autoDelete) {
//            this(name, durable, autoDelete, (Map)null);
//        }
        //声明交换机的时候默认也是持久化的
        return new TopicExchange(DURABLE_EXCHANGE_NAME);
    }
 
    @Bean
    public Binding durableBinding() {
        //如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化，一个非持久化，就不允许建立绑定
        return BindingBuilder.bind(durableQueue()).to(durableExchange()).with(ROUTING_KEY);
    }
 
 
    @Bean
    public Queue queue() {
        //public Queue(String name, boolean durable)
        //durable：是否将队列持久化 true表示需要持久化 false表示不需要持久化
        return new Queue(QUEUE_NAME, false);
    }
 
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME,true,false);
    }
 
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with(ROUTING_KEY);
    }
 
}

消费者手动确认消息配置

listener:
      type: simple
      simple:
        #手动接收消息方式，# none：关闭ack；manual：手动ack；auto：自动ack
        acknowledge-mode: manual

失败消息处理策略

一般消息失败后需要重试，我们在yml中配置

listener: # 开启消费者确认其机制
      simple:
        prefetch: 1  #消费者每次只能获取一条消息，处理完才能获取下一条（可实现能者多劳）
        acknowledge-mode: manual # none：关闭ack；manual：手动ack；auto：自动ack
        retry:
          enabled: true  #开启消费者失败重试
          initial-interval: 1000ms  #初始的失败等待时长为1秒
          multiplier: 1 #下次失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 #最大重试次数
          stateless: true #true无状态;false有状态。如果业务中包含事务，这里改为false

    在开启重试模式后，重试次数耗尽后消息消费仍然失败，则需要通过MessageRecoverer接口来处理，它包含三种不同的实现:

     1、RejectAndDontRequeueRecoverer:重试耗尽后，默认方式：直接丢弃消息，不符合多数业务需求

    2、ImmediateRequeueMessageRecoverer:重试耗尽后，返回nack，消息重新入队，容易造成队列死循环消费，不推荐

     3、RepublishMessageRecoverer:重试耗尽后，将失败消息投递到指定的交换机，人工再处理，推荐使用

以下介绍第三种处理方式：

• 定义接收失败消息的交换机、队列及其绑定关系:

    /**
     * 接收错误消费的日志
     */
    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "errorQueue"),
            exchange = @Exchange(name = "errorExchange", type = ExchangeTypes.DIRECT, ignoreDeclarationExceptions = "true"),
            key = "errorRouting"
    ))
    public void receiveErrorMessage(String message) {
        log.info("消费者收到发送错误的消息: " + message);
    }

• 定义RepublishMessageRecoverer

/**
 * 定义错误消息接收
 */
@Configuration
@Slf4j
public class ErrorConfig {
    @Bean
    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){
        log.debug("加载RepublishMessageRecoverer");
        return new RepublishMessageRecoverer(rabbitTemplate,"errorExchange","errorRouting");
    }
}

这个就开启了消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理。