Explorar o código

处理工单消息时,增加当前工单状态和历史工单状态的对比逻辑

lixing %!s(int64=3) %!d(string=hai) anos
pai
achega
e7c574f6d0

+ 0 - 132
src/main/java/com/persagy/apm/alarmservice/common/configuration/JmsConfig.java

@@ -1,132 +0,0 @@
-//package com.persagy.apm.common.configuration;
-//
-//import com.alibaba.fastjson.JSON;
-//import com.persagy.apm.alarmservice.project.alarmruleobjrel.model.dto.AddProjectAlarmRuleObjRelDTO;
-//import com.persagy.apm.alarmservice.project.alarmruleobjrel.service.IProjectAlarmRuleObjRelService;
-//import com.persagy.apm.model.DmpMessage;
-//
-//import com.rabbitmq.client.Channel;
-//import lombok.extern.slf4j.Slf4j;
-//import org.checkerframework.checker.units.qual.A;
-//import org.springframework.amqp.core.*;
-//import org.springframework.amqp.rabbit.annotation.RabbitListener;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.beans.factory.annotation.Value;
-//import org.springframework.context.annotation.Bean;
-//
-//
-///**
-// * @author: yaoll
-// * @date: 2020-10-22
-// * @verison: 1.0
-// */
-//@Slf4j
-////@Configuration
-//public class JmsConfig {
-//
-//    @Value("${persagy.dmp.exchange}")
-//    private String exchange;
-//
-//    @Value("${persagy.dmp.alarm.routingKey}")
-//    private String dmpAlarmRoutingKey;
-//
-//    @Value("${alarm.obj.routingKey}")
-//    private String alarmObjRoutingKey;
-//
-//    @Value("${persagy.dmp.rwd.routingKey}")
-//    private String rwdRoutingKey;
-//
-//    @Value("${persagy.dmp.alarm.queue}")
-//    private String alarmQueue;
-//
-//    @Autowired
-//    private IProjectAlarmRuleObjRelService projectAlarmRuleObjRelService;
-//
-////    @Value("${persagy.dmp.alarm.rwd.queue}")
-////    private String alarmRwdQueue;
-////
-////    @Bean
-////    public Queue alarmRwdQueue() {
-////        return new Queue(alarmRwdQueue, true);
-////    }
-////
-////    @Bean
-////    public Binding alarmRwdBinding() {
-////        return BindingBuilder.bind(alarmRwdQueue()).to(exchange()).with(rwdRoutingKey);
-////    }
-//
-//    /**
-//     * 报警对象队列
-//     */
-//    private final String alarmObjQueue = "alarm-obj-queue";
-//
-//    @Bean
-//    public TopicExchange exchange() {
-//        return new TopicExchange(exchange);
-//    }
-//
-//    @Bean
-//    public Queue alarmQueue() {
-//        return new Queue(alarmQueue, true);
-//    }
-//
-//    @Bean
-//    public Binding alarmBinding() {
-//        return BindingBuilder.bind(alarmQueue()).to(exchange()).with(dmpAlarmRoutingKey);
-//    }
-//
-//    @Bean
-//    public Queue alarmObjQueue() {
-//        return new Queue(alarmObjQueue, true);
-//    }
-//
-//    @Bean
-//    public Binding alarmObjBinding() {
-//        return BindingBuilder.bind(alarmObjQueue()).to(exchange()).with(alarmObjRoutingKey);
-//    }
-//
-//    @RabbitListener(queues = {alarmObjQueue})    //监听器监听指定的Queue
-//    public void process(String message, Channel channel, Message msg) throws Exception{
-//        // 手动确认消息已消费
-//        try {
-//            log.info("============================== Receive:" + message);
-//            DmpMessage dmpMessages = JSON.parseObject(message,DmpMessage.class);
-//            String mid = dmpMessages.getMid();
-//            String type = dmpMessages.getType();
-//            String projectId = dmpMessages.getProjectId();
-//            if("instanceObjectCreate".equals(type)){
-//                AddProjectAlarmRuleObjRelDTO addProjectAlarmRuleObjRelDTO = new AddProjectAlarmRuleObjRelDTO();
-//                //addProjectAlarmRuleObjRelDTO.setProjectAlarmRuleId();
-//                addProjectAlarmRuleObjRelDTO.setProjectId(projectId);
-//                addProjectAlarmRuleObjRelDTO.setEquipId(mid);
-//                addProjectAlarmRuleObjRelDTO.setOpen(1);
-//                projectAlarmRuleObjRelService.createProjectAlarmRuleObjRel(addProjectAlarmRuleObjRelDTO);
-//            }else if("instanceObjectDelete".equals(type)){
-//
-//            }
-//            channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
-//        } catch (Exception e) {
-//            Boolean isRedeliveredFail = msg.getMessageProperties().getRedelivered();
-//            // true表示消息已经重复处理失败
-//            if (isRedeliveredFail) {
-//                // 拒绝消息,requeue=false 表示不再重新入队,如果配置了死信队列则进入死信队列
-//                log.error("重复消费消息失败,message: {}", message);
-//                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), false);
-//            } else {
-//                // 如果是第一次失败则再次放入队列
-//                // requeue为是否重新回到队列,true重新入队
-//                channel.basicNack(msg.getMessageProperties().getDeliveryTag(), false, true);
-//            }
-//        }
-//    }
-//
-////    @RabbitListener(queues ="${persagy.dmp.alarm.rwd.queue}")    //监听器监听指定的Queue
-////    public void processRwdMsg(DmpMessage message, Channel channel, Message msg) {
-////        consumeMessageThreadPool.consumeMessage(message, channel, msg);
-////    }
-//
-////    @RabbitListener(queues = "${persagy.dmp.alarm.queue}")    //监听器监听指定的Queue
-////    public void processAlarmMsg(DmpMessage message, Channel channel, Message msg) {
-////        consumeMessageThreadPool.consumeMessage(message, channel, msg);
-////    }
-//}

