package com.gkhy.safePlatform.safeCheck.mq.consumer; import com.gkhy.safePlatform.commons.utils.JsonUtils; import com.gkhy.safePlatform.safeCheck.entity.SafeCheckTask; import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckCreateTaskMsg; import com.gkhy.safePlatform.safeCheck.mq.msg.SafeCheckNoticeExecTaskMsg; 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.safeCheckCreateTaskTopic}",consumerGroup = "${rocketmq.consumers.safeCheckCreateGroup}",consumeThreadNumber = 1) public class SafeCheckCreateTaskConsumer implements RocketMQListener { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Value("${rocketmq.topic.safeCheckCreateTaskTopic}") private String safeCheckCreateTaskTopic; @Value("${rocketmq.topic.safeCheckNoticeTaskTopic}") private String safeCheckNoticeTaskTopic; @Autowired private RocketMQTemplate rocketMQTemplate; @Autowired private SafeCheckTaskUnitSchedulesService taskUnitSchedulesService; @Override public void onMessage(MessageExt messageExt) { ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】SafeCheckCreateTaskConsumer接收到消息 MSG ID : " + messageExt.getMsgId() + "---->> 【##】接收时间" + new Date()); // System.out.println("\n【##】SafeCheckCreateTaskConsumer接收到消息 MSG ID : " + messageExt.getMsgId() + "---->> 【##】接收时间" + new Date()); SafeCheckCreateTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), SafeCheckCreateTaskMsg.class); //Message message = MessageBuilder.withPayload(msgStr).build(); if(msg!=null && msg.getWorkId()>0){ //当前时间早于目标处理时间前0秒 Long relaySeconds = (msg.getCreateTaskTime().getTime() - new Date().getTime())/1000; if(relaySeconds > 0){ //重发消息,实现连续延时 Message message = MessageBuilder.withPayload(msg).build(); ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】创建任务的消息延时"); //, 3000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds) if(rocketMQTemplate.syncSend(safeCheckCreateTaskTopic, message,10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds)).getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】重发创建任务消息失败" + msg.getWorkId()); } }else { //创建task ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】开始创建任务" + new Date()); //创建成功后的task SafeCheckTask task = taskUnitSchedulesService.createAutoTask(msg.getWorkId()); if(task == null){ return; } ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】任务创建完成" + new Date()); SafeCheckNoticeExecTaskMsg noticeExecTaskMsg = new SafeCheckNoticeExecTaskMsg(); noticeExecTaskMsg.setTaskId(task.getId()); noticeExecTaskMsg.setWorkId(msg.getWorkId()); noticeExecTaskMsg.setNoticeTaskTime(task.getNoticeTime()); noticeExecTaskMsg.setExecTaskTime(task.getStartTime()); noticeExecTaskMsg.setOutTaskTime(task.getValidTime()); SendResult sendResult = rocketMQTemplate.syncSend(safeCheckNoticeTaskTopic, noticeExecTaskMsg); ConsolePrintIfEnvIsDevUtil.consolePrintIfEnvIsDev("【##】发送消息至通知队列 TASK ID : "+task.getId() +" 时间:"+ new Date()); if(sendResult.getSendStatus() != SendStatus.SEND_OK){ throw new RuntimeException("【智能巡检】通知消息发送失败"+task.getId()); } } } } // @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(safeCheckCreateTaskTopic,"*"); } catch (MQClientException e) { e.printStackTrace(); } } }