miaolijing 3 år sedan
förälder
incheckning
cf8043e45b

+ 8 - 0
pom.xml

@@ -197,5 +197,13 @@
                  </exclusion>-->
             </exclusions>
         </dependency>
+
+        <!-- RabbiMQ -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
+
+
     </dependencies>
 </project>

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

@@ -0,0 +1,100 @@
+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_MAINTENANCE = "exchange_maintenance";
+    public static final String EXCHANGE_INDICATOR = "exchange_indicator";
+    //队列
+    public static final String QUEUE_MAINTENANCE = "QUEUE_maintenance";
+    public static final String QUEUE_INDICATOR = "QUEUE_indicator";
+    //路由键
+    public static final String ROUTINGKEY_MAINTENANCE = "routingKey_maintenance";
+    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);
+    }
+
+    /**
+     * 获取队列B
+     * @return
+     */
+    @Bean
+    public Queue queueMaintenance() {
+        return new Queue(QUEUE_MAINTENANCE, true); //队列持久
+    }
+
+    @Bean
+    public Binding bindingMaintenance() {
+        return BindingBuilder.bind(queueMaintenance()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_MAINTENANCE);
+    }
+}

+ 15 - 3
src/main/java/com/persagy/apm/diagnose/indicatorrecord/service/impl/MonitorIndicatorRecordServiceImpl.java

@@ -1,5 +1,6 @@
 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.constant.DiAgnoseConst;
@@ -11,6 +12,7 @@ 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;
@@ -85,6 +87,8 @@ public class MonitorIndicatorRecordServiceImpl implements IMonitorIndicatorRecor
 
     @Resource(name = "ZillionDao")
     private CoreDao coredao;
