SpringBoot RabbitMq 延迟，延时，死信队列

    
        
        
            org.springframework.boot
            spring-boot-starter
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        

        
        
            org.springframework.boot
            spring-boot-starter-web
        
        
            org.projectlombok
            lombok
        

        
            com.fasterxml.jackson.core
            jackson-databind
            2.9.9
        

        
            org.springframework.boot
            spring-boot-starter-amqp
        
    

设置延时队列

@Configuration
public class TTL {

    public static final String QUEUE_NAME = "ttl_queue";

    @Bean
    public Queue queue() {
        Map params = new HashMap<>();
        params.put("x-message-ttl", 5000);  //队列过期时间
        return new Queue(QUEUE_NAME, true, false, false, params);
    }
    
}

发送数据

@SpringBootTest(classes = ProducerApplication.class)
@RunWith(SpringRunner.class)
public class TTLTest {

    //1.注入RabbitTemplate
    @Autowired
    private RabbitTemplate rabbitTemplate;

    //ttl
    @Test
    public void testTTL(){
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("5000");  //消息过期时间
                return message;
            }
        };
        rabbitTemplate.convertAndSend(TTL.QUEUE_NAME, (Object) "TTL",messagePostProcessor);
    }

}

刚开始

过5秒后

因为延迟队列可以理解为延时队列过期未被消费而成为延时队列
所以我在这一用延时过期的方式验证这两种

对列设置

@Configuration
public class DLX {

    //声明死信交换机
    @Bean("dlxExchange")
    public FanoutExchange bootExchange(){
        return new FanoutExchange("DLX-EXCHANGE");
    }
    //声明死信队列
    @Bean("dlxQueue")
    public Queue bootQueue(){
        return new Queue("dlx-queue");
    }
    //绑定
    @Bean
    public Binding bindQueueExchange(@Qualifier("dlxQueue") Queue queue, @Qualifier("dlxExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }



    //声明业务交换机
    @Bean("bootExchange")
    public DirectExchange bootExchange2(){
        return new DirectExchange("boot_direct_exchange");
    }
    //声明业务队列
    @Bean("bootQueue")
    public Queue bootQueue2(){
        Map map= new HashMap<>();
		map.put("x-dead-letter-exchange","DLX-EXCHANGE"); //当成为死信队列时，转发到那个交换机
        map.put("x-dead-letter-routing-key","key1");  //当成为死信队列时，路由key设置
        return new Queue("boot-queue-dlx",true,false,false,map);
    }

    //绑定
    @Bean
    public Binding bindQueueExchange2(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("key1");
    }
}

发送数据

@SpringBootTest(classes = ProducerApplication.class)
@RunWith(SpringRunner.class)
public class DLXTest {

    //1.注入RabbitTemplate
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void TestDLX(){
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("5000");  //消息过期时间
                return message;
            }
        };
        rabbitTemplate.convertAndSend("boot_direct_exchange","key1","DLX",messagePostProcessor);
    }
}

刚开始，能在正常业务的队列中看到数据，

当时间超过延时时间之后，正常业务中数据将消失。能在死信队列中看到