Bladeren bron

更改接受发数的消息方式为mq

gaoyu 3 jaren geleden
bovenliggende
commit
b0e0bc7a23

+ 1 - 1
src/main/java/com/persagy/client/WebSocketClientFactory.java

@@ -131,7 +131,7 @@ public class WebSocketClientFactory {
                     String finalMessage = message;
                     executor.execute(() -> {
                         try {
-                            alarmHandleService.handleIotMsg(finalMessage);
+//                            alarmHandleService.handleIotMsg(finalMessage);
                         } catch (Exception e) {
                             log.error("数据处理失败", e);
                         }

+ 1 - 1
src/main/java/com/persagy/service/AlarmHandleService.java

@@ -22,7 +22,7 @@ public interface AlarmHandleService {
      * @since 2021/2/6 10:25 上午
      * @version V1.0
      */
-    void handleIotMsg(String msg) throws Exception;
+//    void handleIotMsg(String msg) throws Exception;
 
     void sendToMQ(JSONObject parseObject, String alarmId, String projectId);
 }

+ 107 - 153
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -1,5 +1,6 @@
 package com.persagy.service.impl;
 
+import java.io.IOException;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
@@ -8,12 +9,16 @@ import java.util.Date;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
 import org.apache.commons.lang3.StringUtils;
 import org.quartz.JobDataMap;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -21,6 +26,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -50,6 +56,7 @@ import com.persagy.utils.DateUtils;
 import com.persagy.utils.StringUtil;
 import com.persagy.utils.ValidateUtils;
 import com.persagy.utils.condition.ConditionUtils;
+import com.rabbitmq.client.Channel;
 
 import cn.hutool.core.date.DateUtil;
 import lombok.extern.slf4j.Slf4j;
@@ -101,76 +108,58 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 	 * @company: Persagy Technology Co.,Ltd
 	 * @since: 2021/2/1 3:31 下午
 	 * @version: V1.0
+	 * @throws IOException
 	 */
-	@Override
-	public void handleIotMsg(String msg) throws Exception {
-		log.debug("接收到采集值:[{}]", msg);
-		// 校验采集值
-		boolean validateIotMsgResult = validateIotMsg(msg);
-		if (!validateIotMsgResult) {
-			log.error("采集值校验未通过");
-			return;
-		}
-
-		/* 将iot消息拆分成一条条iot数据,然后处理 */
-		String[] split = msg.split(";");
-		// 每一条iot数据应该包含5个分号
-		int groupCount = 5;
-		int dataCounts = split.length / groupCount;
-		for (int i = 0; i < dataCounts; i++) {
-			int startIndex = i * groupCount;
-			String dataTime = split[startIndex];
-			String itemCode = split[startIndex + 1];
-			String meterId = split[startIndex + 2];
-			String funcId = split[startIndex + 3];
-			String value = split[startIndex + 4];
-			// 使用intern()方法,确保上锁的是同一个String对象
-			synchronized (meterId.intern()) {
-				// 处理iot数据
-				handleIotData(dataTime, itemCode, meterId, funcId, value);
+	@RabbitHandler
+	@RabbitListener(queues = "alarm-data-queue")
+	public void handleIotMsg(Message message, Channel channel) throws IOException {
+		try {
+			JSONArray msgJsonArray = JSON.parseArray(new String(message.getBody()));
+			log.debug("接收到采集值:[{}]", msgJsonArray);
+			// 校验采集值
+			boolean validateIotMsgResult = validateIotMsg(msgJsonArray);
+			if (!validateIotMsgResult) {
+				log.error("采集值校验未通过");
+				return;
 			}
-		}
-	}
-
-	// public static void main(String[] args) throws Exception {
-	// String msg =
-	// "20210331094500;Eq4403050002cfb2cd6c7475465397e8738ee9ada861;Pclose,EnergyData,CsLJCWDetaTemp;0.1,0.81,1.2;";
-	// AlarmHandleServiceImpl alarmHandleService = new AlarmHandleServiceImpl();
-	// alarmHandleService.handleIotMsg(msg);
-	// }
-
-	/**
-	 * @description: 处理iot数据, iot数据可能包含一个设备多个信息点的采集值
-	 * @param: dataTime  采集时间
-	 * @param: itemCode  报警类型
-	 * @param: meterId   设备id
-	 * @param: funcIdStr 信息点id
-	 * @param: valueStr  采集值
-	 * @return: void
-	 * @exception:
-	 * @author: lixing
-	 * @company: Persagy Technology Co.,Ltd
-	 * @since: 2021/2/1 3:44 下午
-	 * @version: V1.0
-	 */
-	private void handleIotData(String dataTime, String itemCode, String meterId, String funcIdStr, String valueStr)
-			throws Exception {
-		/* 如果iot数据为一组数据,先更新缓存中这组iot数据值,然后将这一组iot数据拆分为一条条iot数据处理 */
-		String funcIdSeparator = ",";
-		String[] funcIdArray = funcIdStr.split(funcIdSeparator);
-		String[] values = valueStr.split(funcIdSeparator);
-
-		// 拼装报警条件参数
-		HashMap<String, Object> paramMap = new HashMap<>();
+			for (Object object : msgJsonArray) {
+				JSONObject jsonObject = (JSONObject) object;
+				// 采集时间
+				String dataTime = jsonObject.getString("dataTime");
+				// 报警类型
+				String itemCode = jsonObject.getString("itemCode");
+				// 设备id
+				String meterId = jsonObject.getString("meterId");
+				// 信息点id
+				String funcId = jsonObject.getString("funcId");
+				// 采集值
+				String value = jsonObject.getString("value");
+				synchronized (meterId.intern()) {
+					// 处理iot数据
+					HashMap<String, Object> paramMap = new HashMap<>();
+					JSONArray funcIdArray = JSON.parseArray(funcId);
+					JSONArray valuesArray = JSON.parseArray(value);
+					for (int i = 0; i < funcIdArray.size(); i++) {
+						paramMap.put((String) funcIdArray.getString(i), Double.parseDouble(valuesArray.getString(i)));
+					}
+					triggerOrEndAlarm(dataTime, meterId, itemCode, paramMap);
+				}
 
-		// 拼装报警条件参数
-		for (int i = 0; i < funcIdArray.length; i++) {
-			paramMap.put(funcIdArray[i], Double.parseDouble(values[i]));
+			}
+			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+		} catch (Exception e) {
+			channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
 		}
 
-		triggerOrEndAlarm(dataTime, meterId, itemCode, paramMap);
 	}
 
+//	public static void main(String[] args) throws Exception {
+//		String msg = "[{\"dataTime\":\"20210331094500\",\"itemCode\":\"Eqdx003\",\"meterId\":\"Eq1101060001d5e6c3a9a825488e822e56f92a1b37d0\",\"funcId\":\"ACATAH_RunStatus\",\"value\":\"1.0\"}]";
+//		AlarmHandleServiceImpl alarmHandleService = new AlarmHandleServiceImpl();
+//		Message message = new Message(msg.getBytes(), null);
+//		alarmHandleService.handleIotMsg(message, null);
+//	}
+
 	/**
 	 * 通过报警对象和报警类型找到内存中对应的报警条件,计算报警的产生和恢复
 	 *
@@ -262,26 +251,18 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 	 * @since: 2021/2/1 2:26 下午
 	 * @version: V1.0
 	 */
-	private boolean validateIotMsg(String msg) {
-		// iot数据使用英文分号分隔
-		String separator = ";";
-		// 每一条iot数据应该包含4个分号
+	private boolean validateIotMsg(JSONArray msgJsonArray) {
 		int groupCount = 5;
-		String[] split = msg.split(separator);
-		if (split.length % groupCount != 0) {
-			log.error("采集值有误!每一条iot数据应该包含{}个英文分号", groupCount);
-			return false;
-		}
-		// 数据条数
-		int dataCounts = (split.length) / groupCount;
-		for (int i = 0; i < dataCounts; i++) {
-			int startIndex = i * groupCount;
-			String dataTime = split[startIndex];
-			String itemCode = split[startIndex + 1];
-			String meterId = split[startIndex + 2];
-			String funcId = split[startIndex + 3];
-			String value = split[startIndex + 4];
-
+		for (Object object : msgJsonArray) {
+			JSONObject jsonObject = (JSONObject) object;
+			if (jsonObject.keySet().size() % groupCount != 0) {
+				log.error("采集值有误!每一条iot数据应该包含{}个json 的key", groupCount);
+				return false;
+			}
+			String dataTime = jsonObject.getString("dataTime");
+			String meterId = jsonObject.getString("meterId");
+			String funcId = jsonObject.getString("funcId");
+			String value = jsonObject.getString("value");
 			boolean iotDataValidateResult = validateIotData(dataTime, meterId, funcId, value);
 			if (!iotDataValidateResult) {
 				return false;
@@ -304,40 +285,26 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 	 * @version: V1.0
 	 */
 	private boolean validateIotData(String dataTime, String meterId, String funcId, String value) {
-		String separator = ",";
 		// 校验时间格式
 		boolean dateTimeValid = ValidateUtils.validateDateTime(dataTime);
 		if (!dateTimeValid) {
 			log.error("采集值时间有误,[{}]", dataTime);
 			return false;
 		}
-
 		// 校验信息点和值是否匹配
-		int funcIdsLength = funcId.split(separator).length;
-		int valuesLength = value.split(separator).length;
-		if (funcIdsLength != valuesLength) {
+		JSONArray funcIdArray = JSON.parseArray(funcId);
+		JSONArray valueArray = JSON.parseArray(value);
+		if (funcIdArray.size() != valueArray.size()) {
 			log.error("信息点个数与值的个数不匹配");
 			return false;
 		}
-
-		// 校验值是否为合法的值
-		if (!value.contains(separator)) {
+		for (Object object : valueArray) {
 			try {
-				Double.parseDouble(value);
+				Double.parseDouble(object.toString());
 			} catch (NumberFormatException e) {
-				log.error("采集值不能被转换为数字,[{}]", value);
+				log.error("采集值不能被转换为小数,[{}]", object.toString());
 				return false;
 			}
-		} else {
-			String[] values = value.split(separator);
-			for (String tmp : values) {
-				try {
-					Double.parseDouble(tmp);
-				} catch (NumberFormatException e) {
-					log.error("采集值不能被转换为数字,[{}]", value);
-					return false;
-				}
-			}
 		}
 
 		return true;
@@ -610,14 +577,15 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 //                    alarmLastTimeCache.put(alarmId, Integer.valueOf(String.valueOf(lastTime)));
 //                    log.debug("开始拼装报警持续消息");
 //                    initAlarmContinueMsg(alarmId, lastTime, paramMap, alarmDefine.getProjectId());
-                    ObjectMapper objectMapper = new ObjectMapper();
-                    ObjectNode objectNode = objectMapper.createObjectNode();
-                    objectNode.put("id", alarmId);
-                    objectNode.put("lastTime", lastTime);
-                    objectNode.set("iotData", objectMapper.readTree(objectMapper.writeValueAsString(paramMap)));
-                    objectNode.put("groupCode", groupCode);
-					//推送报警持续消息到MQ
-					this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(objectNode)),alarmId,alarmDefine.getProjectId());
+					ObjectMapper objectMapper = new ObjectMapper();
+					ObjectNode objectNode = objectMapper.createObjectNode();
+					objectNode.put("id", alarmId);
+					objectNode.put("lastTime", lastTime);
+					objectNode.set("iotData", objectMapper.readTree(objectMapper.writeValueAsString(paramMap)));
+					objectNode.put("groupCode", groupCode);
+					// 推送报警持续消息到MQ
+					this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(objectNode)), alarmId,
+							alarmDefine.getProjectId());
 				}
 			}
 			// 之前是是正常状态
@@ -776,50 +744,36 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 		return zktAlarmRecordDO;
 	}
 
+	/**
+	 * @description: 发送创建报警记录消息
+	 * @param: alarmId     报警id
+	 * @param: alarmDefine 报警定义
+	 * @param: triggerTime 触发时间
+	 * @param: triggerInfo 触发值
+	 * @param: expireDate  过期时间
+	 * @return: void
+	 * @exception:
+	 * @author: lixing
+	 * @company: Persagy Technology Co.,Ltd
+	 * @since: 2021/1/8 下午5:38
+	 * @version: V1.0
+	 */
+	private void sendCreateAlarmRecordMessage(String alarmId, AlarmDefine alarmDefine, String triggerTime,
+			HashMap<String, Object> triggerInfo, Date expireDate) throws InterruptedException {
+		Condition condition = alarmDefine.getCondition();
+		AlarmRecord alarmRecord = AlarmRecord.builder().id(alarmId).category(alarmDefine.getCategory())
+				.concern(alarmDefine.getConcern()).level(alarmDefine.getLevel()).projectId(alarmDefine.getProjectId())
+				.state(1).triggerTime(DateUtils.parseDate(triggerTime)).remark(alarmDefine.getRemark())
+				.triggerInfo(JSONObject.toJSONString(triggerInfo)).condition(condition.toString())
+				.effectEndTime(expireDate).groupCode(groupCode).itemCode(alarmDefine.getItemCode())
+				.objId(alarmDefine.getObjId()).classCode(alarmDefine.getClassCode()).createUser(systemId).treatState(1)
+				.build();
+		// TODO: 2021/11/16 消息发送至mq
+		this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(alarmRecord)), alarmId,
+				alarmDefine.getProjectId());
+	}
 
-    /**
-     * @description: 发送创建报警记录消息
-     * @param: alarmId 报警id
-     * @param: alarmDefine 报警定义
-     * @param: triggerTime 触发时间
-     * @param: triggerInfo 触发值
-     * @param: expireDate 过期时间
-     * @return: void
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2021/1/8 下午5:38
-     * @version: V1.0
-     */
-    private void sendCreateAlarmRecordMessage(
-            String alarmId,
-            AlarmDefine alarmDefine, String triggerTime,
-            HashMap<String, Object> triggerInfo, Date expireDate) throws InterruptedException {
-        Condition condition = alarmDefine.getCondition();
-        AlarmRecord alarmRecord = AlarmRecord.builder()
-                .id(alarmId)
-                .category(alarmDefine.getCategory())
-                .concern(alarmDefine.getConcern())
-                .level(alarmDefine.getLevel())
-                .projectId(alarmDefine.getProjectId())
-                .state(1)
-                .triggerTime(DateUtils.parseDate(triggerTime))
-                .remark(alarmDefine.getRemark())
-                .triggerInfo(JSONObject.toJSONString(triggerInfo))
-                .condition(condition.toString())
-                .effectEndTime(expireDate)
-                .groupCode(groupCode)
-                .itemCode(alarmDefine.getItemCode())
-                .objId(alarmDefine.getObjId())
-                .classCode(alarmDefine.getClassCode())
-                .createUser(systemId)
-                .treatState(1)
-                .build();
-        // TODO: 2021/11/16 消息发送至mq
-		this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(alarmRecord)),alarmId,alarmDefine.getProjectId());
-    }
-
-	public void sendToMQ(JSONObject message,String alarmId,String projectId){
+	public void sendToMQ(JSONObject message, String alarmId, String projectId) {
 		DmpMessage dmpMessage = new DmpMessage();
 		dmpMessage.setDefaultValue();
 		dmpMessage.setMid(alarmId);