+ 82 - 0
src/main/java/com/persagy/apm/alarmservice/common/jms/JmsListener.java

@@ -0,0 +1,82 @@
+package com.persagy.apm.alarmservice.common.jms;
+
+import com.persagy.apm.alarmservice.dependency.order.OrderStateChangeService;
+import com.persagy.apm.energyalarmstarter.alarmdata.mq.model.OrderStateMessage;
+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/25 11:07 上午
+ **/
+@Slf4j
+@Configuration
+public class JmsListener {
+    @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());
+    }
+
+    /**
+     * 工单传过来的是一个字符串消息,alarmDataStarter中注册了一个Jackson2JsonMessageConverter的消息转换器
+     * 它会将字符串消息转换为byte数组,如果消息的类型为json,它会将消息转换为对象
+     *
+     * @author lixing
+     * @version V1.0 2021/11/22 10:06 下午
+     */
+    @RabbitListener(queues = "order_state_queue")
+    public void processOrderState(byte[] message, Channel channel, Message msg) {
+        String messageStr = new String(message);
+        log.info("============================== Receive:" + messageStr);
+        try {
+            OrderStateMessage orderStateMessage = StringUtil.transferItemToDTO(messageStr, 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);
+            try {
+                // 将消费失败的消息移除消息队列,避免重复消费
+                channel.basicReject(msg.getMessageProperties().getDeliveryTag(),false);
+            } catch (IOException ex) {
+                log.error("工单状态消息从队列中移除失败,{}", ex.getMessage());
+            }
+        }
+    }
+}

+ 145 - 0
src/main/java/com/persagy/apm/alarmservice/dependency/order/OrderStateChangeService.java

@@ -0,0 +1,145 @@
+package com.persagy.apm.alarmservice.dependency.order;
+
+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 com.persagy.apm.energyalarmstarter.alarmdata.service.AlarmRecordServiceImpl;
+import org.apache.commons.lang3.StringUtils;
+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 {
+        // 根据工单状态获取对应的报警处理状态
+        int newOrderState = orderStateMessage.getOrder_state();
+        Integer alarmTreatState = getAlarmTreatState(newOrderState);
+        if (alarmTreatState == null) {
+            return;
+        }
+
+        // 根据工单id获取要修改的报警记录
+        List<AlarmRecordItem> alarmRecords = getAlarmRecordsByOrderId(orderStateMessage.getOrder_id());
+        if (CollectionUtils.isEmpty(alarmRecords)) {
+            return;
+        }
+
+        // 更新报警记录状态
+        for (AlarmRecordItem alarmRecord : alarmRecords) {
+            String oldOrderStateStr = alarmRecord.getOrderState();
+            // 只有工单状态大于之前报警记录中的工单状态时,才更新
+            if (StringUtils.isNotBlank(oldOrderStateStr)) {
+                int oldOrderState = Integer.parseInt(oldOrderStateStr);
+                if (newOrderState <= oldOrderState) {
+                    continue;
+                }
+            }
+
+            UpdateAlarmRecordDTO updateAlarmRecordDTO = new UpdateAlarmRecordDTO();
+            updateAlarmRecordDTO.setId(alarmRecord.getId());
+            // url参数设置为默认值
+            updateAlarmRecordDTO.setDefaultUrlParam();
+            updateAlarmRecordDTO.setTreatState(alarmTreatState);
+            updateAlarmRecordDTO.setOrderState(String.valueOf(newOrderState));
+            alarmRecordService.update(updateAlarmRecordDTO);
+        }
+    }
+
+    /**
+     * 判断工单状态是否为已完成
+     *
+     * @param workOrderState
+     * @return boolean
+     * @throws
+     * @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;
+        }
+    }
+}