+    @Autowired
+    private MsgProducer msgProducer;
 
 
     /**
@@ -354,6 +358,12 @@ 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);
@@ -380,9 +390,11 @@ public class MonitorIndicatorRecordServiceImpl implements IMonitorIndicatorRecor
                 sendTimeKeyAndDataList.put(dateListEntry.getKey(),value);
                 endTime = DateUtils.str2Date(dateListEntry.getKey(), DateUtils.SDFSECOND);
             }
-            String sentValue = CollectDataUtil.batchBuildSendParam(sendTimeKeyAndDataList,alarmItemCode);
-            AlarmWebSocketServer.sendMsgToClients(projectDTO.getProjectId(), sentValue);
-            log.error("指标发送报表服务数据:" + projectDTO.getProjectId()+";"+ sentValue);
+            JSONArray sendArray = CollectDataUtil.batchBuildSendJsonParam(sendTimeKeyAndDataList,alarmItemCode);
+            msgProducer.sendIndicatorMsg(sendArray);
+//          String sentValue = CollectDataUtil.batchBuildSendParam(sendTimeKeyAndDataList,alarmItemCode);
+            //AlarmWebSocketServer.sendMsgToClients(projectDTO.getProjectId(), sentValue);
+            log.error("指标发送报表服务数据:" + projectDTO.getProjectId()+";"+ sendArray);
             configItemSendTimeDTO.setLastSendTime(endTime);
             configItemSendTimeDTO.setUpdateTime(new Date());
             updateRedisConfigItemSendTimeDTO(projectDTO.getProjectId(), objIdAndAlarmItemCode, configItemSendTimeDTO);

+ 14 - 8
src/main/java/com/persagy/apm/diagnose/maintenance/service/impl/ProjectDataRecordServiceImpl.java

@@ -13,10 +13,7 @@ import com.persagy.apm.diagnose.maintenance.constant.*;
 import com.persagy.apm.diagnose.maintenance.model.dto.CalculateMethodDTO;
 import com.persagy.apm.diagnose.maintenance.model.dto.WeatherDTO;
 import com.persagy.apm.diagnose.maintenance.service.IProjectDataRecordService;
-import com.persagy.apm.diagnose.service.CenterDataService;
-import com.persagy.apm.diagnose.service.EnergyAlarmServiceImpl;
-import com.persagy.apm.diagnose.service.SaasService;
-import com.persagy.apm.diagnose.service.WeatherServiceImpl;
+import com.persagy.apm.diagnose.service.*;
 import com.persagy.apm.diagnose.service.dto.AlarmRuleDTO;
 import com.persagy.apm.diagnose.service.dto.ProjectDTO;
 import com.persagy.apm.diagnose.service.dto.TimeDataDTO;
@@ -96,6 +93,8 @@ public class ProjectDataRecordServiceImpl implements IProjectDataRecordService {
 
     @Resource(name = "ZillionDao")
     private CoreDao coredao;
+    @Autowired
+    private MsgProducer msgProducer;
 
     @Override
     @Async("asyncComputeMaintenanceDataExecutor")
@@ -280,7 +279,12 @@ 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);
@@ -314,9 +318,11 @@ public class ProjectDataRecordServiceImpl implements IProjectDataRecordService {
                 sendTimeKeyAndDataList.put(dateListEntry.getKey(),value);
                 endTime = com.persagy.apm.diagnose.utils.DateUtils.str2Date(dateListEntry.getKey(), com.persagy.apm.diagnose.utils.DateUtils.SDFSECOND);
             }
-            String sentValue = CollectDataUtil.batchBuildSendParam(sendTimeKeyAndDataList,alarmItemCode);
-            AlarmWebSocketServer.sendMsgToClients(projectId, sentValue);
-            log.info("设备维保发送数据服务数据:" + projectId+";"+ sentValue);
+            JSONArray sendArray = CollectDataUtil.batchBuildSendJsonParam(sendTimeKeyAndDataList,alarmItemCode);
+            msgProducer.sendMaintenanceMsg(sendArray);
+//            String sentValue = CollectDataUtil.batchBuildSendParam(sendTimeKeyAndDataList,alarmItemCode);
+         //   AlarmWebSocketServer.sendMsgToClients(projectId, sentValue);
+            log.info("设备维保发送数据服务数据:" + projectId+";"+ sendArray);
             configItemSendTimeDTO.setLastSendTime(endTime);
             configItemSendTimeDTO.setUpdateTime(new Date());
             updateRedisConfigItemSendTimeDTO(projectId, objIdAndAlarmItemCode, configItemSendTimeDTO);

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

@@ -0,0 +1,57 @@
+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);
+    }
+
+    public void sendMaintenanceMsg(JSONArray contentArray) {
+        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
+        //把消息放入routingKey_maintenance对应的队列当中去,对应的是队列QUEUE_maintenance
+        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_MAINTENANCE, RabbitConfig.ROUTINGKEY_MAINTENANCE,contentArray, correlationId);
+    }
+
+    /**
+     * 回调
+     */
+    @Override
+    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
+        logger.info(" 回调id:" + correlationData);
+        if (ack) {
+            logger.info("消息成功消费");
+        } else {
+            logger.info("消息消费失败:" + cause);
+        }
+    }
+}

+ 25 - 0
src/main/java/com/persagy/apm/diagnose/utils/CollectDataUtil.java

@@ -1,5 +1,8 @@
 package com.persagy.apm.diagnose.utils;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.persagy.framework.ems.data.pojo.hbase.MonitorIndicatorRecord;
 import org.springframework.util.CollectionUtils;
 
@@ -79,4 +82,26 @@ public class CollectDataUtil {
         return stringBuilder.toString();
 
     }
+
+    public static JSONArray batchBuildSendJsonParam(Map<String, List<MonitorIndicatorRecord>> sendTimeKeyAndDataList, String alarmItemCode) {
+        JSONArray array = new JSONArray();
+        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);
+                }
+            }
+        }
+//        return JSON.toJSONString(array);
+        return array;
+    }
 }