郑永安
2023-06-19 7a6abd05683528032687c75e80e0bd2030a3e46c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
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<MessageExt>, 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();
        }
    }
}