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