美文网首页
Rabbitmq重试机制

Rabbitmq重试机制

作者: 百恼神烦 | 来源:发表于2021-07-01 17:33 被阅读0次

    设计

    重试数次后,转发到无法处理的队列(后面称之为兜底队列)中,示意图如下:

    消息-->队列1-->失败-->队列1(数次)-->达到上限-->兜底队列
    

    我由于自定义了SimpleRabbitListenerContainerFactory的Bean,因此spring中对于listener.simple.retry的设置统统不起作用,干脆全在这里面实现,也更加客制化。

    要点

    1. 设置好重试的各个参数
      包括但不限于:最大重试次数、初始重试间隔时间、间隔时间增长数、最大间隔时间
      写成代码就这样:
    // SimpleRabbitListenerContainerFactory
    factory.setAdviceChain(org.springframework.amqp.rabbit.config.RetryInterceptorBuilder
                    .stateless()
                    .maxAttempts(5)    // 最大重试次数
                    .backOffOptions(1000, 2, 15000)    // 初始间隔时间、间隔时间增长数、最大间隔时间(单位都是毫秒)
                    .build());
    
    1. 设置上兜底策略
      这个叫法是我起的,大佬们有专业名词麻烦指正下,谢谢。
      其实就是达到重试的上限次数后的操作,我选用设置recoverer进行操作,当达到上限之后、或者被抛出AmqpRejectAndDontRequeueException后,都会触发recoverer的recover方法。而且,AMQP还提供了一个RepublishMessageRecoverer可供使用,它在触发使用后会进行转发,转发的exchange和routekey可以在new的时候定义清楚。
      如下所示
    RepublishMessageRecoverer recoverer = new RepublishMessageRecoverer(rabbitTemplate,
                    getExchange(),
                    getRouteKey());
    factory.setAdviceChain(org.springframework.amqp.rabbit.config.RetryInterceptorBuilder 
                    .recoverer(recoverer)
                    .build());
    

    这样就可以了,但是有两个问题,

    1. 如果兜底队列处理时再出问题怎么办?暂时没好办法,它会不停地被扔兜底队列中,但由于有等待,cpu消耗还是比没有的好很多
    2. 兜底队列该如何处理里面的消息?一种办法是直接打印成特殊的日志消费掉,后面人工查看日志并制定解决方案

    其他

    之前找解决方案时老是找到“死信队列”上,但如你所见,这里并没有用到死信,而且很多博文总是把死信和ttl结合起来,明明死信的来源不止一个……但说回来,即使绑了死信队列,我重试远超上限后仍没有发往死信队列,可能也是我提供了自定义的SimpleRabbitListenerContainerFactory有关,由于时间关系没有尝试,有哪位大佬试过了欢迎分享和指正。

    相关文章

      网友评论

          本文标题:Rabbitmq重试机制

          本文链接:https://www.haomeiwen.com/subject/ylojultx.html