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