瀏覽代碼

推送数据 修改数据结构

miaolijing 3 年之前
父節點
當前提交
f572d51fcf

+ 57 - 0
src/main/java/com/persagy/apm/diagnose/config/JmsConfig.java

@@ -0,0 +1,57 @@
+package com.persagy.apm.diagnose.config;
+
+import com.alibaba.fastjson.JSONArray;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.UUID;
+
+/**
+ * 推送数据 消息
+ */
+@Slf4j
+@Configuration
+public class JmsConfig {
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+    /**
+     * 交换机
+     */
+    private final String exchange = "exchange-alarm-service";
+    /**
+     * 报警记录 路由键
+     */
+    private final String alarmDataRoutingKey = "alarm-data-routing-key";
+    /**
+     * 报警记录队列
+     */
+    private final String alarmDataQueue = "alarm-data-queue";
+
+    @Bean
+    public TopicExchange alarmServiceExchange() {
+        return new TopicExchange(exchange);
+    }
+    @Bean
+    public Queue alarmRecordQueue() {
+        return new Queue(alarmDataQueue, true);
+    }
+    @Bean
+    public Binding alarmRecordBinding() {
+        return BindingBuilder.bind(alarmRecordQueue()).to(alarmServiceExchange()).with(alarmDataRoutingKey);
+    }
+
+    public void sendMsg(JSONArray contentArray) {
+        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
+        //把消息放入routingKey_indicator对应的队列当中去,对应的是队列QUEUE_indicator
+        rabbitTemplate.convertAndSend(exchange, alarmDataRoutingKey, contentArray, correlationId);
+    }
+
+}

+ 0 - 84
src/main/java/com/persagy/apm/diagnose/config/RabbitConfig.java

@@ -1,84 +0,0 @@
-package com.persagy.apm.diagnose.config;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.amqp.core.*;
-import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.beans.factory.config.ConfigurableBeanFactory;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Scope;
-
-
-@Configuration
-public class RabbitConfig {
-
-    private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
-    @Value("${spring.rabbitmq.host}")
-    private String host;
-
-    @Value("${spring.rabbitmq.port}")
-    private int port;
-
-    @Value("${spring.rabbitmq.username}")
-    private String username;
-
-    @Value("${spring.rabbitmq.password}")
-    private String password;
-
-    //交换机
-    public static final String EXCHANGE_INDICATOR = "exchange_indicator";
-    //队列
-    public static final String QUEUE_INDICATOR = "QUEUE_indicator";
-    //路由键
-    public static final String ROUTINGKEY_INDICATOR = "routingKey_indicator";
-
-    @Bean
-    public ConnectionFactory connectionFactory() {
-        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);
-        connectionFactory.setUsername(username);
-        connectionFactory.setPassword(password);
-        connectionFactory.setVirtualHost("/");
-        connectionFactory.setPublisherConfirms(true);
-        return connectionFactory;
-    }
-
-    @Bean
-    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
-    //必须是prototype类型
-    public RabbitTemplate rabbitTemplate() {
-        RabbitTemplate template = new RabbitTemplate(connectionFactory());
-        return template;
-    }
-
-    /**
-     * Direct 交换机
-     * @return
-     */
-    @Bean
-    public TopicExchange defaultExchange() {
-        return new TopicExchange(EXCHANGE_INDICATOR);
-    }
-    /**
-     * 获取队列A
-     * @return
-     */
-    @Bean
-    public Queue queueIndicator() {
-        return new Queue(QUEUE_INDICATOR, true); //队列持久
-    }
-
-    /**
-     * 绑定 将队列和交换机绑定,并设置匹配键
-     * @return
-     */
-    @Bean
-    public Binding binding() {
-        return BindingBuilder.bind(queueIndicator()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_INDICATOR);
-    }
-
-}

+ 10 - 14
src/main/java/com/persagy/apm/diagnose/indicatorrecord/service/impl/MonitorIndicatorRecordServiceImpl.java

@@ -3,6 +3,7 @@ package com.persagy.apm.diagnose.indicatorrecord.service.impl;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.persagy.apm.diagnose.config.ComputePropertiesConfig;
+import com.persagy.apm.diagnose.config.JmsConfig;
 import com.persagy.apm.diagnose.constant.DiAgnoseConst;
 import com.persagy.apm.diagnose.constant.EnumDataTimeType;
 import com.persagy.apm.diagnose.constant.EnumTimeType;
@@ -12,7 +13,6 @@ import com.persagy.apm.diagnose.indicatorrecord.model.dto.*;
 import com.persagy.apm.diagnose.indicatorrecord.service.IMonitorIndicatorRecordService;
 import com.persagy.apm.diagnose.service.CenterDataService;
 import com.persagy.apm.diagnose.service.EnergyAlarmServiceImpl;
-import com.persagy.apm.diagnose.service.MsgProducer;
 import com.persagy.apm.diagnose.service.dto.AlarmRuleDTO;
 import com.persagy.apm.diagnose.service.dto.CalculateMethodDTO;
 import com.persagy.apm.diagnose.service.dto.ProjectDTO;
