package com.gkhy.safePlatform.safeCheck.mq.consumer; import com.gkhy.safePlatform.commons.utils.JsonUtils; import com.gkhy.safePlatform.safeCheck.enums.TaskStatusEnum; import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckTimeOutTaskMsg; 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.safeCheckWaitExecTopic}",consumerGroup = "${rocketmq.consumers.safeCheckWaitExecGroup}") public class SafeCheckWaitExecConsumer implements RocketMQListener, RocketMQPushConsumerLifecycleListener { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Autowired private RocketMQTemplate rocketMQTemplate; @Value("${rocketmq.topic.safeCheckWaitExecTopic}") private String safeCheckWaitExecTopic; @Value("${rocketmq.topic.safeCheckTimeOutTaskTopic}") private String safeCheckTimeOutTaskTopic; @Autowired private SafeCheckTaskUnitSchedulesService taskUnitSchedulesService; @Override public void onMessage(MessageExt messageExt) { ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckWaitExecConsumer接收到消息 MSG ID : "+messageExt.getMsgId() + new Date()); // System.out.println("\n【##】SafeCheckWaitExecConsumer接收到消息\t MSG ID : "+messageExt.getMsgId() + new Date()); SafeCheckWaitExecTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), SafeCheckWaitExecTaskMsg.class); //logger.info("WaitExecConsumer接收到消息内容" + msg.getTaskId() + msg.getExecTaskTime()); if(msg!=null && msg.getTaskId()>0){ //当前时间早于处理时间时 Long relaySeconds = (msg.getExecTaskTime().getTime() - new Date().getTime())/1000; if(relaySeconds > 0){ //重发消息,实现连续延时 Message message = MessageBuilder.withPayload(msg).build(); SendResult sendResult = rocketMQTemplate.syncSend(safeCheckWaitExecTopic, message,10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds)); if(sendResult.getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】任务待巡检-->巡检倒计时失败" + msg.getExecTaskTime()); } ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckWaitExecConsumer消息重发" + sendResult.getMsgId()); // logger.info("\nWaitExecConsumer消息重发" + message); }else { //设置任务为执行中 ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckwait设置任务状态为执行中"); if (taskUnitSchedulesService.updateTaskStatus(msg.getTaskId(), TaskStatusEnum.TASK_CHECK_RUNNING.getStatus()) <= 0){ // throw new RuntimeException("【智能巡检】任务状态:【待巡检-->巡检中】更新失败" + msg.getExecTaskTime()); return; } //任务变为巡检中,同时向超时未巡检broker发送消息,任务有效期倒计时开始 SafeCheckTimeOutTaskMsg timeOutTaskMsg = new SafeCheckTimeOutTaskMsg(); timeOutTaskMsg.setTaskId(msg.getTaskId()); timeOutTaskMsg.setWorkId(msg.getWorkId()); timeOutTaskMsg.setExecTaskTime(msg.getExecTaskTime()); timeOutTaskMsg.setOutTaskTime(msg.getOutTaskTime()); SendResult sendResult = rocketMQTemplate.syncSend(safeCheckTimeOutTaskTopic, timeOutTaskMsg); ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】任务开始执行,发送消息至过期消息队列,开始消息有效期倒计时 TASK ID :"+msg.getTaskId() + " 时间:" + new Date()); if(sendResult.getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】有效期消息发送失败"+msg.getTaskId()); } } }else { throw new RuntimeException("【智能巡检】SafeCheckWaitExecConsumer消费失败" + msg.getExecTaskTime()); } //System.out.println("[MQ] 接收到消息:\t"+new String(messageExt.getBody())); } @Override public void prepareStart(DefaultMQPushConsumer consumer) { // consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET); consumer.setConsumeThreadMin(5); // consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis())); try { consumer.subscribe(safeCheckWaitExecTopic,"*"); } catch (MQClientException e) { e.printStackTrace(); } } }