Browse Source

功能添加:添加getdbpointlist及updatedbpointlist报文兼容,未来蓝天项目涉及该协议。

Jay Lee 6 năm trước cách đây
mục cha
commit
0f1a6f3a23

+ 37 - 4
collect/src/main/java/com/saga/thread/down/CollectThread.java

@@ -2,6 +2,7 @@ package com.saga.thread.down;
 
 
 import java.util.Date;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Map.Entry;
 
 
@@ -395,8 +396,10 @@ public class CollectThread extends Thread {
 						this.updateMeterCollectorControl(rList[0], meterList, p.address);
 						this.updateMeterCollectorControl(rList[0], meterList, p.address);
 					} else if (rList.length > 6 && "pointread".equalsIgnoreCase(rList[2])) {
 					} else if (rList.length > 6 && "pointread".equalsIgnoreCase(rList[2])) {
 						// 1101070037;1;pointread;;123;1001;11
 						// 1101070037;1;pointread;;123;1001;11
-						//String key = rList[0] + "-" + rList[1] + "-" + rList[5] + "." + rList[6];
-						//String id = rList[0] + "-" + rList[1] + "-" + rList[4];
+						// String key = rList[0] + "-" + rList[1] + "-" +
+						// rList[5] + "." + rList[6];
+						// String id = rList[0] + "-" + rList[1] + "-" +
+						// rList[4];
 						// Constant.id2time.put(key,
 						// Constant.id2time.put(key,
 						// System.currentTimeMillis());
 						// System.currentTimeMillis());
 						// Constant.id2point.put(id, key);
 						// Constant.id2point.put(id, key);
@@ -408,8 +411,10 @@ public class CollectThread extends Thread {
 						server.AppendToSend(p.address, new Packet(sb1.toString()));
 						server.AppendToSend(p.address, new Packet(sb1.toString()));
 						LogUtil.info(server.Name() + " AppendToSend: " + p.address + " " + sb.toString());
 						LogUtil.info(server.Name() + " AppendToSend: " + p.address + " " + sb.toString());
 					} else if (rList.length > 6 && "realtimepointdata".equalsIgnoreCase(rList[2])) {
 					} else if (rList.length > 6 && "realtimepointdata".equalsIgnoreCase(rList[2])) {
-						//String key = rList[0] + "-" + rList[1] + "-" + rList[5] + "." + rList[6];
-						//String id = rList[0] + "-" + rList[1] + "-" + rList[4];
+						// String key = rList[0] + "-" + rList[1] + "-" +
+						// rList[5] + "." + rList[6];
+						// String id = rList[0] + "-" + rList[1] + "-" +
+						// rList[4];
 						// Constant.id2time.put(key,
 						// Constant.id2time.put(key,
 						// System.currentTimeMillis());
 						// System.currentTimeMillis());
 						// Constant.id2point.put(id, key);
 						// Constant.id2point.put(id, key);
@@ -420,6 +425,34 @@ public class CollectThread extends Thread {
 						sb1 = this.getCacheData2(rList);
 						sb1 = this.getCacheData2(rList);
 						server.AppendToSend(p.address, new Packet(sb1.toString()));
 						server.AppendToSend(p.address, new Packet(sb1.toString()));
 						LogUtil.info(server.Name() + " AppendToSend: " + p.address + " " + sb.toString());
 						LogUtil.info(server.Name() + " AppendToSend: " + p.address + " " + sb.toString());
+					} else if ("getdbpointlist".equalsIgnoreCase(rList[2])) {
+						// 1101070037;1;getdbpointlist
+						// 1101070037;1;getdbpointlistack;1001;2;0;1
+						String building = rList[0];
+						for (Entry<String, List<Integer>> mfs : HBaseUtil.GetDBPointList(Constant.agent,
+								Constant.BuildingDB.get(building), "dy_pointlist", building).entrySet()) {
+							StringBuffer sb1 = new StringBuffer();
+							sb1.append(building).append(";").append(rList[1]).append(";").append("getdbpointlistack")
+									.append(";").append(mfs.getKey()).append(";").append(mfs.getValue().size());
+							for (Integer f : mfs.getValue()) {
+								sb1.append(";").append(f);
+							}
+
+							server.AppendToSend(p.address, new Packet(sb1.toString()));
+							LogUtil.info(server.Name() + " AppendToSend: " + p.address + " " + sb.toString());
+						}
+
+					} else if ("updatedbpointlist".equalsIgnoreCase(rList[2])) {
+						// 1101070037;1;updatedbpointlist;1001;2;0;Acc;1;Inst
+						String building = rList[0];
+						String meter = rList[3];
+						int count = Integer.valueOf(rList[4]);
+						Map<Integer, String> funcidList = new HashMap<Integer, String>();
+						for (int i = 0; i < count; i++) {
+							funcidList.put(Integer.valueOf(rList[5 + 2 * i]), rList[6 + 2 * i]);
+						}
+						HBaseUtil.InsertDBPointList(Constant.agent, Constant.BuildingDB.get(building), building, meter,
+								funcidList);
 					}
 					}
 				}
 				}
 				LogUtil.info("CollectThread Challenge Constant.CollectReportList release...");
 				LogUtil.info("CollectThread Challenge Constant.CollectReportList release...");

