Rabbitmq业务流程包含容错排查

流程是这样的，订阅者，发送消息到test交换机，通过route key 分发到绑定的队列，这里涉及到交换机的类型，可以看我上一篇文章。如果没有匹配到这个routeKey就默认发送到AE交换机(fanout模式)，这个交换机要设置internal:true意为内部交换机 。AE交换机再把错误的消息，发送到其绑定的队列中，如果test交换机，发送消息被匹配到的队里中，而处理该队列的订阅者，拒绝了或者超时了处理，test交换机就将该消息发送到就死信交换机，然后到死信队列中

一、 进入死信队列(进入死信的三种方式)

• 1.消息被拒绝（basic.reject or basic.nack）并且requeue=false
• 2.消息TTL过期
• 3.队列达到最大长度

代码演示

      - channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); // 拒绝消息
          - true   发送给下一个消费者
          - false  谁都不接受，从队列中删除
  
 

Rabbit设置

• 1.设置AE交换机 设置为内部交换机，模式为fanout
  当发送到正常交换机消息，没有被匹配到route key的消息对进到改交换机

      FanoutExchange fanoutExchange=new FanoutExchange("alter");
      fanoutExchange.setInternal(true);//设置为内部交换机，作为处理了非法的消息,无法匹配到route key的消息
  
 

- 为AE交换机绑定队列 `alter_message`

 

• 2.设置处理正常的交换机 test

绑定参数,设置没有匹配 route key 的消息发送到AE交换机 alternate-exchange

• 3.添加正常的队列

  • hello 测试处理正常逻辑

  • task_queue 模拟被拒绝的消息
    添加超时时间和死信交换机和rk

  x-dead-letter-exchange: dead_letter_exchange

  x-dead-letter-routing-key: task_queue.fail

  x-message-ttl: 600

• 4.设置死信交换机 dead_letter_exchange

  • 另外创建死信队列 dead
  • 绑定 route key task_queue.fail

死信交换机

模拟死信队列

代码实例 Python

      import pika
      #认证，生产者
      credentials = pika.PlainCredentials('guest', 'guest')
      #链接rabbit服务器（localhost是本机，如果是其他服务器请修改为ip地址）
      connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1',5672,'/',credentials))
      #通过tcp协议获取一个连接
      channel = connection.channel()
      #声明一个对下列和贾环加
      #channel.queue_declare(queue='hello')
      #被hello接受了
      channel.basic_publish(exchange='test',
                        routing_key='hello',
                        body='Hello World!')
      #发送了一个没有匹配的消息，匹配到了alter_message
      channel.basic_publish(exchange='test',
                        routing_key='hello12312',
                        body='Hello World!')
      #模拟一条虽然能被匹配到，但是无法消费的消息，然后被发送到死信队列消息
      channel.basic_publish(exchange='test',
                        routing_key='task_queue',
                        body='Hello World!')
  
 

  
 
 

• 正常队列

  
  
  

• 没有匹配到的到

  
  
  

• 被拒绝或者超时进入私信队列的

  
  
  

使用代码去创建队列和交换机 Java

         @Bean
         public ConnectionFactory connectionFactory() throws Exception {
             CachingConnectionFactory connectionFactory = new CachingConnectionFactory("127.0.0.1", 5672);
              connectionFactory.setUsername("liuxin");
              connectionFactory.setPassword("930914lx");
              connectionFactory.setVirtualHost("az");
              connectionFactory.setPublisherConfirms(true); // 必须要设置回调
             Channel channel = connectionFactory.createConnection().createChannel(false);
             //String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments
             Map<String, Object> arguments = new HashMap<>();
             arguments.put("internal",true);
             //String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             //设置AE交换机
              channel.exchangeDeclare("alter", "fanout", false, false, false, arguments);
              channel.queueDeclare("alter_message", false, false, false, null);
              channel.queueBind("alter_message", "alter", "");
             //声明死信交换机并绑定
              channel.exchangeDeclare("dead_letter_exchange", "direct", false, false, null);
              channel.queueDeclare("dead", false, false, false, null);
              channel.queueBind("dead", "dead_letter_exchange", "task_queue.fail");
             arguments = new HashMap<>();
             arguments.put("alternate-exchange", "alter");//指定AE交换机
              channel.exchangeDeclare("test", "direct", false, false, arguments);
             //声明接受正式的队列，不需要参数
              channel.queueDeclare("hello", false, false, false, null);
              channel.queueBind("hello", "test", "hello");
             arguments = new HashMap<>();
             arguments.put("x-dead-letter-exchange", "dead_letter_exchange");
             arguments.put("x-dead-letter-routing-key", "task_queue.fail");
             arguments.put("x-message-ttl",6000);//6s没有被处理，就死了
             //设置测试死信队列的task_queue，推送该队列里面，被拒绝会到dead_letter_exchange，并最终到dead，routeKey，task_queue.fail 为并设置死信队列参数
              channel.queueDeclare("task_queue", false, false, false, arguments);
              channel.queueBind("task_queue", "test", "task_queue");
             return connectionFactory;
          }
           /**
       * 接受消息的监听，这个监听客户交易流水的消息
       * 针对消费者配置
       *
       * @return
       */
         @Bean
         public SimpleMessageListenerContainer messageContainer1(ConnectionFactory connectionFactory, PayMentConsumeImpl transactionConsume) {
             SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
              container.addQueueNames("hello");
              container.setExposeListenerChannel(true);
              container.setMaxConcurrentConsumers(8);
              container.setConcurrentConsumers(4);
              container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认,当设置了此模式，必须返回ACK，否则会进入死信队列
              container.setMessageListener(transactionConsume);
              container.setPrefetchCount(1000);
             return container;
          }
  
 

  
 
 

文章来源: springlearn.blog.csdn.net，作者：西魏陶渊明，版权归原作者所有，如需转载，请联系作者。

原文链接：springlearn.blog.csdn.net/article/details/102425324