当前位置：首页 > news >正文

RabbitMq应用延时消息

news 2025/8/25 8:26:18

一.建立绑定关系

package com.lx.mq.bind;import com.lx.constant.MonitorEventConst;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/*** @author liuweiping.com* @version 1.0* @date 2023-06-26 10:04:03*/
@Slf4j
@Configuration
public class MonitorRabbitMqBinding {@Value(value = "-${spring.profiles.active}")private String profile;/*** Description: 延迟消息 <br/>* Created By: liu wei ping <br/>* Creation Time: 2023年6月26日 下午6:59:43 <br/>* <br/>* @return <br/>*/@Bean("delayExchange")public CustomExchange buildDelayedMessageNoticeExchange(){Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange(MonitorEventConst.MONITOR_DELAYED_MESSAGE_EXCHANGE + profile, "x-delayed-message", Boolean.FALSE, Boolean.FALSE, args);}@Beanpublic Queue buildDelayedMessageNoticeQueue(){return QueueBuilder.durable(MonitorEventConst.MONITOR_DELAYED_MESSAGE_QUEUE + profile).build();}@Beanpublic Binding buildDelayedMessageNoticeBinding(){return BindingBuilder.bind(buildDelayedMessageNoticeQueue()).to(buildDelayedMessageNoticeExchange()).with(MonitorEventConst.MONITOR_DELAYED_MESSAGE_ROUTING_KEY).noargs();}/*** 交车完成事件消息定时处理队列*/@Beanpublic Queue deliveryCompleteEventHandQueue() {return QueueBuilder.durable(MonitorEventConst.DELIVERY_COMPLETE_DELAYED_QUEUE + profile).build();}/*** 交车完成事件消息定时处理队列绑定*/@Beanpublic Binding deliveryCompleteBinding() {return BindingBuilder.bind(deliveryCompleteEventHandQueue()).to(buildDelayedMessageNoticeExchange()).with(MonitorEventConst.DELIVERY_COMPLETE_DELAYED_ROUTING_KEY).noargs();}}

二.建立生产者

1.消息实体

package com.lx.dto.monitor;import lombok.Data;import java.util.Date;/*** @author liuweiping.com* @version 1.0* @date 2023-06-26 10:11:06*/
@Data
public class MonitorEventMessage {/*** 事件id*/private String eventId;/*** 事件编码*/private String eventCode;/*** 业务数据*/private String businessUniqueKey;/*** 业务类型*/private String businessType;/*** 到期时间*/private Long expireMillis;/***  时间处理唯一版本号*/private Integer eventHandVersion;/*** 定时处理时间*/private Date timedOperationTime;public void setTimedOperationTime(Date timedOperationTime) {this.timedOperationTime = timedOperationTime;expireMillis = timedOperationTime.getTime() - new Date().getTime();if (expireMillis < 0) {expireMillis = 0L;}}
}

package com.lx.mq.producer;import com.lx.constant.MonitorEventConst;
import com.lx.designPattern.strategypattern.workorderbase.service.sysField.JsonUtil;
import com.lx.dto.monitor.MonitorEventMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;/*** 监控事件消息发送类*/
@Slf4j
@Component
public class MonitorEventMessageProducer {@Value(value = "-${spring.profiles.active}")private String profile;@Autowiredprivate RabbitTemplate rabbitTemplate;/***  交车完成监控事件定时发送*/public void sendDeliveryCompleteEventHandMessage(MonitorEventMessage monitorEventMessage) {String message = JsonUtil.toJson(monitorEventMessage);;rabbitTemplate.convertAndSend(MonitorEventConst.MONITOR_DELAYED_MESSAGE_EXCHANGE + profile,MonitorEventConst.DELIVERY_COMPLETE_DELAYED_ROUTING_KEY,message,msg -> {msg.getMessageProperties().setDelay(monitorEventMessage.getExpireMillis().intValue());return msg;});log.info("sending event processing messages: {}", message);//发送事件处理消息}}

三.建立消费者

package com.lx.mq.consumer;import com.lx.constant.MonitorEventConst;
import com.lx.designPattern.strategypattern.workorderbase.service.sysField.JsonUtil;
import com.lx.dto.monitor.MonitorEventMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import java.nio.charset.StandardCharsets;/*** 监控事件消息发送类*/
@Slf4j
@Component
public class MonitorEventMessageConsumer {@Value(value = "-${spring.profiles.active}")private String profile;/***  交车完成事件处理mq监听*/@RabbitListener(queues = MonitorEventConst.DELIVERY_COMPLETE_DELAYED_QUEUE + "-${spring.profiles.active}")public void dealWithDeliveryCompleteEventHandMessage(String eventMessage, Channel channel, Message message) {log.info("dealWithDeliveryCompleteEventHandMessage:【{}】", JsonUtil.toJson(eventMessage));String str = new String(message.getBody(), StandardCharsets.UTF_8);log.info("Received the message of regular loading and unloading of goods: {}", str); //收到商品定时上下架消息MonitorEventMessage monitorEventMessage = JsonUtil.toBean(eventMessage, MonitorEventMessage.class);try {analyzeHand(monitorEventMessage);}catch (Exception e){log.error("交车完成事件分析失败,参数：{}，e:{}",JsonUtil.toJson(monitorEventMessage),JsonUtil.toJson(e));}}/***  事件分析* @param monitorEventMessage*/private void  analyzeHand(MonitorEventMessage monitorEventMessage) throws Exception {}
}

四.测试类测试

package com.lx.controller;import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;import com.lx.conf.MQConfig;
import com.lx.conmon.ResultData;
import com.lx.dto.monitor.MonitorEventMessage;
import com.lx.mq.producer.MonitorEventMessageProducer;
import com.lx.utils.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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;import com.lx.constant.RedisMange;
import com.lx.utils.RedisUtil;
import org.thymeleaf.util.DateUtils;/*** @Author : liu wei ping* @CreateTime : 2019/9/3* @Description :**/@RestController
public class SendMessageController {@Autowiredprivate MonitorEventMessageProducer messageProducer;@GetMapping("/sendTopicMessage3")public ResultData<String> sendTopicMessage3() {MonitorEventMessage monitorEventMessage = new MonitorEventMessage();monitorEventMessage.setEventCode("delivery");//设置定时处理时间= 当前时间+ 定时处理时长monitorEventMessage.setTimedOperationTime(DateUtil.date(DateUtil.getCurrentMillis() + 30 * 1000));monitorEventMessage.setBusinessType("deliveryType");messageProducer.sendDeliveryCompleteEventHandMessage(monitorEventMessage);return new ResultData<>("ok");}
}

五.效果如图所示