@@ -21,7 +21,6 @@ import com.persagy.apm.diagnose.timerecord.model.CalculateTimeRecord;
 import com.persagy.apm.diagnose.timerecord.service.ICalculateTimeRecordService;
 import com.persagy.apm.diagnose.utils.*;
 import com.persagy.apm.energyalarmstarter.alarmdata.service.AlarmConfigServiceImpl;
-import com.persagy.apm.energyalarmstarter.collectdata.websocket.AlarmWebSocketServer;
 import com.persagy.framework.ems.data.core.enumeration.EMSOrder;
 import com.persagy.framework.ems.data.core.enumeration.SpecialOperator;
 import com.persagy.framework.ems.data.mvc.dao.CoreDao;
@@ -88,7 +87,7 @@ public class MonitorIndicatorRecordServiceImpl implements IMonitorIndicatorRecor
     @Resource(name = "ZillionDao")
     private CoreDao coredao;
     @Autowired
-    private MsgProducer msgProducer;
+    private JmsConfig jmsConfig;
 
 
     /**
@@ -358,16 +357,16 @@ public class MonitorIndicatorRecordServiceImpl implements IMonitorIndicatorRecor
 
         for (Map.Entry<String, List<String>> entry : objIdAlarmItemCodeAndInfoCodes.entrySet()) {
             String objIdAndAlarmItemCode = entry.getKey();
-            long time = System.currentTimeMillis() + (20 * 1000);
-            if (!lockUtil.lock(projectDTO.getProjectId()+objIdAndAlarmItemCode+ "_sendData", String.valueOf(time))) {
-                // 获取不到锁,略过
-                log.error("【监测指标数据计算线程】:项目:" + projectDTO.getProjectId() +objIdAndAlarmItemCode+ "被占用");
-                continue;
-            }
-
             int indexOf = objIdAndAlarmItemCode.indexOf("_");
             String objId = objIdAndAlarmItemCode.substring(0, indexOf);
             String alarmItemCode = objIdAndAlarmItemCode.substring(indexOf + 1);
+            String lockKey = objId+"-"+alarmItemCode;
+            Object isLock = redisUtil.get(lockKey);
+            if (null == isLock){
+                continue;
+            }
+            long time = System.currentTimeMillis() + (10 * 1000);
+            lockUtil.lock(lockKey, String.valueOf(time));
 
             List<String> infCodeList = entry.getValue();
 
@@ -391,10 +390,7 @@ public class MonitorIndicatorRecordServiceImpl implements IMonitorIndicatorRecor
                 endTime = DateUtils.str2Date(dateListEntry.getKey(), DateUtils.SDFSECOND);
             }
             JSONArray sendArray = CollectDataUtil.batchBuildSendJsonParam(sendTimeKeyAndDataList,alarmItemCode);
-            msgProducer.sendIndicatorMsg(sendArray);
-            //加锁
-            long time1 = System.currentTimeMillis() + (20 * 1000);
-            lockUtil.lock(projectDTO.getProjectId()+objIdAndAlarmItemCode+ "_sendData", String.valueOf(time1));
+            jmsConfig.sendMsg(sendArray);
 //          String sentValue = CollectDataUtil.batchBuildSendParam(sendTimeKeyAndDataList,alarmItemCode);
             //AlarmWebSocketServer.sendMsgToClients(projectDTO.getProjectId(), sentValue);
             log.error("指标发送报表服务数据:" + projectDTO.getProjectId()+";"+ sendArray);

+ 10 - 13
src/main/java/com/persagy/apm/diagnose/maintenance/service/impl/ProjectDataRecordServiceImpl.java

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.persagy.apm.diagnose.config.ComputePropertiesConfig;
+import com.persagy.apm.diagnose.config.JmsConfig;
 import com.persagy.apm.diagnose.constant.DiAgnoseConst;
 import com.persagy.apm.diagnose.constant.EnumDataTimeType;
 import com.persagy.apm.diagnose.indicatorrecord.model.dto.ConfigItemDataSendTimeDTO;
@@ -94,7 +95,7 @@ public class ProjectDataRecordServiceImpl implements IProjectDataRecordService {
     @Resource(name = "ZillionDao")
     private CoreDao coredao;
     @Autowired
-    private MsgProducer msgProducer;
+    private JmsConfig jmsConfig;
 
     @Override
     @Async("asyncComputeMaintenanceDataExecutor")
@@ -278,18 +279,17 @@ public class ProjectDataRecordServiceImpl implements IProjectDataRecordService {
 
         for (Map.Entry<String, List<String>> entry : objIdAlarmItemCodeAndInfoCodes.entrySet()) {
             String objIdAndAlarmItemCode = entry.getKey();
-            long time = System.currentTimeMillis() + (20 * 1000);
-            if (!lockUtil.lock(projectId +objIdAndAlarmItemCode+ "_sendMaintenance", String.valueOf(time))) {
-                // 获取不到锁,略过
-                log.error("【监测维保数据推送线程】:项目:" + projectId + "被占用");
-                continue;
-            }
             int indexOf = objIdAndAlarmItemCode.indexOf("_");
             String objId = objIdAndAlarmItemCode.substring(0, indexOf);
             String alarmItemCode = objIdAndAlarmItemCode.substring(indexOf + 1);
-
+            String lockKey = objId+"-"+alarmItemCode;
+            Object isLock = redisUtil.get(lockKey);
+            if (null == isLock){
+                continue;
+            }
+            long time = System.currentTimeMillis() + (10 * 1000);
+            lockUtil.lock(lockKey, String.valueOf(time));
             List<String> infCodeList = entry.getValue();
-
             //通过redis查询对象该报警类型下的发数截止时间
             ConfigItemDataSendTimeDTO configItemSendTimeDTO = getConfigItemSendTimeDTO(projectId, objIdAndAlarmItemCode);
 
@@ -318,10 +318,7 @@ public class ProjectDataRecordServiceImpl implements IProjectDataRecordService {
                 endTime = com.persagy.apm.diagnose.utils.DateUtils.str2Date(dateListEntry.getKey(), com.persagy.apm.diagnose.utils.DateUtils.SDFSECOND);
             }
             JSONArray sendArray = CollectDataUtil.batchBuildSendJsonParam(sendTimeKeyAndDataList,alarmItemCode);
-            msgProducer.sendIndicatorMsg(sendArray);
-            //加锁
-            long time1 = System.currentTimeMillis() + (20 * 1000);
-            lockUtil.lock(projectId +objIdAndAlarmItemCode+ "_sendMaintenance", String.valueOf(time1));
+            jmsConfig.sendMsg(sendArray);
 //            String sentValue = CollectDataUtil.batchBuildSendParam(sendTimeKeyAndDataList,alarmItemCode);
          //   AlarmWebSocketServer.sendMsgToClients(projectId, sentValue);
             log.info("设备维保发送数据服务数据:" + projectId+";"+ sendArray);

+ 0 - 51
src/main/java/com/persagy/apm/diagnose/service/MsgProducer.java

@@ -1,51 +0,0 @@
-package com.persagy.apm.diagnose.service;
-
-import com.alibaba.fastjson.JSONArray;
-import com.persagy.apm.diagnose.config.RabbitConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.amqp.rabbit.connection.CorrelationData;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-import java.util.UUID;
-
-/**
- * 生产者
- */
-@Component
-public class MsgProducer implements RabbitTemplate.ConfirmCallback {
-
-    private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
-    //由于rabbitTemplate的scope属性设置为ConfigurableBeanFactory.SCOPE_PROTOTYPE,所以不能自动注入
-    private RabbitTemplate rabbitTemplate;
-    /**
-     * 构造方法注入rabbitTemplate
-     */
-    @Autowired
-    public MsgProducer(RabbitTemplate rabbitTemplate) {
-        this.rabbitTemplate = rabbitTemplate;
-        rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果为单例的话,那回调就是最后设置的内容
-    }
-
-    public void sendIndicatorMsg(JSONArray contentArray) {
-        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
-        //把消息放入routingKey_indicator对应的队列当中去,对应的是队列QUEUE_indicator
-        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_INDICATOR, RabbitConfig.ROUTINGKEY_INDICATOR, contentArray, correlationId);
-    }
-
-    /**
-     * 回调
-     */
-    @Override
-    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
-        logger.info(" 回调id:" + correlationData);
-        if (ack) {
-            logger.info("消息成功消费");
-        } else {
-            logger.info("消息消费失败:" + cause);
-        }
-    }
-}

