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<MessageExt> {
|
|
|
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();
|
}
|
}
|
|
|
}
|
