rabbitmq实现延迟消息

小鱼儿 2022-12-23 13:28 145阅读 0赞

pom

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

配置application.properties

server.port=8181
    spring.rabbitmq.addresses=192.168.75.128
    spring.rabbitmq.port=15672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=admin

配置订单队列，并配置相应的exchange和routingKey

import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class OrderQueueConfig {
    
        /**
         * 订单队列
         */
        public static final String ORDER_QUEUE = "woniu.order.queue";
    
        /**
         * 超时订单队列
         */
        public static final String ORDER_EXCHANGE = "woniu.order.exchage";
    
        /**
         * 订单exchange
         */
        public static final String ORDER_TIMEOUT_QUEUE = "woniu.order.timeout.queue";
    
    
        /**
         * 订单exchange
         */
        @Bean
        public DirectExchange orderExchange(){
            return new DirectExchange(ORDER_EXCHANGE,true,false,null);
        }
    
        /**
         * 订单队列
         */
        @Bean
        public Queue orderQueue() {
            // 设置超时转发策略 超时后消息会通过x-dead-letter-exchange 转发到x-dead-letter-routing-key绑定的队列中
            Map<String, Object> arguments = new HashMap<>(2);
            arguments.put("x-dead-letter-exchange", ORDER_EXCHANGE);
            arguments.put("x-dead-letter-routing-key", ORDER_TIMEOUT_QUEUE);
            Queue queue = new Queue(ORDER_QUEUE,true,false,false,arguments);
            return queue;
        }
    
        /**
         * 超时订单队列
         * @return
         */
        @Bean
        public Queue orderTimeoutQueue() {
            Queue queue = new Queue(ORDER_TIMEOUT_QUEUE,true,false,false);
            return queue;
        }
    
        /**
         * 订单队列绑定exchange
         * @return
         */
        @Bean
        public Binding orderQueueBinding() {
            return BindingBuilder.bind(orderQueue()).to(orderExchange()).with(ORDER_QUEUE);
        }
    
    
        /**
         * 超时订单队列绑定exchange
         * @return
         */
        @Bean
        public Binding  orderTimeoutQueueBinding() {
            return BindingBuilder.bind(orderTimeoutQueue()).to(orderExchange()).with(ORDER_TIMEOUT_QUEUE);
        }
    }

创建消费者，消费**超时队列**

import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.messaging.handler.annotation.Payload;
    import org.springframework.stereotype.Component;
    
    @Slf4j
    @Component
    public class DirectReceiver {
        @RabbitHandler
        @RabbitListener(queues = OrderQueueConfig.ORDER_TIMEOUT_QUEUE)
        public void consumeTimeOutQueue(@Payload String orderId){
            System.out.println("接收到消息的时间"+System.currentTimeMillis());
            System.out.println("消息ID="+orderId);
        }
        
    }

创建生产者，发送消息

import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    public class controller {
        @Autowired
        RabbitTemplate rabbitTemplate;
        @GetMapping("mq")
        public void send(){
            rabbitTemplate.convertAndSend(OrderQueueConfig.ORDER_EXCHANGE, OrderQueueConfig.ORDER_QUEUE, "1", message -> {
                // 设置超时时间 3000ms
                message.getMessageProperties().setExpiration("3000");
                return message;
            });
        }
    }

启动访问：http://ip:8181/mq

延迟3秒接收到消息

![20201125120048350.png][]

[20201125120048350.png]: /images/20221120/cabf1e7a37b64e538218ee6876dedab4.png