package com.gkhy.safePlatform.doublePrevention.mq.consumer;
|
|
import com.gkhy.safePlatform.commons.utils.CalcuteDelayLevelUtils;
|
import com.gkhy.safePlatform.doublePrevention.entity.PreventDangerCheckTask;
|
import com.gkhy.safePlatform.doublePrevention.enums.CheckTaskStatusEnum;
|
import com.gkhy.safePlatform.doublePrevention.enums.StatusEnum;
|
import com.gkhy.safePlatform.doublePrevention.enums.SyncEnum;
|
import com.gkhy.safePlatform.doublePrevention.mq.msg.PreventTimeOutTaskMsg;
|
import com.gkhy.safePlatform.doublePrevention.service.baseService.PreventDangerCheckTaskService;
|
import com.gkhy.safePlatform.doublePrevention.service.baseService.PreventDangerCheckWorkService;
|
import com.gkhy.safePlatform.commons.utils.JsonUtils;
|
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.redisson.api.RLock;
|
import org.redisson.api.RedissonClient;
|
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;
|
import java.util.concurrent.TimeUnit;
|
|
@Service
|
@RocketMQMessageListener(topic = "${rocketmq.topic.preventTimeOutTaskTopic}",consumerGroup = "${rocketmq.consumers.TimeOutGroup}")
|
public class TimeOutTaskConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
|
|
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
|
@Value("${rocketmq.topic.preventTimeOutTaskTopic}")
|
private String preventTimeOutTaskTopic;
|
|
@Autowired
|
private RedissonClient redissonClient;
|
@Autowired
|
private RocketMQTemplate rocketMQTemplate;
|
@Autowired
|
private PreventDangerCheckWorkService preventDangerCheckWorkService;
|
@Autowired
|
private PreventDangerCheckTaskService preventDangerCheckTaskService;
|
|
@Override
|
public void onMessage(MessageExt messageExt) {
|
|
logger.info("\n【双重预防】preventTimeOutTaskConsumer接收到消息 \t MSG ID : " + messageExt.getMsgId() + new Date());
|
|
PreventTimeOutTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), PreventTimeOutTaskMsg.class);
|
|
//logger.info("TimeOutTaskConsumer接收到消息内容" + msg.getTaskId()+msg.getWorkId()+msg.getOutTaskTime());
|
//TODO 要对过期的消息做处理，可能要使用配置中心
|
if(msg!=null && msg.getWorkId()>0){
|
//当前时间早于处理时间时
|
Long relaySeconds = (msg.getOutTaskTime().getTime() - new Date().getTime())/1000;
|
if(relaySeconds > 0){
|
//重发消息，实现连续延时
|
Message message = MessageBuilder.withPayload(msg).build();
|
|
SendResult sendResult = rocketMQTemplate.syncSend(preventTimeOutTaskTopic, message, 10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds));
|
//logger.info("重发内容= "+ message + " " + message.getPayload() + " " +relaySeconds+ " " + CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds));
|
if(sendResult.getSendStatus() != SendStatus.SEND_OK){
|
throw new RuntimeException("任务通知失败消息失败" + msg.getOutTaskTime());
|
}
|
logger.info("\n【双重预防】TimeOut重发消息" + sendResult.getMsgId());
|
}else {
|
|
//1-1、加分布式锁-锁workId
|
String lockName = "PREVENT_TIMEOUT_" + msg.getTaskId();
|
RLock timeoutLock = redissonClient.getLock(lockName);
|
timeoutLock.lock(3, TimeUnit.SECONDS);
|
//1-2检查加锁的作业状态是否已经修改
|
PreventDangerCheckTask taskById = preventDangerCheckTaskService.getTaskById(msg.getTaskId());
|
//如果task状态为 3 超时未执行，终止方法。
|
if(taskById.getTaskStatus() == 3){
|
return;
|
}
|
|
if (taskById.getResult() == 3){
|
//设置任务超时
|
logger.info("\n【双重预防】TimeOut设置任务状态为超时...");
|
if (preventDangerCheckTaskService.resetTaskStatus(msg.getTaskId()) < 0){
|
throw new RuntimeException("任务状态设置超时失败" + msg.getOutTaskTime());
|
}
|
}else {
|
logger.info("超时消息有误"+ msg.getOutTaskTime() + " " + msg.getTaskId());
|
}
|
|
//设置作业为开启
|
logger.info("\n【双重预防】TimeOut重置作业状态开始...");
|
if (preventDangerCheckWorkService.resetWorkStatus(msg.getWorkId()) < 0){
|
throw new RuntimeException("作业状态重置失败" + msg.getOutTaskTime());
|
}
|
|
//释放分布式锁
|
timeoutLock.unlock();
|
}
|
}
|
}
|
|
@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(preventTimeOutTaskTopic,"*");
|
} catch (MQClientException e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
