Browse Source

1、添加压缩标记。

Jay Lee 6 years ago
parent
commit
c526682f79

+ 6 - 1
collect/src/main/java/com/saga/main/Main.java

@@ -140,7 +140,12 @@ public class Main {
 		}
 
 		try {
-			new CollectThread(Integer.valueOf(Constant.TypePort.get("Down"))).start();
+			String ports = Constant.TypePort.get("Down");
+			new CollectThread(Integer.valueOf(ports)).start();
+			// for (String p : ports.split(",")) {
+			// new CollectThread(Integer.valueOf(p)).start();
+			// }
+
 			Thread.sleep(3L);
 		} catch (Exception e) {
 			// TODO Auto-generated catch block

+ 4 - 0
collect/src/main/java/com/saga/thread/common/LoadConfigThread.java

@@ -78,6 +78,10 @@ public class LoadConfigThread extends Thread {
 					for (int i = 0; i < port.size(); i++) {
 						JsonObject subObject = port.get(i).getAsJsonObject();
 						Constant.TypePort.put(subObject.get("type").getAsString(), subObject.get("port").getAsString());
+						Constant.PortCompress.put(Integer.valueOf(subObject.get("port").getAsString()),
+								subObject.get("Compress").getAsString());
+						Constant.PortHistory.put(Integer.valueOf(subObject.get("port").getAsString()),
+								subObject.get("History").getAsString());
 					}
 
 				} catch (Exception e) {

+ 26 - 5
collect/src/main/java/com/saga/thread/down/CollectThread.java

@@ -29,15 +29,36 @@ public class CollectThread extends Thread {
 		try {
 			try {
 				this.ip = Constant.getLocalHostLANAddress().getHostAddress() + ":" + Constant.TypePort.get("Down");
+
 			} catch (Exception e) {
 				// TODO Auto-generated catch block
 				this.ip = "0.0.0.0" + ":" + Constant.TypePort.get("Down");
+
 				e.printStackTrace();
 			}
 			this.port = port;
-			this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", 1000));
-			this.serverList.put("TCP",
-					new TCPServerManager("0.0.0.0", this.port, "utf-8", 1000, (byte) '(', (byte) ')'));
+			if (Constant.PortCompress == null || !Constant.PortCompress.containsKey(this.port)) {
+				this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", false, 1000));
+				this.serverList.put("TCP",
+						new TCPServerManager("0.0.0.0", this.port, "utf-8", false, 1000, (byte) '(', (byte) ')'));
+			} else {
+				if ("true".equalsIgnoreCase(Constant.PortCompress.get(this.port))) {
+					// this.serverList.put("UDP", new
+					// UDPServerManager("0.0.0.0",
+					// this.port, "utf-8", 1000));
+					this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", true, 1000));
+					this.serverList.put("TCP",
+							new TCPServerManager("0.0.0.0", this.port, "utf-8", true, 1000, (byte) '(', (byte) ')'));
+				} else {
+					// this.serverList.put("UDP", new
+					// UDPServerManager("0.0.0.0",
+					// this.port, "utf-8", 1000));
+					this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", false, 1000));
+					this.serverList.put("TCP",
+							new TCPServerManager("0.0.0.0", this.port, "utf-8", false, 1000, (byte) '(', (byte) ')'));
+				}
+			}
+
 		} catch (Exception e) {
 			LogUtil.error("CollectThread.Exception: " + e.toString());
 			LogUtil.error("CollectThread(" + port + ") failed to excute, exit!");
@@ -99,11 +120,11 @@ public class CollectThread extends Thread {
 									// ps.buildingSign);
 									// 1101070037;1;senddownset;;123;1001;11;3.1
 									StringBuffer sb = new StringBuffer();
-									
+
 									sb.append(ps.buildingSign).append(";").append(ps.mac).append(";senddownset;;")
 											.append(ps.ID).append(";").append(ps.meterSign).append(";")
 											.append(ps.funcID).append(";").append(ps.dataSet).append(";");
-									
+
 									String cmd = sb.toString();
 
 									String id = ps.buildingSignUP + "-" + ps.mac + "-" + ps.ID;

+ 15 - 1
collect/src/main/java/com/saga/util/Constant.java

@@ -44,9 +44,12 @@ public class Constant {
 	// public static Map<String, String> CollectorPointSetList = new
 	// HashMap<String, String>();
 	public static Map<String, String> BuildingDB = new HashMap<String, String>();
+	public static String defaultDB;
 	public static Map<String, String> BuildingConv = new HashMap<String, String>();
 	public static Map<String, List<String>> BuildingAdds = new HashMap<String, List<String>>();
 	public static Map<String, String> TypePort = new HashMap<String, String>();
+	public static Map<Integer, String> PortCompress = new HashMap<Integer, String>();
+	public static Map<Integer, String> PortHistory = new HashMap<Integer, String>();
 	public static Map<String, Integer> RecordAddress = new HashMap<String, Integer>();
 	public static List<Record> RecordList = new ArrayList<Record>();
 	public static Map<String, Long> nodeList = new HashMap<String, Long>();
@@ -88,6 +91,7 @@ public class Constant {
 				if (subObject.get("building") != null) {
 					String building = subObject.get("building").getAsString();
 					Constant.BuildingDB.put(building, subObject.get("database").getAsString());
+					Constant.defaultDB = subObject.get("database").getAsString();
 
 					if (subObject.get("buildingadds") != null) {
 						JsonArray buildingAdds = subObject.get("buildingadds").getAsJsonArray();
@@ -99,6 +103,7 @@ public class Constant {
 							Constant.BuildingAdds.get(building).add(addObject.get("building").getAsString());
 							Constant.BuildingDB.put(addObject.get("building").getAsString(),
 									subObject.get("database").getAsString());
+							Constant.defaultDB = subObject.get("database").getAsString();
 							Constant.BuildingConv.put(addObject.get("building").getAsString(), building);
 						}
 
@@ -110,7 +115,16 @@ public class Constant {
 			JsonArray port = object.get("Port").getAsJsonArray();
 			for (int i = 0; i < port.size(); i++) {
 				JsonObject subObject = port.get(i).getAsJsonObject();
-				Constant.TypePort.put(subObject.get("type").getAsString(), subObject.get("port").getAsString());
+				String ports = subObject.get("port").getAsString();
+				Constant.TypePort.put(subObject.get("type").getAsString(), ports);
+				Constant.PortCompress.put(Integer.valueOf(ports), subObject.get("Compress").getAsString());
+				Constant.PortHistory.put(Integer.valueOf(ports), subObject.get("History").getAsString());
+				// for (String p : ports.split(",")) {
+				// Constant.PortCompress.put(Integer.valueOf(p),
+				// subObject.get("Compress").getAsString());
+				// Constant.PortHistory.put(Integer.valueOf(p),
+				// subObject.get("History").getAsString());
+				// }
 			}
 
 		} catch (Exception e) {

+ 46 - 9
collect/src/main/java/com/saga/util/HBaseUtil.java

@@ -465,6 +465,9 @@ public class HBaseUtil {
 
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();
@@ -496,6 +499,9 @@ public class HBaseUtil {
 
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();
@@ -528,6 +534,9 @@ public class HBaseUtil {
 
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();
@@ -563,6 +572,9 @@ public class HBaseUtil {
 
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();
@@ -596,6 +608,9 @@ public class HBaseUtil {
 
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();
@@ -679,6 +694,17 @@ public class HBaseUtil {
 	public static void BatchInsert(ZillionAgent agent, String DB, List<Record> dataList) throws Exception {
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			boolean history = false;
+			for (String h : Constant.PortHistory.values()) {
+				if ("true".equalsIgnoreCase(h)) {
+					history = true;
+				} else {
+					history = false;
+				}
+			}
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null && dataList.size() > 0) {
 				for (int i = 0; i < dataList.size(); i++) {
 
@@ -706,15 +732,17 @@ public class HBaseUtil {
 					wrapperObject.put("Datatable", "original_month");
 					wrapperObject.put("InsertObjects", InsertObjects);
 					agent.Query(wrapperObject);
-					wrapperObject = new JSONObject();
-					wrapperObject.put("QueryType", "batch_put");
-					wrapperObject.put("Database", DB);
-					wrapperObject.put("Datatable", "original_present");
-					wrapperObject.put("InsertObjects", InsertObjects);
-					agent.Query(wrapperObject);
-					// LogUtil.info("insert into " + DB + "." +
-					// "original_month&" + "original_present "
-					// + InsertObjects.toFormatString());
+					if (!history) {
+						wrapperObject = new JSONObject();
+						wrapperObject.put("QueryType", "batch_put");
+						wrapperObject.put("Database", DB);
+						wrapperObject.put("Datatable", "original_present");
+						wrapperObject.put("InsertObjects", InsertObjects);
+						agent.Query(wrapperObject);
+						// LogUtil.info("insert into " + DB + "." +
+						// "original_month&" + "original_present "
+						// + InsertObjects.toFormatString());
+					}
 					LogUtil.info(
 							"insert into " + DB + "." + "original_month&" + "original_present " + InsertObjects.size());
 				}
@@ -729,6 +757,9 @@ public class HBaseUtil {
 	public static void BatchInsert_Set1(ZillionAgent agent, String DB, PointSet ps) throws Exception {
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();
@@ -774,6 +805,9 @@ public class HBaseUtil {
 	public static void BatchInsert_Set2(ZillionAgent agent, String DB, PointSet ps) throws Exception {
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();
@@ -820,6 +854,9 @@ public class HBaseUtil {
 	public static void BatchInsert_Set3(ZillionAgent agent, String DB, PointSet ps) throws Exception {
 		JSONArray InsertObjects = new JSONArray();
 		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
 			if (DB != null) {
 
 				JSONObject InsertObject = new JSONObject();