SpringBoot 项目中集成RabbitMQ消息队列，发送消息，监听消息,延迟队列

分手后的思念是犯贱 2022-11-06 01:00 221阅读 0赞

1.添加依赖包

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

2.spring配置文件

> spring:  
>   rabbitmq:  
>     addresses:   
>     port: 5672  
>     username: admin  
>     password: admin  
>      \#虚拟host 可以不设置,使用server默认host  
>     virtual-host: /user

3.RabbitMqConfig 配置文件

>     @Configuration
>     public class RabbitMqConfig {
>         /**
>          * 交换机
>          */
>         public static final String MIDDLE_USER_EXCHANGE = "user_exchange";
>     
>         /**
>          * 活动中心
>          */
>         public static final String USER_ACTIVITY_NOTIFY = "user_register_activity_queue";
>     
>         /**
>          * 用户交换机
>          *
>          * @return
>          */
>         @Bean
>         public FanoutExchange userFanoutExchange() {
>             return new FanoutExchange(MIDDLE_USER_EXCHANGE, true, false);
>         }
>     
>     
>         /**
>          * 活动中心消费队列
>          *
>          * @return
>          */
>         @Bean
>         public Queue userActivityQueue() {
>             return new Queue(USER_ACTIVITY_NOTIFY, true);
>         }
>     
>     
>         /**
>          * 活动中心绑定交换器
>          *
>          * @return
>          */
>         @Bean
>         public Binding userActivityQueueBinding() {
>             return BindingBuilder.bind(userActivityQueue()).to(userFanoutExchange());
>         }
>     }

4.发送消息

>     Map map = new HashMap<String, Object>();
>     
>     map.put("orderId", questionOrderRecordVO.getOrderId());
>     
>     
>     rabbitTemplate.convertAndSend(RabbitMqConfig.MIDDLE_USER_EXCHANGE, null, map);
> 
>

5. 监听消息队列

>     @RabbitListener(bindings = {@QueueBinding(
>             value = @Queue(value = USER_ACTIVITY_NOTIFY , durable = "true"),
>             exchange = @Exchange(value = MIDDLE_USER_EXCHANGE, type = ExchangeTypes.FANOUT))})
>     public void receiveUserRegister(@Payload String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
>         JSONObject json = JSONObject.parseObject(msg);
>         if (!"register".equals(json.getString("type"))) return;
>         log.info("监听到用户注册消息:{}", msg);
>         try {
>     
>         } catch (Exception e) {
>             log.error("消费用户信息出错：{}", e.getMessage());
>         } finally {
>             log.info("========ack========");
>             //ack
>             channel.basicAck(deliveryTag, false);
>         }
>     }
> 
>

6.发送延迟队列

>     Map map = new HashMap<String, Object>();
>     
>     map.put("orderId", questionOrderRecordVO.getOrderId());
>     
>     rabbitTemplate.convertAndSend(RabbitMqConfig.MIDDLE_USER_EXCHANGE, null, map, message -> {
>     //延迟一分钟
>         message.getMessageProperties().setExpiration(1000 * 60 * 1 + "");
>         return message;
>     });