Browse Source

alarm-data-starter中处理工单消息

lixing 3 years ago
parent
commit
0e1dd7082b

+ 18 - 0
AlarmDataStarter/pom.xml

@@ -62,5 +62,23 @@
             <artifactId>fastjson</artifactId>
             <version>1.2.71</version>
         </dependency>
+
+        <!--rabbitmq -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.boot</groupId>
+                    <artifactId>spring-boot-starter-logging</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <!-- lang -->
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
     </dependencies>
 </project>

+ 13 - 0
AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/model/dto/common/UrlParamDTO.java

@@ -66,4 +66,17 @@ public class UrlParamDTO implements Serializable {
         this.projectId = fetchParams.getProjectId();
         this.appId = fetchParams.getAppId();
     }
+
+    /**
+     * url中的参数设置为默认值
+     *
+     * @author lixing
+     * @version V1.0 2021/9/23 9:57 上午
+     */
+    public void setDefaultUrlParam() {
+        this.userId = "system";
+        this.groupCode = "0";
+        this.projectId = "0";
+        this.appId = "0";
+    }
 }

+ 75 - 0
AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/mq/OrderMsgHandler.java

@@ -0,0 +1,75 @@
+package com.persagy.apm.energyalarmstarter.alarmdata.mq;
+
+
+import com.persagy.apm.energyalarmstarter.alarmdata.mq.model.OrderStateMessage;
+import com.persagy.apm.energyalarmstarter.alarmdata.service.OrderStateChangeService;
+import com.persagy.apm.energyalarmstarter.alarmdata.utils.StringUtil;
+import com.rabbitmq.client.Channel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.io.IOException;
+
+/**
+ * 处理工单消息
+ *
+ * @author lixing
+ * @version V1.0 2021/11/15 12:13 下午
+ */
+@Slf4j
+@Configuration
+public class OrderMsgHandler {
+    @Autowired
+    OrderStateChangeService orderStateChangeService;
+
+    /**
+     * 工单状态变化exchange, 下一期升级为对接集成框架
+     */
+    private final String orderStateExchange = "workorder_state_publish_exchange";
+    /**
+     * 工单状态变化queue
+     */
+    private final String orderStateQueue = "order_state_queue";
+
+    @Bean
+    public FanoutExchange orderStateExchange() {
+        return new FanoutExchange(orderStateExchange);
+    }
+
+    @Bean
+    public Queue orderStateQueue() {
+        return new Queue(orderStateQueue, true);
+    }
+
+    @Bean
+    public Binding orderStateBinding() {
+        return BindingBuilder.bind(orderStateQueue()).to(orderStateExchange());
+    }
+
+    @RabbitListener(queues = "order_state_queue")    //监听器监听指定的Queue
+    public void processOrderState(String message, Channel channel, Message msg) {
+        log.info("============================== Receive:" + message);
+        try {
+            OrderStateMessage orderStateMessage = StringUtil.transferItemToDTO(message, OrderStateMessage.class);
+            log.info("currentThread:{}", Thread.currentThread().getId());
+            log.info("order_id: {}", orderStateMessage.getOrder_id());
+            log.info("order_state: {}", orderStateMessage.getOrder_state());
+            // 根据工单状态消息更新报警记录状态
+            orderStateChangeService.updateAlarmWhenOrderStateChange(orderStateMessage);
+            // 手动确认消息已消费
+            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
+        } catch (Exception e) {
+            log.error("工单状态消息消费失败: {}", e.getMessage());
+            try {
+                // 将消费失败的消息移除消息队列,避免重复消费
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(),false);
+            } catch (IOException ex) {
+                log.error("工单状态消息从队列中移除失败,{}", ex.getMessage());
+            }
+        }
+    }
+}

+ 18 - 0
AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/mq/model/OrderStateMessage.java

@@ -0,0 +1,18 @@
+package com.persagy.apm.energyalarmstarter.alarmdata.mq.model;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * @description:
+ * @author: lixing
+ * @company: Persagy Technology Co.,Ltd
+ * @since: 2020/12/15 11:22 上午
+ * @version: V1.0
+ **/
+@Getter
+@Setter
+public class OrderStateMessage {
+    private String order_id;
+    private int order_state;
+}

+ 139 - 0
AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/service/OrderStateChangeService.java

