众泰项目新迁

郑永安

2023-09-19 69185134fcfaf913ea45f1255677225a2cc311a4

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
package com.gk.hotwork.doublePrevention.mq.consumer;
 
import com.gk.hotwork.doublePrevention.utils.CalcuteDelayLevelUtils;
import com.gk.hotwork.doublePrevention.mq.msg.PreventCreateTaskMsg;
import com.gk.hotwork.doublePrevention.service.DangerService;
import com.gk.hotwork.doublePrevention.utils.JsonUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
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.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.preventCreateTaskTopic}",consumerGroup = "${rocketmq.consumers.CreateGroup}",consumeThreadNumber = 1)
public class CreateTaskConsumer implements RocketMQListener<MessageExt> {
 
 
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
    @Value("${rocketmq.topic.preventCreateTaskTopic}")
    private String preventCreateTaskTopic;
 
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private DangerService dangerService;
 
    @Override
    public void onMessage(MessageExt messageExt) {
        System.out.println("\n【##】CreateTaskConsumer接收到消息  MSG ID : " + messageExt.getMsgId() + "---->>  【##】接收时间" + new Date());
 
        PreventCreateTaskMsg msg = JsonUtils.parse(new String(messageExt.getBody(), StandardCharsets.UTF_8), PreventCreateTaskMsg.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();
                    logger.info("【##】创建任务的消息延时"); //, 3000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds)
                    System.out.println("【##】创建任务的消息延时");
                    if(rocketMQTemplate.syncSend(preventCreateTaskTopic, message,10000, CalcuteDelayLevelUtils.calcuteDelayLevel(relaySeconds)).getSendStatus() != SendStatus.SEND_OK){
                        throw new RuntimeException("重发创建任务消息失败" + msg.getWorkId());
                    }
                }else if(relaySeconds < -4 * 60 * 60L){
                    logger.info("【##】任务消息过时，不执行任务创建");
                }else {
                    //创建task
                    logger.info("【##】开始创建任务" + new Date());
                    System.out.println(msg);
                    if(dangerService.createAutoTask(msg.getWorkId()) != true){
                       throw new RuntimeException("mq消息创建任务失败");
                    }
                }
            }
    }
 
//    @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(preventCreateTaskTopic,"*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
 
 
}