RabbitMQ 高级特性 TTL死信队列

桃扇骨 2023-01-16 11:09 95阅读 0赞

# 1 简介 #

**TTL：**  
time to live 消息存活时间，如果消息在存活时间内未被消费，则会别清除。 RabbitMQ⽀持两种ttl设置，单独消息进⾏配置ttl，整个队列进⾏配置ttl（居多）。

**死信队列：**  
没有被及时消费的消息存放的队列。

**死信交换机：**  
Dead Letter Exchange（死信交换机，缩写：DLX）当消息成为死信后，会被重新发送到另⼀个交换机，这个交换机就是DLX死信交换机。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]

**消息成为死信的条件：**

(1) 消费者拒收消息（basic.reject/ basic.nack），并且没有重新⼊队 requeue=false。  
(2) 消息在队列中未被消费，且超过队列或者消息本身的过期时间TTL(time-to-live)。  
(3) 队列的消息⻓度达到极限。

消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列。

**PS：** 死信队列、死信交换机和普通队列、普通交换机没有区别。

# 2 死信队列代码实现 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]

## 2.1 pom ##

<parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.3.0.RELEASE</version>
            <relativePath/> 
        </parent>
    
        <properties>
            <java.version>1.8</java.version>
        </properties>
    
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-logging</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.71</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-collections4</artifactId>
                <version>4.2</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-lang3</artifactId>
            </dependency>
    
        </dependencies>
    
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                    <version>2.3.0.RELEASE</version>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.8.1</version>
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>

## 2.2 application.yml ##

server:
      port: 8080
    
    spring:
      rabbitmq:
        host: 192.168.38.80
        port: 5672
        username: guest
        password: guest
        virtual-host: /
        listener:
          simple:
            #手动确认
            acknowledge-mode: manual

## 2.3 RabbitMqConsts ##

public final class RabbitMqConsts { 
    
        private RabbitMqConsts() { 
    
        }
    
    
        public static final String BUSINESS_EXCHANGE_NAME = "business_exchange";
    
        public static final String BUSINESS_ROUTING_KEY = "business_routing_key";
    
        public static final String DEAD_EXCHANGE_NAME = "dead_exchange";
    
        public static final String BUSINESS_QUEUE = "business_queue";
    
        public static final String DEAD_QUEUE = "dead_queue";
    
        public static final String DEAD_ROUTING_KEY = "dead_routing_key";
    
    
    }

## 2.4 DeadRabbitMqConfig ##

@Configuration
    public class DeadRabbitMqConfig { 
    
    
        /** * 业务交换机 */
        @Bean
        public Exchange businessExchange() { 
            return new DirectExchange(RabbitMqConsts.BUSINESS_EXCHANGE_NAME);
        }
    
    
        /** * 业务队列 */
        @Bean
        public Queue businessQueue() { 
            //绑定死信交换机
            Map<String, Object> paramMap = new HashMap<>();
            paramMap.put("x-dead-letter-exchange", RabbitMqConsts.DEAD_EXCHANGE_NAME);
            paramMap.put("x-dead-letter-routing-key", RabbitMqConsts.DEAD_ROUTING_KEY);
            //当前消息过期时间为5秒
            paramMap.put("x-message-ttl", 5000);
            return QueueBuilder.durable(RabbitMqConsts.BUSINESS_QUEUE).withArguments(paramMap).build();
        }
    
    
        @Bean
        public Binding businessBind(@Qualifier("businessQueue") Queue queue,
                                    @Qualifier("businessExchange") Exchange exchange) { 
    
            return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConsts.BUSINESS_ROUTING_KEY).noargs();
        }
    
        /** * 死信交换机 */
        @Bean
        public Exchange deadExchange() { 
            return new DirectExchange(RabbitMqConsts.DEAD_EXCHANGE_NAME);
        }
    
        /** * 死信队列 */
        @Bean
        public Queue deadQueue() { 
            return QueueBuilder.durable(RabbitMqConsts.DEAD_QUEUE).build();
        }
    
        /** * 绑定死信交换机和死信队列 */
        @Bean
        public Binding deadBing(@Qualifier("deadExchange") Exchange exchange, @Qualifier("deadQueue") Queue queue) { 
    
            return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConsts.DEAD_ROUTING_KEY).noargs();
        }
    
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]

## 2.5 RabbitmqConsumer ##

**描述：** 死信消费者。

@Component
    @RabbitListener(queues = RabbitMqConsts.DEAD_QUEUE)
    @Slf4j
    public class RabbitmqConsumer { 
    
    
        @RabbitHandler
        public void receiveMessage(Message message, String body, Channel channel) throws IOException { 
            System.out.println("死信队列接收到消息：" + body);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    
    }

## 2.6 ProviderController ##

@RestController
    @RequestMapping("/provider")
    @Slf4j
    public class ProviderController { 
    
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
    
        @GetMapping("/send")
        public String send() { 
            String message = "Hello SpringBoot RabbitMq";
            rabbitTemplate.convertAndSend(RabbitMqConsts.BUSINESS_EXCHANGE_NAME, RabbitMqConsts.BUSINESS_ROUTING_KEY, message );
            return "发送成功";
        }
    
    
    
    }

## 2.7 启动类 ##

@SpringBootApplication
    public class RabbitmqApplication { 
    
        public static void main(String[] args) { 
    
            SpringApplication.run(RabbitmqApplication.class);
    
        }
    
    }

## 2.8 测试 ##

**结果：** 消息发送到业务队列，队列中消息存活时间为5S，超过5S未被消费，消息进入死信队列，消费端监听死信队列消息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]  
**描述：** 过5秒后

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 6]

![在这里插入图片描述][20210425153901909.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]: /images/20221022/685336cf2bf0406b8948f848ac8267a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]: /images/20221022/f1e671de55a54cf0bce6c74b42214d6c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]: /images/20221022/fdf8ed455ee842028aac44721e3e79f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]: /images/20221022/350020991d6a4b0ba2e2241880f0ad26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 4]: /images/20221022/a4ad78cc395146a5825c3310b2b8668c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 5]: /images/20221022/498680f2f4764cad9365224bdee78283.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 6]: /images/20221022/993025d43c164216a4a816b8e8d3ca4f.png
[20210425153901909.png]: /images/20221022/9e0fe9ca511f4d1fad0c52dd37b3724a.png