项目实战之RabbitMQ重试机制进行消息补偿通知-Toy模板网

这篇具有很好参考价值的文章主要介绍了项目实战之RabbitMQ重试机制进行消息补偿通知。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

🧑‍💻作者名称：DaenCode
🎤作者简介：啥技术都喜欢捣鼓捣鼓，喜欢分享技术、经验、生活。
😎人生感悟：尝尽人生百味，方知世间冷暖。

项目实战之RabbitMQ重试机制进行消息补偿通知,项目实战经验,rabbitmq,分布式

🌟架构图

项目实战之RabbitMQ重试机制进行消息补偿通知,项目实战经验,rabbitmq,分布式

🌟application.properties

##----------rabbit配置--------------
spring.rabbitmq.host=49.233.48.98
spring.rabbitmq.port=5672
#需要手工创建虚拟主机
spring.rabbitmq.virtual-host=dev
spring.rabbitmq.username=admin
spring.rabbitmq.password=password
#消息确认方式，manual(手动ack) 和auto(自动ack); 消息消费重试到达指定次数进到异常交换机和异常队列，需要改为自动ack确认消息
spring.rabbitmq.listener.simple.acknowledge-mode=auto
#开启重试，消费者代码不能try catch捕获异常，要抛出异常
spring.rabbitmq.listener.simple.retry.enabled=true
#最大重试次数
spring.rabbitmq.listener.simple.retry.max-attempts=4
# 重试消息的时间间隔，5秒
spring.rabbitmq.listener.simple.retry.initial-interval=5000

🌟异常MQ配置

@Configuration
@Slf4j
public class ErrorRabbitMQConfig {
    /**
     * 异常交换机
     */
    private String errorExchange="error.exchange";
    /**
     * 异常队列
     */
    private String errorQueue="error.queue";

    private String errorRoutingKey="error.routing.key";

    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 创建异常交换机
     * @return
     */
    @Bean
    public TopicExchange errorTopicExchange(){
        return new TopicExchange(errorExchange,true,false);
    }

    /**
     * 创建异常队列
     * @return
     */
    @Bean
    public Queue errorQueue(){
        return new Queue(errorQueue,true);
    }

    /**
     * 建立绑定关系
     * @return
     */
    @Bean
    public Binding bindingErrorQueueAndExchange(){
        return BindingBuilder.bind(errorQueue()).to(errorExchange()).with(errorRoutingKey);
    }
    /**
    *消息重新发布到异常MQ
    */
    @Bean
    public MessageRecoverer messageRecoverer(){
        return new RepublishMessageRecoverer(rabbitTemplate,errorExchange,errorRoutingKey);
    }
}

🌟异常MQ消费者

@Component
@Slf4j
@RabbitListener(queuesToDeclare = { @Queue("error.queue") })
public class ErrorMQListener {
    @RabbitHandler
    private void errorMessageHandler(EventMessage eventMessage, Message message, Channel channel) throws IOException {
        log.info("监听到消息：{}",message);
        //进行消息消费
        //发送邮箱通知、告警服务
    }
}