@@ -0,0 +1,139 @@
+package com.persagy.apm.energyalarmstarter.alarmdata.service;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.Lists;
+import com.persagy.apm.energyalarmstarter.alarmdata.feign.DmpResult;
+import com.persagy.apm.energyalarmstarter.alarmdata.model.dto.alarmrecord.QueryAlarmRecordDTO;
+import com.persagy.apm.energyalarmstarter.alarmdata.model.dto.alarmrecord.UpdateAlarmRecordDTO;
+import com.persagy.apm.energyalarmstarter.alarmdata.model.vo.AlarmRecordItem;
+import com.persagy.apm.energyalarmstarter.alarmdata.mq.model.OrderStateMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
+
+/**
+ * @description: 工单状态变化处理
+ * @author: lixing
+ * @company: Persagy Technology Co.,Ltd
+ * @since: 2020/12/16 9:45 上午
+ * @version: V1.0
+ **/
+@Service
+public class OrderStateChangeService  {
+    @Autowired
+    private AlarmRecordServiceImpl alarmRecordService;
+
+    /**
+     * 当工单状态变化时,更新报警记录
+     *
+     * @param orderStateMessage 工单状态变化消息
+     * @author lixing
+     * @version V1.0 2021/11/15 12:08 下午
+     */
+    public void updateAlarmWhenOrderStateChange(OrderStateMessage orderStateMessage) throws Exception {
+        // 根据工单状态获取对应的报警处理状态
+        Integer alarmTreatState = getAlarmTreatState(orderStateMessage.getOrder_state());
+        if (alarmTreatState == null) {
+            return;
+        }
+
+        // 根据工单id获取要修改的报警记录
+        List<AlarmRecordItem> alarmRecords = getAlarmRecordsByOrderId(orderStateMessage.getOrder_id());
+        if (CollectionUtils.isEmpty(alarmRecords)) {
+            return;
+        }
+
+        // 更新报警记录状态
+        for (AlarmRecordItem alarmRecord : alarmRecords) {
+            // 如果报警记录中工单状态已经为处理完成,不再更新报警
+            if (workOrderDone(Integer.valueOf(alarmRecord.getOrderState()))) {
+                continue;
+            }
+
+            UpdateAlarmRecordDTO updateAlarmRecordDTO = new UpdateAlarmRecordDTO();
+            // url参数设置为默认值
+            updateAlarmRecordDTO.setDefaultUrlParam();
+            updateAlarmRecordDTO.setTreatState(alarmTreatState);
+            updateAlarmRecordDTO.setOrderState(String.valueOf(orderStateMessage.getOrder_state()));
+            alarmRecordService.update(updateAlarmRecordDTO);
+        }
+    }
+
+    /**
+     * 判断工单状态是否为已完成
+     *
+     * @param workOrderState
+     * @return boolean
+     * @exception
+     * @author lixing
+     * @version V1.0 2021/9/14 6:02 下午
+     */
+    private boolean workOrderDone(Integer workOrderState) {
+        if (workOrderState == null) {
+            return false;
+        }
+        switch (workOrderState){
+            case 7:
+            case 8:
+            case 9:
+            case 10:
+                return true;
+            default:
+                return false;
+        }
+    }
+
+    /**
+     * 根据工单id获取对应的报警记录
+     *
+     * @param orderId 工单id
+     * @return 报警记录列表
+     * @author lixing
+     * @version V1.0 2021/11/15 12:01 下午
+     */
+    private List<AlarmRecordItem> getAlarmRecordsByOrderId(String orderId) throws Exception {
+        QueryAlarmRecordDTO queryAlarmRecordDTO = new QueryAlarmRecordDTO();
+        // url参数设置为默认值
+        queryAlarmRecordDTO.setDefaultUrlParam();
+        // 查询条件
+        queryAlarmRecordDTO.setOrderId(Lists.newArrayList(orderId));
+        DmpResult<List<AlarmRecordItem>> queryResult = alarmRecordService.query(queryAlarmRecordDTO);
+
+        String resultSuccess = "success";
+        if (resultSuccess.equals(queryResult.getResult())) {
+            return queryResult.getData();
+        } else {
+            return null;
+        }
+    }
+
+    /**
+     * @description: 根据工单状态获取对应的报警处理状态
+     * @param: orderState 工单状态
+     * @return: java.lang.Integer
+     * @exception:
+     * @author: lixing
+     * @company: Persagy Technology Co.,Ltd
+     * @since: 2020/12/16 9:58 上午
+     * @version: V1.0
+     */
+    public Integer getAlarmTreatState(int orderState) {
+        int handling = 2, done = 3;
+        switch (orderState) {
+            case 4:
+            case 5:
+            case 6:
+                return handling;
+            case 7:
+            case 8:
+            case 9:
+            case 10:
+                return done;
+            default:
+                return null;
+        }
+    }
+}

+ 22 - 0
AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/utils/StringUtil.java

@@ -0,0 +1,22 @@
+package com.persagy.apm.energyalarmstarter.alarmdata.utils;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * @description:
+ * @author: lixing
+ * @company: Persagy Technology Co.,Ltd
+ * @since: 2020/12/3 4:40 下午
+ * @version: V1.0
+ **/
+public class StringUtil {
+    public static <T, R> T transferItemToDTO(String content, Class<T> t) throws Exception {
+        if (StringUtils.isNotBlank(content)) {
+            return JSONObject.parseObject(content, t);
+        }
+        return t.newInstance();
+    }
+
+
+}