SpringBoot使用RabbitMQ延迟队列

忘是亡心i 2023-10-02 13:37 60阅读 0赞

引入pom.xml依赖

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

### 一、生产者 ###

#### 1.配置文件 ####

spring
      # rabbitMQ消息中间件
      rabbitmq:
        host: MQ服务IP地址
        port: 5672
        username: 用户名
        password: 密码
        virtual-host: /
        listener:
          simple:
            retry:
              enabled: true #是否开启重试
              initial-interval: 3000 #重试时间间隔 ms
              max-attempts: 3 #重试次数
              max-interval: 15000 #重试最大时间间隔 ms
              multiplier: 2 #倍数

#### 2.MQ常量类 ####

public class RabbitMQConstant {
        
    
        public static final String EXCHANGE_NAME = "delay_exchange";
    
        public static final String EXCHANGE_TYPE = "x-delayed-message";
    
        public static final String ONE_QUEUE_NAME = "one_queue";
    
        public static final String TWO_QUEUE_NAME = "two_queue";
    
    }

#### 3.MQ枚举类 ####

@Getter
    @AllArgsConstructor
    public enum TopicEnum {
        
    
        ONE(RabbitMQConstant.ONE_QUEUE_NAME, "one_routing"),
    
        TWO(RabbitMQConstant.TWO_QUEUE_NAME, "two_routing");
    
        /**
         * 队列名称
         */
        private final String queueName;
    
        /**
         * 路由key
         */
        private final String routingKey;
    }

#### 4.配置类 ####

import me.zhengjie.rabbitmq.common.RabbitMQConstant;
    import me.zhengjie.rabbitmq.common.TopicEnum;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    @Configuration
    public class RabbitMQConfig {
        
    
        /**
         * 定义一个延迟交换机
         */
        @Bean
        public CustomExchange customExchange() {
        
            Map<String, Object> map = new HashMap<>();
            map.put("x-delayed-type", "direct");
            return new CustomExchange(RabbitMQConstant.EXCHANGE_NAME, RabbitMQConstant.EXCHANGE_TYPE, true, false, map);
        }
    
        /**
         * 创建延迟队列1
         */
        @Bean()
        public Queue createOneQueue() {
        
            return QueueBuilder.durable(TopicEnum.ONE.getQueueName()).build();
        }
    
        /**
         * 绑定路由与队列1
         */
        @Bean
        public Binding exchangeBindOneQueue() {
        
            return BindingBuilder.bind(createOneQueue()).to(customExchange()).with(TopicEnum.ONE.getRoutingKey()).noargs();
        }
    
        /**
         * 创建延迟队列2
         */
        @Bean()
        public Queue createTwoQueue() {
        
            return QueueBuilder.durable(TopicEnum.TWO.getQueueName()).build();
        }
    
        /**
         * 绑定路由与队列2
         */
        @Bean
        public Binding exchangeBindTwoQueue() {
        
            return BindingBuilder.bind(createOneQueue()).to(customExchange()).with(TopicEnum.TWO.getRoutingKey()).noargs();
        }
    
        /**
         * 序列化配置
         */
        @Bean
        public MessageConverter messageConverter() {
        
            return new Jackson2JsonMessageConverter();
        }
    }

#### 5.生产消息类 ####

import me.zhengjie.rabbitmq.common.RabbitMQConstant;
    import me.zhengjie.rabbitmq.common.TopicEnum;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.stereotype.Component;
    
    import java.sql.Timestamp;
    
    @Component
    public class RabbitDelayProducer {
        
    
        private final RabbitTemplate rabbitTemplate;
    
        public RabbitDelayProducer(RabbitTemplate rabbitTemplate) {
        
            this.rabbitTemplate = rabbitTemplate;
        }
    
        /**
         * 发送消息 One
         * @param obj 消息对象
         * @param topicEnum
         * @param delayTimeMillis 延时时间（毫秒）
         */
        public void sendMsg(Object obj, TopicEnum topicEnum, long delayTimeMillis) {
        
            rabbitTemplate.convertAndSend(RabbitMQConstant.EXCHANGE_NAME, topicEnum.getRoutingKey(), obj, (message) -> {
        
                message.getMessageProperties().setHeader(MessageProperties.X_DELAY, delayTimeMillis);
                return message;
            });
        }
    
        /**
         * 发送消息 Two
         * @param obj 消息对象
         * @param topicEnum
         * @param runTimestamp 执行时间
         */
        public void sendMsg(Object obj, TopicEnum topicEnum, Timestamp runTimestamp) {
        
            long delayTimeMillis = runTimestamp.getTime() - new Timestamp(System.currentTimeMillis()).getTime();
            rabbitTemplate.convertAndSend(RabbitMQConstant.EXCHANGE_NAME, topicEnum.getRoutingKey(), obj, (message) -> {
        
                message.getMessageProperties().setHeader(MessageProperties.X_DELAY, delayTimeMillis);
                return message;
            });
        }
    }

### 二、消费者 ###

#### 1.配置文件 ####

spring
      # rabbitMQ消息中间件
      rabbitmq:
        host: MQ服务IP地址
        port: 5672
        username: 用户名
        password: 密码
        listener:
          simple:
            retry:
              enabled: true #是否开启重试
              initial-interval: 3000 #重试时间间隔 ms
              max-attempts: 3 #重试次数
              max-interval: 15000 #重试最大时间间隔 ms
              multiplier: 2 #倍数

#### 2.消费消息 ####

@Component
    public class Consumer {
        
        
         /**
         * RabbitMQConstant.ONE_QUEUE_NAME：指队列名
         * @param obj 生产者发送的消息
         */
        @RabbitListener(queues = RabbitMQConstant.ONE_QUEUE_NAME)
        public void getOneMessage(Object obj) {
        
            System.out.println("接收队列一的消息：" + obj);
        }
    
         /**
         * RabbitMQConstant.TWO_QUEUE_NAME：指队列名
         * @param obj 生产者发送的消息
         */
        @RabbitListener(queues = RabbitMQConstant.TWO_QUEUE_NAME)
        public void getTwoMessage(Object obj) {
        
            System.out.println("接收队列二的消息：" + obj);
        }
    }