Bläddra i källkod

报警改成缓冲队列方式,保证先后一致性

menglu 3 år sedan
förälder
incheckning
8548af6f28

+ 51 - 10
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/AlarmDurationThread.java

@@ -5,14 +5,17 @@ import java.util.Date;
 
 import javax.annotation.PostConstruct;
 
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.persagy.ibms.core.data.SceneDataObject;
 import com.persagy.ibms.core.data.SceneDataValue;
 import com.persagy.ibms.data.sdk.util.RWDAlarmUtil;
 import com.persagy.ibms.data.sdk.util.RepositoryContainer;
 
+import lombok.extern.slf4j.Slf4j;
+
 @Component
 @Slf4j
 public class AlarmDurationThread extends Thread {
@@ -23,23 +26,61 @@ public class AlarmDurationThread extends Thread {
 
 	public void run() {
 		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
+		Date last_refresh_time = null;
 		while (true) {
 			try {
-				Thread.sleep(1000L * 5);
+				Thread.sleep(1L);
 			} catch (InterruptedException e) {
 				log.error(e.getMessage(), e);
 			}
 
-			Date currentTime = new Date();
-			for (String id : RepositoryContainer.RepositoryProject.id2alarmList.keySet()) {
-				SceneDataValue alarmList = RepositoryContainer.RepositoryProject.id2alarmList.get(id);
-				for (int i = 0; i < alarmList.value_array.set.size(); i++) {
-					SceneDataObject sdoInner = (SceneDataObject) alarmList.value_array.set.get(i);
-					RWDAlarmUtil.chixushijian(sdf, currentTime, sdoInner);
+			while (true) {
+				JSONObject AlarmJob = RepositoryContainer.RepositoryProject.alarmBuffer.poll();
+				if (AlarmJob == null) {
+					break;
+				}
+
+				try {
+					String type = (String) AlarmJob.get("type");
+					if (type.equals("refresh")) {
+						while (true) {
+							Thread.sleep(1L);
+							if (AlarmJob.containsKey("Content")) {
+								break;
+							}
+						}
+						JSONArray Content = AlarmJob.getJSONArray("Content");
+						RWDAlarmUtil.exe_refresh(Content);
+					} else if (type.equals("alarm")) {
+						JSONObject alarm = AlarmJob.getJSONObject("alarm");
+						RWDAlarmUtil.exe_ProcessAlarm(alarm, true);
+					} else if (type.equals("order")) {
+						String id = AlarmJob.getString("id");
+						JSONObject alarm_order = AlarmJob.getJSONObject("alarm_order");
+						RWDAlarmUtil.exe_ProcessOrderDesc(id, alarm_order);
+					} else if (type.equals("comment")) {
+						String id = AlarmJob.getString("id");
+						JSONObject dtoJSON = AlarmJob.getJSONObject("dtoJSON");
+						RWDAlarmUtil.exe_ProcessAlarm_comment(id, dtoJSON);
+					}
+				} catch (Exception e) {
+					log.error(e.getMessage(), e);
 				}
 			}
-			for (SceneDataObject sdoInner : RepositoryContainer.RepositoryProject.alarmArray.set) {
-				RWDAlarmUtil.chixushijian(sdf, currentTime, sdoInner);
+
+			Date currTime = new Date();
+			if (last_refresh_time == null || currTime.getTime() - last_refresh_time.getTime() > 1000L * 15) {
+				last_refresh_time = currTime;
+				for (String id : RepositoryContainer.RepositoryProject.id2alarmList.keySet()) {
+					SceneDataValue alarmList = RepositoryContainer.RepositoryProject.id2alarmList.get(id);
+					for (int i = 0; i < alarmList.value_array.set.size(); i++) {
+						SceneDataObject sdoInner = (SceneDataObject) alarmList.value_array.set.get(i);
+						RWDAlarmUtil.chixushijian(sdf, currTime, sdoInner);
+					}
+				}
+				for (SceneDataObject sdoInner : RepositoryContainer.RepositoryProject.alarmArray.set) {
+					RWDAlarmUtil.chixushijian(sdf, currTime, sdoInner);
+				}
 			}
 		}
 	}

+ 1 - 1
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/WebSocketCheckThread.java

@@ -29,7 +29,7 @@ public class WebSocketCheckThread extends Thread {
 	 * @since: 2020/10/20 9:54
 	 * @version: V1.0
 	 */
-	@Scheduled(initialDelay = 20000, fixedDelay = 120000)
+	@Scheduled(initialDelay = 60000, fixedDelay = 5000)
 	public void holderConneted() {
 		try {
 			WebSocketClient outCallWebSocketClientHolder = IOTWebSocketClient.client;

+ 34 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/PacketBuffer.java

@@ -0,0 +1,34 @@
+package com.persagy.ibms.data.sdk.util;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+public class PacketBuffer<T> {
+	private Queue<T> buffer = new LinkedList<T>();
+
+	public PacketBuffer() {
+	}
+
+	public synchronized boolean offer(T packet, int size) {
+		if (this.buffer.size() >= size) {
+			return false;
+		}
+
+		this.buffer.offer(packet);
+		return true;
+	}
+
+	public synchronized T poll() {
+		T MyPackage = this.buffer.poll();
+		return MyPackage;
+	}
+
+	public synchronized T take() {
+		T MyPackage = this.buffer.peek();
+		return MyPackage;
+	}
+
+	public synchronized int BufferSize() {
+		return this.buffer.size();
+	}
+}

+ 50 - 7
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RWDAlarmUtil.java

@@ -419,7 +419,7 @@ public class RWDAlarmUtil {
 		return result;
 	}
 
-	public static void chixushijian(SimpleDateFormat sdf, Date currentTime, SceneDataObject sdoInner) {
+	public static synchronized void chixushijian(SimpleDateFormat sdf, Date currentTime, SceneDataObject sdoInner) {
 		String triggerTime = (String) sdoInner.get("triggerTime").value_prim.value;
 		String duration;
 		long keep_time;
@@ -457,7 +457,7 @@ public class RWDAlarmUtil {
 		}
 	}
 
-	public static synchronized void ProcessAlarm_comment(String id, JSONObject dtoJSON) {
+	public static synchronized void exe_ProcessAlarm_comment(String id, JSONObject dtoJSON) {
 		SceneDataSet alarmList = RepositoryContainer.RepositoryProject.alarmArray;
 		for (int i = 0; i < alarmList.set.size(); i++) {
 			SceneDataObject sdoInner = (SceneDataObject) alarmList.set.get(i);
@@ -471,7 +471,15 @@ public class RWDAlarmUtil {
 		}
 	}
 
-	public static synchronized void ProcessAlarm(JSONObject alarm) {
+	public static void ProcessAlarm_comment(String id, JSONObject dtoJSON) {
+		JSONObject AlarmJob = new JSONObject();
+		AlarmJob.put("type", "comment");
+		AlarmJob.put("id", id);
+		AlarmJob.put("dtoJSON", dtoJSON);
+		RepositoryContainer.RepositoryProject.alarmBuffer.offer(AlarmJob, 16384);
+	}
+
+	public static synchronized void exe_ProcessAlarm(JSONObject alarm, boolean addWaitCompute) {
 		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
 		Date currentTime = new Date();
 		String id = (String) alarm.get("id");
@@ -575,7 +583,7 @@ public class RWDAlarmUtil {
 		}
 
 		RepositoryImpl Repository = RepositoryContainer.instance;
-		if (Repository.enable_factor) {
+		if (addWaitCompute && Repository.enable_factor) {
 			if (alarmArrayRowChange) {
 				Repository.addWaitCompute(RepositoryContainer.RepositoryProject.alarmArray);
 			} else {
@@ -607,7 +615,14 @@ public class RWDAlarmUtil {
 		}
 	}
 
-	public static synchronized void ProcessOrderDesc(String id, JSONObject alarm_order) {
+	public static void ProcessAlarm(JSONObject alarm) {
+		JSONObject AlarmJob = new JSONObject();
+		AlarmJob.put("type", "alarm");
+		AlarmJob.put("alarm", alarm);
+		RepositoryContainer.RepositoryProject.alarmBuffer.offer(AlarmJob, 16384);
+	}
+
+	public static synchronized void exe_ProcessOrderDesc(String id, JSONObject alarm_order) {
 		for (SceneDataObject sdo : RepositoryContainer.RepositoryProject.alarmArray.set) {
 			String idInner = (String) sdo.get("id").value_prim.value;
 			if (idInner.equals(id)) {
@@ -632,12 +647,31 @@ public class RWDAlarmUtil {
 		}
 	}
 
-	public static synchronized int refresh() throws Exception {
+	public static void ProcessOrderDesc(String id, JSONObject alarm_order) {
+		JSONObject AlarmJob = new JSONObject();
+		AlarmJob.put("type", "order");
+		AlarmJob.put("id", id);
+		AlarmJob.put("alarm_order", alarm_order);
+		RepositoryContainer.RepositoryProject.alarmBuffer.offer(AlarmJob, 16384);
+	}
+
+	public static synchronized void exe_refresh(JSONArray Content) throws Exception {
 		RepositoryContainer.RepositoryProject.alarmArray.set.clear();
 		for (String objId : RepositoryContainer.RepositoryProject.id2alarmList.keySet()) {
 			RepositoryContainer.RepositoryProject.id2alarmList.get(objId).value_array.set.clear();
 			RepositoryContainer.RepositoryProject.id2alarmCount.get(objId).value_prim.value = 0;
 		}
+		for (int i = 0; i < Content.size(); i++) {
+			JSONObject alarm = Content.getJSONObject(i);
+			RWDAlarmUtil.exe_ProcessAlarm(alarm, false);
+		}
+		RepositoryContainer.instance.recomputeAlarm();
+	}
+
+	public static int refresh() throws Exception {
+		JSONObject AlarmJob = new JSONObject();
+		AlarmJob.put("type", "refresh");
+		RepositoryContainer.RepositoryProject.alarmBuffer.offer(AlarmJob, 16384);
 
 		JSONArray Content;
 		if (Constant.alarm_service_name.equalsIgnoreCase("ibms-alarm")) {
@@ -731,8 +765,9 @@ public class RWDAlarmUtil {
 				log.error(e.getMessage(), e);
 				log.error(LogUtil.GetExceptionStackTrace(e));
 			}
-			RWDAlarmUtil.ProcessAlarm(alarm);
 		}
+
+		AlarmJob.put("Content", Content);
 		return Content.size();
 	}
 
@@ -812,6 +847,10 @@ public class RWDAlarmUtil {
 					String post_result = HttpClientUtil.instance("alarm").post(Constant.alarm_url + "/alarmToWorkOrder/transfer",
 							handleJSON.toJSONString());
 					result = JSON.parseObject(post_result);
+					String resultString = result.getString("Result");
+					if (!resultString.equals("success")) {
+						throw new Exception(result.toJSONString());
+					}
 				}
 				{
 					JSONObject handleJSON = new JSONObject();
@@ -822,6 +861,10 @@ public class RWDAlarmUtil {
 					String post_result = HttpClientUtil.instance("alarm").put(Constant.alarm_url + "/alarm-record/handleAlarm",
 							handleJSON.toJSONString());
 					result = JSON.parseObject(post_result);
+					String resultString = result.getString("Result");
+					if (!resultString.equals("success")) {
+						throw new Exception(result.toJSONString());
+					}
 				}
 				if (paramJSON.containsKey("comment") && paramJSON.containsKey("userId")) {
 					JSONObject commentJSON = new JSONObject();

+ 3 - 2
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RWDLoadUtil.java

@@ -621,7 +621,7 @@ public class RWDLoadUtil {
 							log.error(e.getMessage(), e);
 							log.error(LogUtil.GetExceptionStackTrace(e));
 						}
-						RWDAlarmUtil.ProcessAlarm(alarm);
+						RWDAlarmUtil.exe_ProcessAlarm(alarm, true);
 					}
 				}
 			}
@@ -1116,7 +1116,8 @@ public class RWDLoadUtil {
 			param.put("appId", "0");
 			param.put("projectId", projectId);
 			param.put("year", "" + year);
-			String post_result = HttpClientUtil.instance("zkt_control").post(Constant.zkt_control_url + "/calendar/binding/query", param.toJSONString());
+			String post_result = HttpClientUtil.instance("zkt_control").post(Constant.zkt_control_url + "/calendar/binding/query",
+					param.toJSONString());
 			JSONObject resultJSON = JSON.parseObject(post_result);
 			JSONObject Item = (JSONObject) resultJSON.get("Item");
 			JSONArray arrayToday = new JSONArray();

+ 1 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RepositoryProject.java

@@ -25,6 +25,7 @@ public class RepositoryProject {
 	public SceneDataSet alarmArray = new SceneDataSet(false, true);
 	public ConcurrentHashMap<String, SceneDataValue> id2alarmList = new ConcurrentHashMap<String, SceneDataValue>();
 	public ConcurrentHashMap<String, SceneDataValue> id2alarmCount = new ConcurrentHashMap<String, SceneDataValue>();
+	public PacketBuffer<JSONObject> alarmBuffer = new PacketBuffer<JSONObject>();
 
 	// 指定路径的控制值
 	public ConcurrentHashMap<String, JSONObject> controlValueMap = new ConcurrentHashMap<String, JSONObject>();

+ 10 - 3
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/websocket/AlarmWebSocketClient.java

@@ -7,6 +7,7 @@ import org.java_websocket.handshake.ServerHandshake;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
+import com.persagy.ibms.core.util.ComputeUtil;
 import com.persagy.ibms.core.util.LogUtil;
 import com.persagy.ibms.data.sdk.util.RWDAlarmUtil;
 import com.persagy.ibms.data.sdk.util.RepositoryContainer;
@@ -29,9 +30,15 @@ public class AlarmWebSocketClient extends WebSocketClient {
 	public void onOpen(ServerHandshake arg0) {
 		log.warn("onOpen: " + serverUri.toString());
 		try {
-			log.warn("onOpen refresh begin");
-			int alarm_size = RWDAlarmUtil.refresh();
-			log.warn("onOpen refresh finish alarm count: " + alarm_size);
+			ComputeUtil.fixedThreadPool.execute(() -> {
+				try {
+					log.warn("onOpen refresh begin");
+					int alarm_size = RWDAlarmUtil.refresh();
+					log.warn("onOpen refresh finish alarm count: " + alarm_size);
+				} catch (Exception e) {
+					e.printStackTrace();
+				}
+			});
 		} catch (Exception e) {
 			log.error("onOpen refresh error");
 			log.error(e.getMessage(), e);