五分钟带你玩转rabbitmq(四)延时重试队列

举报
小鲍侃java 发表于 2021/09/11 01:35:53 2021/09/11
【摘要】 如果只是网络抖动 出现异常那么直接进入死信队列 那么是不合理的 这就可以使用延时重试队列 原理: 1.发送到业务队里 如果正常收到 正常运行 2.如果处理失败 重试  并投入延时队列 如果超过延时时间 重新投入业务队列 3.如果重试次数大于3 那么进入死信队列 1.业务队列 @Configurationpub...

如果只是网络抖动 出现异常那么直接进入死信队列 那么是不合理的

这就可以使用延时重试队列

原理:

1.发送到业务队里 如果正常收到 正常运行

2.如果处理失败 重试  并投入延时队列 如果超过延时时间 重新投入业务队列

3.如果重试次数大于3 那么进入死信队列

1.业务队列


      @Configuration
      public class BusinessConfig {
         /**
       * yewu1模块direct交换机的名字
       */
         public static final String YEWU1_EXCHANGE = "yewu1_direct_exchange";
         /**
       * demo业务的队列名称
       */
         public static final String YEWU1_DEMO_QUEUE = "yewu1_demo_queue";
         /**
       * demo业务的routekey
       */
         public static final String YEWU1_DEMO_ROUTINGKEY = "yewu1_demo_key";
         /**
       * 业务交换机交换机(一个项目一个业务交换机即可)
       * 1.定义direct exchange,绑定queueTest
       * 2.durable="true" rabbitmq重启的时候不需要创建新的交换机
       * 3.direct交换器相对来说比较简单,匹配规则为:如果路由键匹配,消息就被投送到相关的队列
       * fanout交换器中没有路由键的概念,他会把消息发送到所有绑定在此交换器上面的队列中。
       * topic交换器你采用模糊匹配路由键的原则进行转发消息到队列中
       */
         @Bean
         public DirectExchange yewu1Exchange() {
              DirectExchange directExchange = new DirectExchange(YEWU1_EXCHANGE, true, false);
             return directExchange;
          }
         /**
       * 新建队列(一个业务需要一个队列一个routekey 命名格式 项目名-业务名)
       * 1.队列名称
       * 2.durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
       * 3.exclusive 表示该消息队列是否只在当前connection生效,默认是false
       * 4.auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
       * 5.对nack或者发送超时的 发送给死信队列 args是绑定死信队列
       *
       */
         @Bean
         public Queue yewu1DemoQueue() {
             return new Queue(YEWU1_DEMO_QUEUE, true, false, false);
          }
         /**
       * 交换机与routekey绑定
       *
       * @return
       */
         @Bean
         public Binding yewu1DemoBinding() {
             return BindingBuilder.bind(yewu1DemoQueue()).to(yewu1Exchange())
                  .with(YEWU1_DEMO_ROUTINGKEY);
          }
      }
  
 

2.延时队列


      @Configuration
      public class RetryConfig {
         /**
       * 延时队列 交换机配置标识符(固定)
       */
         public static final String RETRY_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
         /**
       * 延时队列交换机绑定配置键标识符(固定)
       */
         public static final String RETRY_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";
         /**
       * 延时队列消息的配置超时时间枚举(固定)
       */
         public static final String RETRY_MESSAGE_TTL = "x-message-ttl";
         /**
       * yewu1模块延时队列交换机
       */
         public final static String YEWU1_RETRY_EXCHANGE_NAME = "yewu1_retry_exchange";
         /**
       * yewu1模块DEMO业务延时队列
       */
         public final static String YEWU1_DEMO_RETRY_QUEUE_NAME = "yewu1_demo_retry_queue";
         /**
       * yewu1模块DEMO延时队列routekey
       */
         public final static String YEWU1_DEMO_RETRY_ROUTING_KEY = "yewu1_demo_retry_key";
         /**
       * 延时队列交换机
       *
       * @return
       */
          @Bean
         public DirectExchange yewu1RetryExchange() {
              DirectExchange directExchange = new DirectExchange(YEWU1_RETRY_EXCHANGE_NAME, true, false);
             return directExchange;
          }
         /**
       * 新建延时队列 一个业务队列需要一个延时队列
       *
       * @return
       */
          @Bean
         public Queue yewu1DemoRetryQueue() {
              Map<String, Object> args = new ConcurrentHashMap<>(3);
             // 将消息重新投递到业务交换机Exchange中
              args.put(RETRY_LETTER_QUEUE_KEY, BusinessConfig.YEWU1_EXCHANGE);
              args.put(RETRY_LETTER_ROUTING_KEY, BusinessConfig.YEWU1_DEMO_ROUTINGKEY);
             // 消息在队列中延迟3s后超时,消息会重新投递到x-dead-letter-exchage对应的队列中,routingkey为自己指定
              args.put(RETRY_MESSAGE_TTL, 3 * 1000);
             return new Queue(YEWU1_DEMO_RETRY_QUEUE_NAME, true, false, false, args);
          }
         /**
       * 绑定以上定义关系
       *
       * @return
       */
          @Bean
         public Binding retryDirectBinding() {
             return BindingBuilder.bind(yewu1DemoRetryQueue()).to(yewu1RetryExchange())
                  .with(YEWU1_DEMO_RETRY_ROUTING_KEY);
          }
      }
  
 