+ 60 - 0
collect/src/main/java/com/saga/util/HBaseUtil.java

@@ -1,8 +1,12 @@
 package com.saga.util;
 package com.saga.util;
 
 
 import java.io.File;
 import java.io.File;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.List;
 import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 
 
 import org.zillion.util.log.LogUtil;
 import org.zillion.util.log.LogUtil;
 
 
@@ -76,6 +80,36 @@ public class HBaseUtil {
 		LogUtil.info("Reload Project:" + project + "  PointStatus End...");
 		LogUtil.info("Reload Project:" + project + "  PointStatus End...");
 	}
 	}
 
 
+	public static Map<String, List<Integer>> GetDBPointList(ZillionAgent agent, String db, String tb, String building)
+			throws Exception {
+
+		Map<String, List<Integer>> dbPointList = new HashMap<String, List<Integer>>();
+		JSONObject Criteria = new JSONObject();
+		Criteria.put("building", building);
+
+		JSONObject wrapperObject = new JSONObject();
+		wrapperObject.put("QueryType", "select");
+		wrapperObject.put("Database", db);
+		wrapperObject.put("Datatable", tb);
+		wrapperObject.put("Criteria", Criteria);
+		JSONObject queryResult = agent.Query(wrapperObject);
+		JSONArray resultContent = (JSONArray) queryResult.get("Content");
+
+		for (int i = 0; i < resultContent.size(); i++) {
+			JSONObject dataItem = (JSONObject) resultContent.get(i);
+
+			String meter = (String) dataItem.get("meter");
+			Integer funcid = (Integer) dataItem.get("funcid");
+			if (!dbPointList.containsKey(meter)) {
+				dbPointList.put(meter, new ArrayList<Integer>());
+			}
+			dbPointList.get(meter).add(funcid);
+
+		}
+		return dbPointList;
+
+	}
+
 	public static void Getset(ZillionAgent agent, String db, String tb, String building, String meter, long funcid,
 	public static void Getset(ZillionAgent agent, String db, String tb, String building, String meter, long funcid,
 			String query_from, String query_to) throws Exception {
 			String query_from, String query_to) throws Exception {
 
 
@@ -567,6 +601,32 @@ public class HBaseUtil {
 		}
 		}
 	}
 	}
 
 
+	public static void InsertDBPointList(ZillionAgent agent, String DB, String building, String meter,
+			Map<Integer, String> funcidList) throws Exception {
+		JSONArray InsertObjects = new JSONArray();
+		for (Entry<Integer, String> f : funcidList.entrySet()) {
+			JSONObject InsertObject = new JSONObject();
+			InsertObject.put("building", building);
+			InsertObject.put("meter", meter);
+			InsertObject.put("funcid", (long) f.getKey());
+			InsertObject.put("data_type", f.getValue());
+			InsertObject.put("collect_cycle", "15min");
+			InsertObjects.add(InsertObject);
+		}
+
+		if (InsertObjects.size() > 0) {
+
+			JSONObject wrapperObject = new JSONObject();
+			wrapperObject = new JSONObject();
+			wrapperObject.put("QueryType", "batch_insert");
+			wrapperObject.put("Database", DB);
+			wrapperObject.put("Datatable", "dy_pointlist");
+			wrapperObject.put("InsertObjects", InsertObjects);
+			LogUtil.info("InsertDBPointList" + wrapperObject);
+			agent.Query(wrapperObject);
+		}
+	}
+
 	public static void InsertPointInfo(ZillionAgent agent, String DB, String project, String node, String point,
 	public static void InsertPointInfo(ZillionAgent agent, String DB, String project, String node, String point,
 			String BusNo, String site) throws Exception {
 			String BusNo, String site) throws Exception {