Browse Source

IOT实时性还可以了

menglu 3 years ago
parent
commit
e3c7020dbc

+ 8 - 4
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/DaemonThread.java

@@ -2,12 +2,15 @@ package com.persagy.ibms.data.sdk.service;
 
 import java.io.File;
 import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
 
 import javax.annotation.PostConstruct;
 
 import org.springframework.stereotype.Component;
 
 import com.persagy.ibms.core.util.KeywordUtil;
+import com.persagy.ibms.core.util.RepositoryComputeThread;
 import com.persagy.ibms.data.sdk.test.CheckTest;
 import com.persagy.ibms.data.sdk.util.Constant;
 import com.persagy.ibms.data.sdk.util.DeamonUtil;
@@ -46,6 +49,8 @@ public class DaemonThread extends Thread {
 	public static String path_zkt_physical_world;
 	public static String path_ibms_physical_world;
 
+	public static List<RepositoryComputeThread> threadList = new ArrayList<RepositoryComputeThread>();
+
 	public void run() {
 		KeywordUtil.keyWordMap = Constant.keyWordMap;
 		KeywordUtil.prefixMap = Constant.prefixMap;
@@ -138,12 +143,11 @@ public class DaemonThread extends Thread {
 				log.warn("ibms_physical_world: " + tmp_ibms_physical_world);
 			}
 			try {
-				RepositoryImpl Repository = new RepositoryImpl();
-				Repository.use_thread = Constant.use_thread;
-				Repository.enable_factor = Constant.enable_factor;
-				Repository.RepositoryBase = RepositoryContainer.RepositoryBase;
+				RepositoryImpl Repository = new RepositoryImpl(RepositoryContainer.RepositoryBase, Constant.use_thread, Constant.enable_factor,
+						Constant.compute_thread_count, Constant.compute_thread_sdv_interval);
 				RWDRepositoryUtil.DoRepository(Repository, true);
 				RepositoryContainer.instance = Repository;
+				Repository.threadStart();
 				log.warn("compute finish");
 				path_config = tmp_config;
 				path_physical_world = tmp_physical_world;

+ 0 - 76
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/RealtimeComputeThread.java

@@ -1,76 +0,0 @@
-package com.persagy.ibms.data.sdk.service;
-
-import javax.annotation.PostConstruct;
-
-import org.springframework.stereotype.Component;
-
-import com.persagy.ibms.core.data.SceneDataValue;
-import com.persagy.ibms.core.util.ComputeUtil;
-import com.persagy.ibms.core.util.PathUtil;
-import com.persagy.ibms.data.sdk.util.RepositoryContainer;
-import com.persagy.ibms.data.sdk.util.RepositoryImpl;
-
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@Component
-public class RealtimeComputeThread extends Thread {
-	@PostConstruct
-	public void init() {
-		new Thread(this).start();
-	}
-
-	private volatile boolean stop = false;
-
-	public void requestStop() {
-		stop = true;
-		try {
-			this.join();
-		} catch (InterruptedException e) {
-			log.error(e.getMessage(), e);
-		}
-	}
-
-	public void run() {
-		while (true) {
-			try {
-				Thread.sleep(1000L);
-			} catch (InterruptedException e) {
-				log.error(e.getMessage(), e);
-			}
-
-			if (DaemonThread.initialized) {
-				break;
-			}
-		}
-
-		while (!stop) {
-			try {
-				Thread.sleep(1000L);
-			} catch (InterruptedException e) {
-				log.error(e.getMessage(), e);
-			}
-
-			RepositoryImpl Repository = RepositoryContainer.instance;
-			if (Repository.enable_factor) {
-				int WaitCompute_size = Repository.WaitCompute.size();
-				if (WaitCompute_size > 0) {
-					log.warn("WaitCompute: " + WaitCompute_size);
-				}
-				while (true) {
-					SceneDataValue sdv = Repository.WaitCompute.poll();
-					if (sdv == null) {
-						break;
-					}
-					try {
-						log.warn(PathUtil.getDataPath(sdv));
-						ComputeUtil.computeProperty(Repository, sdv);
-						Repository.addWaitCompute(sdv);
-					} catch (Exception e) {
-						e.printStackTrace();
-					}
-				}
-			}
-		}
-	}
-}

+ 6 - 4
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/service/RefreshThread.java

@@ -119,12 +119,14 @@ public class RefreshThread extends Thread {
 
 				try {
 					log.warn("recompute begin");
-					RepositoryImpl Repository = new RepositoryImpl();
-					Repository.use_thread = Constant.use_thread;
-					Repository.enable_factor = Constant.enable_factor;
-					Repository.RepositoryBase = RepositoryContainer.RepositoryBase;
+					RepositoryImpl Repository = new RepositoryImpl(RepositoryContainer.RepositoryBase, Constant.use_thread, Constant.enable_factor,
+							Constant.compute_thread_count, Constant.compute_thread_sdv_interval);
 					RWDRepositoryUtil.DoRepository(Repository, changed_physical_world || changed_zkt_physical_world || changed_ibms_physical_world);
+					RepositoryImpl RepositoryOld = RepositoryContainer.instance;
 					RepositoryContainer.instance = Repository;
+					RepositoryOld.threadStop();
+					Repository.threadStart();
+					Repository.recomputeIOT();
 					log.warn("recompute finish");
 					DaemonThread.path_config = this.path_config;
 					DaemonThread.path_physical_world = this.path_physical_world;

+ 2 - 4
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/test/CheckTest.java

@@ -22,10 +22,8 @@ import lombok.extern.slf4j.Slf4j;
 public class CheckTest {
 
 	public static void test(String[] args) throws Exception {
-		RepositoryImpl Repository = new RepositoryImpl();
-		Repository.use_thread = Constant.use_thread;
-		Repository.enable_factor = Constant.enable_factor;
-		Repository.RepositoryBase = RepositoryContainer.RepositoryBase;
+		RepositoryImpl Repository = new RepositoryImpl(RepositoryContainer.RepositoryBase, Constant.use_thread, Constant.enable_factor,
+				Constant.compute_thread_count, Constant.compute_thread_sdv_interval);
 		RWDRepositoryUtil.DoRepository(Repository, true);
 		RepositoryContainer.instance = Repository;
 		{

+ 14 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/Constant.java

@@ -29,6 +29,9 @@ public class Constant {
 	public static boolean enable_factor = true;
 	public static boolean clear_old_dir = true;
 
+	public static int compute_thread_count;
+	public static long compute_thread_sdv_interval;
+
 	public static String sdk_version;
 	public static String rwd_url;
 	public static String zkt_monitor_url;
@@ -82,6 +85,17 @@ public class Constant {
 				Iterator iter;
 				Element element;
 
+				iter = document.selectNodes("/root/compute_thread_count").iterator();
+				if (iter.hasNext()) {
+					element = (Element) iter.next();
+					compute_thread_count = Integer.parseInt(element.getText());
+				}
+				iter = document.selectNodes("/root/compute_thread_sdv_interval").iterator();
+				if (iter.hasNext()) {
+					element = (Element) iter.next();
+					compute_thread_sdv_interval = Long.parseLong(element.getText());
+				}
+
 				iter = document.selectNodes("/root/groupCode").iterator();
 				if (iter.hasNext()) {
 					element = (Element) iter.next();

+ 2 - 0
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RWDRepositoryUtil.java

@@ -58,6 +58,8 @@ public class RWDRepositoryUtil {
 			Repository.objectArrayDic = RepositoryContainer.instance.objectArrayDic;
 			Repository.objectArrayAll = RepositoryContainer.instance.objectArrayAll;
 			Repository.object2info2point = RepositoryContainer.instance.object2info2point;
+			Repository.point2ObjectInfoList = RepositoryContainer.instance.point2ObjectInfoList;
+			Repository.set2ObjectInfoList = RepositoryContainer.instance.set2ObjectInfoList;
 			Repository.objType2id2Value = RepositoryContainer.instance.objType2id2Value;
 			Repository.relationArrayDic = RepositoryContainer.instance.relationArrayDic;
 			Repository.graphCodeDic = RepositoryContainer.instance.graphCodeDic;

+ 31 - 18
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/util/RepositoryImpl.java

@@ -1,6 +1,5 @@
 package com.persagy.ibms.data.sdk.util;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -16,7 +15,7 @@ import com.persagy.ibms.core.util.RepositoryBase;
 
 public class RepositoryImpl extends RepositoryBase {
 	public Map<String, JSONObject> general_queryMap;
-	public RepositoryProject RepositoryBase;
+	public RepositoryProject RepositoryProject;
 
 	// to do lirong
 	// 报警配置数据
@@ -68,12 +67,18 @@ public class RepositoryImpl extends RepositoryBase {
 	public SceneDataSet InfoPointListArray = new SceneDataSet(false);
 	public SceneDataSet InfoPointRelationArray = new SceneDataSet(false);
 
+	public RepositoryImpl(RepositoryProject RepositoryProject, boolean use_thread, boolean enable_factor, int thread_count,
+			long interval_between_compute) {
+		super(use_thread, enable_factor, thread_count, interval_between_compute);
+		this.RepositoryProject = RepositoryProject;
+	}
+
 	public ConcurrentHashMap<SceneDataValue, String> sdv2point() {
-		return RepositoryBase.sdv2point;
+		return RepositoryProject.sdv2point;
 	}
 
 	public ConcurrentHashMap<SceneDataValue, String> sdv2set() {
-		return RepositoryBase.sdv2set;
+		return RepositoryProject.sdv2set;
 	}
 
 	public SceneDataSet ParseSource(JSONObject descSet, String Source) {
@@ -168,7 +173,7 @@ public class RepositoryImpl extends RepositoryBase {
 		} else if (Source.equals("ibms-alarm-type")) {
 			result = this.ZKTAlarmTypeArray;
 		} else if (Source.equals("alarm")) {
-			result = this.RepositoryBase.alarmArray;
+			result = this.RepositoryProject.alarmArray;
 		} else if (Source.equals("info-point-list")) {
 			result = this.InfoPointListArray;
 		} else if (Source.equals("info-point-relation")) {
@@ -236,23 +241,31 @@ public class RepositoryImpl extends RepositoryBase {
 		this.sdv2Children.clear();
 	}
 
-	public void addWaitCompute(String point) {
-		List<ObjectInfo> ObjectInfoList = this.point2ObjectInfoList.get(point);
-		for (ObjectInfo ObjectInfo : ObjectInfoList) {
-			SceneDataValue sdv = ObjectInfo.obj.get(ObjectInfo.infoCode);
-			this.addWaitCompute(sdv);
+	public void recomputeIOT() {
+		// 加入计算队列
+		if (this.enable_factor) {
+			for (String point : RepositoryContainer.RepositoryBase.point2sdv.keySet()) {
+				SceneDataValue sdv = RepositoryContainer.RepositoryBase.point2sdv.get(point);
+				if (sdv.value_prim.value != null) {
+					this.addWaitCompute(point);
+				}
+			}
+			for (String point : RepositoryContainer.RepositoryBase.set2sdv.keySet()) {
+				SceneDataValue sdv = RepositoryContainer.RepositoryBase.set2sdv.get(point);
+				if (sdv.value_prim.value != null) {
+					this.addWaitCompute(point);
+				}
+			}
 		}
 	}
 
-	public void addWaitCompute(SceneDataValue sdv) {
-		try {
-			List<SceneDataValue> sdvAffectList = new ArrayList<SceneDataValue>();
-			this.dependency.get_after(this, sdv, sdvAffectList);
-			for (SceneDataValue sdvAffect : sdvAffectList) {
-				this.WaitCompute.offer(sdvAffect);
+	public void addWaitCompute(String point) {
+		if (this.point2ObjectInfoList.containsKey(point)) {
+			List<ObjectInfo> ObjectInfoList = this.point2ObjectInfoList.get(point);
+			for (ObjectInfo ObjectInfo : ObjectInfoList) {
+				SceneDataValue sdv = ObjectInfo.obj.get(ObjectInfo.infoCode);
+				this.addWaitCompute(sdv);
 			}
-		} catch (Exception e) {
-			e.printStackTrace();
 		}
 	}
 }

+ 19 - 10
ibms-data-sdk/src/main/java/com/persagy/ibms/data/sdk/websocket/IOTWebSocketClient.java

@@ -90,24 +90,27 @@ public class IOTWebSocketClient extends WebSocketClient {
 							if (value.endsWith(".0")) {
 								value = value.substring(0, value.length() - ".0".length());
 							}
+							Object valueNew;
 							try {
-								data.value_prim.value = Long.parseLong(value);
+								valueNew = Long.parseLong(value);
 							} catch (Exception e1) {
 								try {
-									data.value_prim.value = Double.parseDouble(value);
+									valueNew = Double.parseDouble(value);
 								} catch (Exception e) {
-									data.value_prim.value = value;
+									valueNew = value;
 								}
 							}
+							boolean valueEqual = valueNew.equals(data.value_prim.value);
+							data.value_prim.value = valueNew;
 							// 加入计算队列
-							if (Repository.enable_factor) {
+							if (!valueEqual && Repository.enable_factor) {
 								Repository.addWaitCompute(point);
 							}
 						} else if (type.equals("text")) {
 							data.value_prim.value = value;
 						}
 					} catch (Exception e) {
-						// log.info("onMessage" + "\t" + meter + "-" + funcid + "\t" + value);
+						log.error(e.getMessage(), e);
 					}
 				}
 			} else if (type.equals("pointset")) {
@@ -123,6 +126,9 @@ public class IOTWebSocketClient extends WebSocketClient {
 					String point = meter + "-" + funcid;
 					try {
 						SceneDataValue sdvInner = new SceneDataValue(null, null, null, null);
+						sdvInner.finish = true;
+						sdvInner.value_prim = new SceneDataPrimitive();
+						sdvInner.value_prim.change = true;
 						SceneDataValue exist_sdv = RepositoryContainer.RepositoryBase.set2sdv.putIfAbsent(point, sdvInner);
 						if (exist_sdv == null) {
 							RepositoryContainer.RepositoryBase.sdv2set.putIfAbsent(sdvInner, point);
@@ -131,21 +137,24 @@ public class IOTWebSocketClient extends WebSocketClient {
 						if (value.endsWith(".0")) {
 							value = value.substring(0, value.length() - ".0".length());
 						}
+						Object valueNew;
 						try {
-							data.value_prim.value = Long.parseLong(value);
+							valueNew = Long.parseLong(value);
 						} catch (Exception e1) {
 							try {
-								data.value_prim.value = Double.parseDouble(value);
+								valueNew = Double.parseDouble(value);
 							} catch (Exception e) {
-								data.value_prim.value = value;
+								valueNew = value;
 							}
 						}
+						boolean valueEqual = valueNew.equals(data.value_prim.value);
+						data.value_prim.value = valueNew;
 						// 加入计算队列
-						if (Repository.enable_factor) {
+						if (!valueEqual && Repository.enable_factor) {
 							Repository.addWaitCompute(point);
 						}
 					} catch (Exception e) {
-						// log.info("onMessage" + "\t" + meter + "-" + funcid + "\t" + value);
+						log.error(e.getMessage(), e);
 					}
 				}
 			}

+ 3 - 0
ibms-data-sdk/src/main/resources/config.xml

@@ -1,5 +1,8 @@
 <?xml version="1.0" encoding="utf-8"?>
 <root>
+	<compute_thread_count>16</compute_thread_count>
+	<compute_thread_sdv_interval>10</compute_thread_sdv_interval>
+
 	<groupCode>WD</groupCode>
 	<projectId>Pj4403070003</projectId>
 	<sdk_version>SDK-D</sdk_version>