3.死信队列


      @Configuration
      public class DeadConfig {
         /**
       * 死信队列
       */
         public final static String FAIL_QUEUE_NAME = "fail_queue";
         /**
       * 死信交换机
       */
         public final static String FAIL_EXCHANGE_NAME = "fail_exchange";
         /**
       * 死信routing
       */
         public final static String FAIL_ROUTING_KEY = "fail_routing";
         /**
       * 创建配置死信队列
       *
       */
          @Bean
         public Queue deadQueue() {
             return new Queue(FAIL_QUEUE_NAME, true, false, false);
          }
         /**
       * 死信交换机
       *
       * @return
       */
          @Bean
         public DirectExchange deadExchange() {
              DirectExchange directExchange = new DirectExchange(FAIL_EXCHANGE_NAME, true, false);
             return directExchange;
          }
         /**
       * 绑定关系
       *
       * @return
       */
          @Bean
         public Binding failBinding() {
             return BindingBuilder.bind(deadQueue()).to(deadExchange()).with(FAIL_ROUTING_KEY);
          }
      }
  
 

4.生产者


      @RestController
      @RequestMapping("/TestRabbit")
      public class ProducerDemo {
         @Resource
         private RabbitTemplate rabbitTemplate;
         //@RequestMapping("/sendDirect")
          String sendDirect(@RequestBody String message) throws Exception {
              System.out.println("开始生产");
              CorrelationData data = new CorrelationData(UUID.randomUUID().toString());
              rabbitTemplate.convertAndSend(BusinessConfig.YEWU1_EXCHANGE, BusinessConfig.YEWU1_DEMO_ROUTINGKEY,
                  message, data);
              System.out.println("结束生产");
              System.out.println("发送id:" + data);
             return "OK,sendDirect:" + message;
          }
      }
  
 

5.消费者


      public enum RabbitEnum {
         /**
       * 处理成功
       */
          ACCEPT,
         /**
       * 可以重试的错误
       */
          RETRY,
         /**
       * 无需重试的错误
       */
          REJECT
  
 

      @Component
      public class ConsumerDemo {
         private final static Logger logger = LoggerFactory.getLogger(ConsumerDemo.class);
          @Resource
         private RabbitTemplate rabbitTemplate;
          // @RabbitListener(queues = "yewu1_demo_queue")
         protected void consumer(Message message, Channel channel) throws Exception {
              RabbitEnum ackSign = RabbitEnum.RETRY;
              System.out.println(message.getMessageProperties().getCorrelationId());
             try {
                 // 可以加入重复消费判断
                 int i = 1 / 0;
              } catch (Exception e) {
                  ackSign = RabbitEnum.RETRY;
                 throw e;
              } finally {
                 // 通过finally块来保证Ack/Nack会且只会执行一次
                 if (ackSign == RabbitEnum.ACCEPT) {
                      channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                  } else if (ackSign == RabbitEnum.RETRY) {
                     String correlationData =
                          (String)message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
                      System.out.println(message.getMessageProperties().getCorrelationId());
                      long retryCount = getRetryCount(message.getMessageProperties());
                     if (retryCount >= 3) {
                         // 重试次数超过3次,则将消息发送到失败队列等待特定消费者处理或者人工处理
                         try {
                              channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                              rabbitTemplate.convertAndSend(DeadConfig.FAIL_EXCHANGE_NAME, DeadConfig.FAIL_ROUTING_KEY,
                                  message, new CorrelationData(correlationData));
                              logger.info("连续失败三次,将消息发送到死信队列,发送消息:" + new String(message.getBody()));
                          } catch (Exception e1) {
                              channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                              logger.error("发送死信队列报错:" + e1.getMessage() + ",原始消息:" + new String(message.getBody()));
                          }
                      } else {
                         try {
                              channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                             // 重试次数不超过3次,则将消息发送到重试队列等待重新被消费
                              rabbitTemplate.convertAndSend(RetryConfig.YEWU1_RETRY_EXCHANGE_NAME,
                                  RetryConfig.YEWU1_DEMO_RETRY_ROUTING_KEY, message,
                                 new CorrelationData(correlationData));
                              logger.info("消费失败,消息发送到重试队列;" + "原始消息:" + new String(message.getBody()) + ";第"
                                  + (retryCount + 1) + "次重试");
                          } catch (Exception e1) {
                              channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                              logger.error("消息发送到重试队列的时候,异常了:" + e1.getMessage() + ",重新发送消息");
                          }
                      }
                  }
              }
          }
         /**
       * 获取消息被重试的次数
       */
         public long getRetryCount(MessageProperties messageProperties) {
              Long retryCount = 0L;
             if (null != messageProperties) {
                 List<Map<String, ?>> deaths = messageProperties.getXDeathHeader();
                 if (deaths != null && deaths.size() > 0) {
                      Map<String, Object> death = (Map<String, Object>)deaths.get(0);
                      retryCount = (Long)death.get("count");
                  }
              }
             return retryCount;
          }
      }
  
 

参考:https://www.cnblogs.com/mfrank/p/11260355.html

          https://d.wps.cn/v/8niSQ

文章来源: baocl.blog.csdn.net,作者:小黄鸡1992,版权归原作者所有,如需转载,请联系作者。

原文链接:baocl.blog.csdn.net/article/details/106214436

【版权声明】本文为华为云社区用户转载文章,如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。