延时队列两种实现方式

rabbitmq_delayed_message_exchange插件

//配置文件 /properties：
spring.rabbitmq.host=localhost
spring.rabbitmq.password=guest
spring.rabbitmq.username=guest
spring.rabbitmq.virtual-host


//绑定队列和交换机，配置类
@Configuration
public class RabbitConfig {
    
    public static final String QUEUE_NAME = "javaboy_delay_queue";
    public static final String EXCHANGE_NAME = "javaboy_delay_exchange";
    public static final String EXCHANGE_TYPE = "x-delayed-message";

    @Bean
    Queue queue() {
    
        return new Queue(QUEUE_NAME, true, false, false);
    }

    @Bean
    CustomExchange customExchange() {
    
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(EXCHANGE_NAME, EXCHANGE_TYPE, true, false,args);
    }
    
    @Bean
    Binding binding() {
    
        return BindingBuilder.bind(queue())
                .to(customExchange()).with(QUEUE_NAME).noargs();
    }
}


//消费者
@Component
public class MsgReceiver {
    
    private static final Logger logger = LoggerFactory.getLogger(MsgReceiver.class);
    @RabbitListener(queues = RabbitConfig.QUEUE_NAME)
    public void handleMsg(String msg) {
    
        logger.info("handleMsg,{}",msg);
    }
}

//生产者
@SpringBootTest
class MqDelayedMsgDemoApplicationTests {
    

    @Autowired
    RabbitTemplate rabbitTemplate;
    @Test
    void contextLoads() throws UnsupportedEncodingException {
    
        Message msg = MessageBuilder.withBody(("hello world"+new Date()).getBytes("UTF-8")).setHeader("x-delay", 3000).build();
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_NAME, RabbitConfig.QUEUE_NAME, msg);
    }

}

使用死信队列方式

//properties配置文件
spring.rabbitmq.host=localhost
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=8888

//死信队列和普通队列绑定配置类

@Configuration
public class QueueConfig {
    
    public static final String JAVABOY_QUEUE_NAME = "javaboy_queue_name";
    public static final String JAVABOY_EXCHANGE_NAME = "javaboy_exchange_name";
    public static final String JAVABOY_ROUTING_KEY = "javaboy_routing_key";
    public static final String DLX_QUEUE_NAME = "dlx_queue_name";
    public static final String DLX_EXCHANGE_NAME = "dlx_exchange_name";
    public static final String DLX_ROUTING_KEY = "dlx_routing_key";

    /** * 死信队列 * @return */
    @Bean
    Queue dlxQueue() {
    
        return new Queue(DLX_QUEUE_NAME, true, false, false);
    }

    /** * 死信交换机 * @return */
    @Bean
    DirectExchange dlxExchange() {
    
        return new DirectExchange(DLX_EXCHANGE_NAME, true, false);
    }

    /** * 绑定死信队列和死信交换机 * @return */
    @Bean
    Binding dlxBinding() {
    
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange())
                .with(DLX_ROUTING_KEY);
    }

    /** * 普通消息队列 * @return */
    @Bean
    Queue javaboyQueue() {
    
        Map<String, Object> args = new HashMap<>();
        //设置消息过期时间
        args.put("x-message-ttl", 1000*10);
        //设置死信交换机
        args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
        //设置死信 routing_key
        args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);
        return new Queue(JAVABOY_QUEUE_NAME, true, false, false, args);
    }

    /** * 普通交换机 * @return */
    @Bean
    DirectExchange javaboyExchange() {
    
        return new DirectExchange(JAVABOY_EXCHANGE_NAME, true, false);
    }

    /** * 绑定普通队列和与之对应的交换机 * @return */
    @Bean
    Binding javaboyBinding() {
    
        return BindingBuilder.bind(javaboyQueue())
                .to(javaboyExchange())
                .with(JAVABOY_ROUTING_KEY);
    }
}

//死信队列消费者
@Component
public class DlxConsumer {
    
    private static final Logger logger = LoggerFactory.getLogger(DlxConsumer.class);

    @RabbitListener(queues = QueueConfig.DLX_QUEUE_NAME)
    public void handle(String msg) {
    
        logger.info(msg);
    }
}

//不配置普通队列消费者，过期之后消息会自动进入死信队列进行消费

//生产者
@SpringBootTest
class DelayQueueApplicationTests {
    

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() {
    
        System.out.println(new Date());
        rabbitTemplate.convertAndSend(QueueConfig.JAVABOY_EXCHANGE_NAME, QueueConfig.JAVABOY_ROUTING_KEY, "hello world!");
    }

}