+ 13 - 12
src/main/java/com/persagy/apm/diagnose/utils/CollectDataUtil.java

@@ -88,20 +88,21 @@ public class CollectDataUtil {
         for (Map.Entry<String, List<MonitorIndicatorRecord>> entry : sendTimeKeyAndDataList.entrySet()) {
             String date = entry.getKey();
             List<MonitorIndicatorRecord> list = entry.getValue();
-            if (!CollectionUtils.isEmpty(list) && list.size() > 0){
-                for (int i = 0; i < list.size(); i++) {
-                    MonitorIndicatorRecord record = list.get(i);
-                    JSONObject jsonObject = new JSONObject();
-                    jsonObject.put("dataTime",date);
-                    jsonObject.put("itemCode",alarmItemCode);
-                    jsonObject.put("meterId",record.getObjId());
-                    jsonObject.put("funcId",record.getMonitorIndicatorId());
-                    jsonObject.put("value",record.getDataValue());
-                    array.add(jsonObject);
-                }
+            MonitorIndicatorRecord record1 = list.get(0);
+            JSONObject jsonObject = new JSONObject();
+            jsonObject.put("dataTime",date);
+            jsonObject.put("itemCode",alarmItemCode);
+            jsonObject.put("meterId",record1.getObjId());
+            JSONArray funcIdArray = new JSONArray();
+            JSONArray valueArray = new JSONArray();
+            for (int i = 0; i < list.size(); i++) {
+                funcIdArray.add(list.get(i).getMonitorIndicatorId());
+                valueArray.add(list.get(i).getDataValue());
             }
+            jsonObject.put("funcId",funcIdArray);
+            jsonObject.put("value",valueArray);
+            array.add(jsonObject);
         }
-//        return JSON.toJSONString(array);
         return array;
     }
 }