rabbitmq的死信队列

╰半橙微兮° 2023-10-08 12:57 9阅读 0赞

**死信队列的概念：**

> 用来存放变成死信的消息的队列

**死信产生的途径：**

1.  消息被否定确认，使用 channel.basicNack 或 channel.basicReject ，并且此时requeue 属性被设置为false。
2.  消息在队列的存活时间超过设置的TTL时间
3.  消息队列的消息数量已经超过最大队列长度

**如何配置死信队列**

1.  配置死信队列，配置死信交换机，将死信队列绑定到死信交换机上，指定路由key
2.  配置业务队列（同时绑定对应的死信交换机和路由key），配置业务交换机，再将业务队列绑定到业务交换机

> 在需要使用死信的业务队列中配置一个死信交换机，这里同一个项目的死信交换机可以共用一个，然后为每个业务队列分配一个单独的路由key。  
> 有了死信交换机和路由key后，就像配置业务队列一样，配置死信队列，然后绑定在死信交换机上。  
> 也就是说，死信队列并不是什么特殊的队列，只不过是绑定在死信交换机上的队列。死信交换机也不是什么特殊的交换机，只不过是用来接受死信的交换机，然后死信的产生途径不是客户端发送的，而是由上面说的几种方式产生的。

**用法**

首先消息消费端改成 消息手动确认

# 开启手动确认消息，这样订阅者必须要手动 ack
    spring.rabbitmq.listener.direct.acknowledge-mode=manual
    spring.rabbitmq.listener.simple.acknowledge-mode=manual

消息发送端：

String DEAD_LETTER_QUEUEA_ROUTING_KEY = "dead_letter_key";
    
         // 声明死信队列A
         @Bean("deadLetterQueueA")
         public Queue deadLetterQueueA(){
        
              return new Queue("deadLetterQueueA");
         }
    
         // 声明死信Exchange
         @Bean("deadLetterExchange")
         public DirectExchange deadLetterExchange(){
        
              return new DirectExchange("deadLetterExchange");
         }
    
         // 声明 死信队列A 与 死信交换机 的绑定关系
         @Bean
         public Binding deadLetterBindingA(@Qualifier("deadLetterQueueA") Queue queue,
                                           @Qualifier("deadLetterExchange") DirectExchange exchange){
        
              return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUEA_ROUTING_KEY);
         }
    
         // 声明业务Exchange
         @Bean("businessExchange")
         public FanoutExchange businessExchange(){
        
              return new FanoutExchange("businessExchange");
         }
    
         // 声明业务队列A
         @Bean("businessQueueA")
         public Queue businessQueueA(){
        
              Map<String, Object> args = new HashMap<>(2);
              // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
              args.put("x-dead-letter-exchange", "deadLetterExchange");
              // x-dead-letter-routing-key  这里声明当前队列的死信路由key
              args.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUEA_ROUTING_KEY);
              return QueueBuilder.durable("businessQueueA").withArguments(args).build();
         }
    
         // 声明 业务队列A 与 业务交换机 的绑定关系
         @Bean
         public Binding businessBindingA(@Qualifier("businessQueueA") Queue queue,
                                         @Qualifier("businessExchange") FanoutExchange exchange){
        
              return BindingBuilder.bind(queue).to(exchange);
         }

消息接收端（消费端）

// 业务 + 死信队列
        @RabbitListener(queues = "businessQueueA")
        public void receivebusinessQueueA(Message message, Channel channel) throws IOException {
        
            String msg = new String(message.getBody());
            if(msg.equals("1")){
        
                // 业务处理
                System.out.println("收到业务消息：" + msg);
                // 如果已经配置了（默认）自动确认，就不要在手动确认了
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }else{
        
                // 模拟这种场景下要将他放进死信队列
                System.out.println("消息消费异常，放进死信队列");
                // channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            }
        }
        @RabbitListener(queues = "deadLetterQueueA")
        public void receiveA(Message message, Channel channel) throws IOException {
        
            System.out.println("收到死信消息A：" + new String(message.getBody()));
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }

发送消息的接口

@GetMapping("/send6")
        @ResponseBody
        public String send6(String index){
        
            // 业务逻辑代码
    
            // 发送消息到消息队列
            // template.convertAndSend("businessExchange","","1");
            template.convertAndSend("businessExchange","",index);
            return "发送消息到队列成功";
        }