郑永安
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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();
        }
    }
}