package com.gkhy.safePlatform.safeCheck.mq.consumer; import com.gkhy.safePlatform.commons.utils.JsonUtils; import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckNoticeExecTaskMsg; import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckWaitExecTaskMsg; import com.gkhy.safePlatform.safeCheck.service.SafeCheckTaskUnitSchedulesService; import com.gkhy.safePlatform.safeCheck.util.CalcuteDelayLevelUtils; import com.gkhy.safePlatform.safeCheck.util.ConsolePrintIfEnvIsDevUtil; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendStatus; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Service; import java.nio.charset.StandardCharsets; import java.util.Date; @Service @RocketMQMessageListener(topic = "${rocketmq.topic.safeCheckNoticeTaskTopic}",consumerGroup = "${rocketmq.consumers.safeCheckNoticeGroup}") public class SafeCheckNoticeTaskConsumer implements RocketMQListener, RocketMQPushConsumerLifecycleListener { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Value("${rocketmq.topic.safeCheckNoticeTaskTopic}") private String safeCheckNoticeTaskTopic; @Value("${rocketmq.topic.safeCheckWaitExecTopic}") private String safeCheckWaitExecTopic; @Autowired private RocketMQTemplate rocketMQTemplate; @Autowired private SafeCheckTaskUnitSchedulesService taskUnitSchedulesService; @Override public void onMessage(MessageExt messageExt) { ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckNoticeTaskConsumer接收到消息 MSG ID : "+ messageExt.getMsgId() + " "+ new Date()); // System.out.println("\n【##】SafeCheckNoticeTaskConsumer接收到消息\t MSG ID : "+ messageExt.getMsgId() + " "+ new Date()); SafeCheckNoticeExecTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), SafeCheckNoticeExecTaskMsg.class); //logger.info("notice接收到消息内容" + msg.getTaskId() + msg.getNoticeTaskTime()); if(msg!=null && msg.getTaskId()>0){ //当前时间早于处理时间时 Long relaySeconds = (msg.getNoticeTaskTime().getTime() - new Date().getTime())/1000; if(relaySeconds > 0){ //重发消息,实现连续延时 Message message = MessageBuilder.withPayload(msg).build(); SendResult sendResult = rocketMQTemplate.syncSend(safeCheckNoticeTaskTopic, message,10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds)); if(sendResult.getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】任务通知失败消息失败" + sendResult.getMsgId()); } ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckNoticeTaskConsumer重发消息" + sendResult.getMsgId()); }else { //TODO 通知消息暂无操作方法 ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("notice暂时无动作"); //通知完后开始计算执行时间 SafeCheckWaitExecTaskMsg waitExecTaskMsg = new SafeCheckWaitExecTaskMsg(); waitExecTaskMsg.setTaskId(msg.getTaskId()); waitExecTaskMsg.setExecTaskTime(msg.getExecTaskTime()); waitExecTaskMsg.setOutTaskTime(msg.getOutTaskTime()); waitExecTaskMsg.setWorkId(msg.getWorkId()); SendResult sendResult = rocketMQTemplate.syncSend(safeCheckWaitExecTopic, waitExecTaskMsg); ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】已经进行通知,发送消息至等待执行队列 TASK ID :"+msg.getTaskId() +" 时间:"+ new Date()); if(sendResult.getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】任务待巡检-->巡检倒计时失败"+msg.getTaskId()); } } } } @Override public void prepareStart(DefaultMQPushConsumer consumer) { // consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); consumer.setConsumeThreadMin(4); // consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis())); try { consumer.subscribe(safeCheckNoticeTaskTopic,"*"); } catch (MQClientException e) { e.printStackTrace(); } } }