RabbitMQ延时队列实现方法

 @RabbitListener(bindings = @QueueBinding(
            value=@Queue(name="dlx.queue",durable =" true"),
            exchange = @Exchange(name="dlx.direct"),
            key = {"hi"}
    ))
    public void listenDlxQueue(String message)throws Exception{
        log.info("消费者监听到 dlx.queue的消息，{}",message);

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class NormalConfiguration {
    @Bean
    public DirectExchange normalExchange(){
        return  new DirectExchange("normal.direct");
    }
    @Bean
    public Queue normalQueue(){
        return  QueueBuilder
                .durable("normal.queue")
                .deadLetterExchange("dlx.direct")
                .build();
    }
//推荐依赖注入的方式
//需要注意代码规范，方法名应为小驼峰，反之，会找不到bean
  @Bean
    public Binding normalExchangeBinding(Queue normalQueue, DirectExchange normalExchange){
        return  BindingBuilder
                .bind(normalQueue)
                .to(normalExchange)
                .with("hi");
    }
//只能使用当前类下的bean
   /* @Bean
    public Binding normalExchangeBinding(){
        return  BindingBuilder
                .bind(normalQueue())
                .to(normalExchange())
                .with("hi");
    }*/
}

@Test
    public void testSendDelayMessage() throws Exception {
        rabbitTemplate.convertAndSend("normal.direct", "hi", "hello everyone__", message -> {
            message.getMessageProperties().setExpiration(  "10000");
            return message;
        });
    }

docker volume inspect mq-plugins

cd /var/lib/docker/volumes/mq-plugins/_data

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

@RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "delay.queue", durable = "true"),
            exchange = @Exchange(value = "delay.direct", delayed = "true"),
            key = "hidelay"
    ))
    public void listenDelayQueue(String msg) {
        log.info("delay.queue：" + msg);
    }
//通过  delayed = "true" 声明为延迟交换机

@Configuration
public class DirectConfiguration {
    @Bean
    public DirectExchange delayExchange() {
        return ExchangeBuilder
                .directExchange("delay.direct")
                .delayed()//这里声明为延迟交换机
                .durable(true)
                .build();
    }
    @Bean
    public Queue delayedQueue() {
        return new Queue("delay.queue");
    }
    @Bean
    public Binding delayQueueBinding() {
        return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
    }
}

 @Test
    public void testSendDelayMessageByplugin() {
        rabbitTemplate.convertAndSend("delay.direct", "hidelay", "hello", new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelay(10000);
                return message;
            }
        });
        log.info("消息发送成功");
    }