SpringBoot+RabbitMQ实现延迟队列

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    
    @Configuration
    public class TTLDLXConfig {
    
    //死信交换机的名称；
    public static final String DLX_EXCHANGE_NAME = "dlx_boot_exchange";
    //死信队列名称；
    public static final String DLX_QUEUE_NAME = "dlx_boot_queue";
    
    //普通交换机的名称;
    public static final String TEST_EXCHANGE_NAME = "test_boot_exchange";
    //普通队列的名称;
    public static final String TEST_QUEUE_NAME = "test_boot_queue";
    
    
    /** * 声明死信交换机,我们会定义多个交换机，
     * 所以给这个注入的Bean起一个名字，同理在绑定的时候用@Qualifier注解；
     * durablie：持久化
     */
    
    @Bean("dlxExchange")
    public Exchange dlxExchange(){
        return ExchangeBuilder.topicExchange(DLX_EXCHANGE_NAME).durable(true).build();
    
    }
    
    //声明死信队列；
    @Bean("dlxQueue")
    public Queue dlxQueue(){
        return QueueBuilder.durable(DLX_QUEUE_NAME).build();
    
    }
    
    
    //绑定死信交换机和死信队列；
    @Bean
    public Binding bindDLXExchangeQueue(@Qualifier("dlxQueue") Queue queue,
                                           @Qualifier("dlxExchange") Exchange exchange){
    
        return BindingBuilder.bind(queue).to(exchange).with("text.#").noargs();
    
    }
    
    
    //声明普通交换机
    @Bean("testExchange")
    public Exchange testExchange(){
        return ExchangeBuilder.topicExchange(TEST_EXCHANGE_NAME).durable(true).build();
    
    }
    
    
    /** * 声明延迟队列;
     * ttl:设置消息过期的时间,单位为毫秒,方便测试效果,这里设置10秒过期,
     * 也就是10秒后会发送到死信交换机再到死信队列;
     * deadLetterExchange:指定死信交换机的名称;
     * deadLetterRoutingKey:发送死信消息时指定的RoutingKey
     * 上面我们死信队列和死信交换机绑定时已经指定了"text.#"为其绑定RoutingKey;
     * */
    @Bean("testQueue")
    public Queue testQueue(){
        return QueueBuilder.durable(TEST_QUEUE_NAME).ttl(10000).deadLetterExchange(DLX_EXCHANGE_NAME).
                deadLetterRoutingKey("text.info").build();
    
    }
    
    
    //绑定普通交换机与延迟队列;
    //这里的RoutingKey只是我们普通交换机与延迟队列绑定,并不会影响死信消息发送到死信交换机;
    @Bean
    public Binding bindTestExchangeQueue(@Qualifier("testQueue") Queue queue,
                                           @Qualifier("testExchange") Exchange exchange){
    
        return BindingBuilder.bind(queue).to(exchange).with("demo.#").noargs();
    
    }
    
    }

    import com.itlw.config.TTLDLXConfig;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class ProducedTest {
    
    //从IOC容器中拿模板类；
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Test
    public void test(){
        //发送消息；
        rabbitTemplate.convertAndSend(TTLDLXConfig.TEST_EXCHANGE_NAME,
                "demo.error","这是一条死信消息....");
    
    }
    }

    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    
    @Component
    public class MyListener {
    
    /** * 监听一个队列的消息；
     * queues的value一定要和我们生产者里面的死信队列的queue一致；
     * @param message 从队列中获取到的消息
     */
    
    @RabbitListener(queues = "dlx_boot_queue")
    public void myTestListener(String message){
        System.out.println(message);
        System.out.println("拿到消息后进行其他操作......");
    
    }
    }