RabbitMQ高级之失败重试机制(含源码)

一、失败重试机制

    当消费者出现异常后，消息会不断requeue(重新入队）到队列，再重新发送给消费者，然后再次异常，再次requeue，无限循环，导致mq的消息处理飙升，带来不必要的压力。Chh28资讯网——每日最新资讯28at.com

    为此，可利用Spring的retry机制，在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。Chh28资讯网——每日最新资讯28at.com

配置参数如下：Chh28资讯网——每日最新资讯28at.com

listener: # 开启消费者确认其机制      simple:        prefetch: 1  #消费者每次只能获取一条消息，处理完才能获取下一条（可实现能者多劳）        acknowledge-mode: AUTO  # 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

测试结果：Chh28资讯网——每日最新资讯28at.com

图片Chh28资讯网——每日最新资讯28at.com

但是重试三次后，队列里面的消息被踢出了：Chh28资讯网——每日最新资讯28at.com

图片Chh28资讯网——每日最新资讯28at.com

二、失败消息处理策略

    在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处理，它包含三种不同的实现:Chh28资讯网——每日最新资讯28at.com

     1、RejectAndDontRequeueRecoverer:重试耗尽后，直接reject，丢弃消息。默认就是这种方式。Chh28资讯网——每日最新资讯28at.com

    2、ImmediateRequeueMessageRecoverer:重试耗尽后，返回nack，消息重新入队(不建议采用：会出现死循环)。Chh28资讯网——每日最新资讯28at.com

     3、RepublishMessageRecoverer:重试耗尽后，将失败消息投递到指定的交换机。(推荐使用)Chh28资讯网——每日最新资讯28at.com

图片Chh28资讯网——每日最新资讯28at.com

三、实现方式

3.1、定义接收失败消息的交换机、队列及其绑定关系:Chh28资讯网——每日最新资讯28at.com

/**     * 功能描述:定义接收错误消费的日志     * @MethodName: receiveErrorMessage     * @MethodParam: [message]     * @Return: void     * @Author: yyalin     * @CreateDate: 2023/11/15 9:55     */    @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);    }

3.2、定义RepublishMessageRecoverer:

/** * @Description: TODO:定义错误消息接收 * @Author: yyalin * @CreateDate: 2023/11/15 9:58 * @Version: V1.0 */@Configuration@Slf4jpublic class ErrorConfig {    @Bean    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){        log.debug("加载RepublishMessageRecoverer");        return new RepublishMessageRecoverer(rabbitTemplate,"errorExchange","errorRouting");    }}

Chh28资讯网——每日最新资讯28at.com

3.3、测试结果：

图片Chh28资讯网——每日最新资讯28at.com

3.4、总结

消费者如何保证消息一定被消费?Chh28资讯网——每日最新资讯28at.com

•     开启消费者确认机制为auto，由spring确认消息处理成功后返回ack，异常时返回nack。如果一直处理异常会一直重试。
•     开启消费者失败重试机制，并设置MessageRecoverer，多次重试失败后将消息投递到异常交换机，交由人工处理。

本文链接：http://www.28at.com/showinfo-26-34653-0.htmlRabbitMQ高级之失败重试机制(含源码)

声明：本网页内容旨在传播知识，若有侵权等问题请及时与本网联系，我们将在第一时间删除处理。邮件：2376512515@qq.com

上一篇： 双向绑定与单向数据流之争，Solid会取代React吗

下一篇： 新一代WebFlux框架核心技术Reactor响应式编程基本用法