Browse Source

V2.0点位日志结构

Jay Lee 6 years ago
parent
commit
4ff2f4b9bd

+ 6 - 2
collect/config.json

@@ -51,11 +51,15 @@
     "Port": [
         {
             "type": "Up", 
-            "port": "9999"
+            "port": "9999",
+            "Compress":"false",
+            "History":"false"
         }, 
         {
             "type": "Down", 
-            "port": "80"
+            "port": "80" ,
+            "Compress":"false",
+            "History":"false"
         }
     ]
 }

+ 16 - 0
collect/src/main/java/com/saga/entity/EntityPoint.java

@@ -0,0 +1,16 @@
+package com.saga.entity;
+
+public class EntityPoint {
+	public String building;
+	public String meter;
+	public long funcid;
+	public String system;
+	public String source;
+	public String address_1;
+	public String address_2;
+	public String address_3;
+	public String address_4;
+	public String data_type;
+	public String collect_cycle;
+	public String unit;
+}

+ 99 - 40
collect/src/main/java/com/saga/main/Main.java

@@ -37,6 +37,13 @@ public class Main {
 
 	private static void threadStart() {
 		// TODO Auto-generated method stub
+		// 点位状态监控
+		try {
+			new CheckThread(Constant.agent).start();
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.CheckThread: " + e.toString());
+		}
 		// 配置实时加载进程
 		try {
 			new LoadConfigThread().start();
@@ -101,13 +108,7 @@ public class Main {
 			// TODO Auto-generated catch block
 			LogUtil.error("Main.WTDBThread: " + e.toString());
 		}
-		// 点位状态监控
-		try {
-			new CheckThread(Constant.agent).start();
-		} catch (Exception e) {
-			// TODO Auto-generated catch block
-			LogUtil.error("Main.CheckThread: " + e.toString());
-		}
+
 		// 统计进程
 		try {
 			new StatsThread().start();
@@ -126,7 +127,7 @@ public class Main {
 			initDB();
 			// 加载DB配置
 			loadDB();
-			// 加载Config文件配置
+			//// 加载Config文件配置
 			// loadFileConfig();
 		} catch (Exception e) {
 			// TODO Auto-generated catch block
@@ -163,17 +164,71 @@ public class Main {
 
 	private static void loadDB() {
 		// TODO Auto-generated method stub
-		// 重新启动加载节点、点位历史列表
+		// // V1.0点位日志结构
+		// {
+		// loadStatus();
+		// }
+		// V2.0点位日志结构
+		{
+			// 先加载状态再加载列表,如无状态记录则根据列表生成null
+			loadStatus2();
+			loadPointList2();
+
+		}
+
+	}
+
+	private static void loadPointList2() {
+		// TODO Auto-generated method stub
+		// 重新启动加载点位列表
+		for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+			try {
+				String project = bd.getKey();
+				String db = bd.getValue();
+				HBaseUtil.LoadDB2Cache_dy_pointlist(Constant.agent, db, "dy_pointlist", project);
+
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+				LogUtil.error("Main.loadPointList2: " + e.toString());
+			}
+		}
+	}
+
+//	private static void loadStatus() {
+//		// TODO Auto-generated method stub
+//		// 重新启动加载节点、点位状态
+//		for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+//			try {
+//				String project = bd.getKey();
+//				String db = bd.getValue();
+//				HBaseUtil.GetCollectorStatus(Constant.agent, db, "pointgatewaystatus", project);
+//				HBaseUtil.GetPointStatus(Constant.agent, db, "pointstatus", project);
+//			} catch (Exception e) {
+//				// TODO Auto-generated catch block
+//				e.printStackTrace();
+//				LogUtil.error("Main.loadStatus: " + e.toString());
+//			}
+//		}
+//	}
+
+	private static void loadStatus2() {
+		// TODO Auto-generated method stub
+		// 重新启动加载节点、点位状态
 		for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
 			try {
 				String project = bd.getKey();
 				String db = bd.getValue();
-				HBaseUtil.GetCollectorStatus(Constant.agent, db, "pointgatewaystatus", project);
-				HBaseUtil.GetPointStatus(Constant.agent, db, "pointstatus", project);
+				HBaseUtil.LoadDB2Cache_ps_physical_status(Constant.agent, db, "ps_physical_status", project);
+				HBaseUtil.LoadDB2Cache_ps_virtual_meter_status(Constant.agent, db, "ps_virtual_meter_status", project);
+				HBaseUtil.LoadDB2Cache_ps_virtual_point_status(Constant.agent, db, "ps_virtual_point_status", project);
+				HBaseUtil.LoadDB2Cache_ps_virtual_project_status(Constant.agent, db, "ps_virtual_project_status",
+						project);
+
 			} catch (Exception e) {
 				// TODO Auto-generated catch block
 				e.printStackTrace();
-				LogUtil.error("Main.loadDB: " + e.toString());
+				LogUtil.error("Main.loadStatus2: " + e.toString());
 			}
 		}
 	}
@@ -189,21 +244,22 @@ public class Main {
 			try {
 
 				createDB(db);
-				// V1.0点位日志结构
-				{
-					Thread.sleep(Constant.Millisecond.SECOND_1);
-					// dropTB(db);
-					// Thread.sleep(Constant.Millisecond.SECOND_1);
-					createTB(db);
-				}
 
-				// // V2.0点位日志结构
+				// // V1.0点位日志结构
 				// {
-				// // dropTB2(db);
+				// Thread.sleep(Constant.Millisecond.SECOND_1);
+				// // dropTB(db);
 				// // Thread.sleep(Constant.Millisecond.SECOND_1);
-				// createTB2(db);
+				// createTB(db);
 				// }
 
+				// V2.0点位日志结构
+				{
+					// dropTB2(db);
+					// Thread.sleep(Constant.Millisecond.SECOND_1);
+					createTB2(db);
+				}
+
 				Thread.sleep(Constant.Millisecond.SECOND_1);
 			} catch (InterruptedException e) {
 				// TODO Auto-generated catch block
@@ -227,6 +283,9 @@ public class Main {
 			// dy_pointlist
 			HBaseUtil.CreateTable_dy_pointlist(Constant.agent, db, "dy_pointlist");
 			Thread.sleep(Constant.Millisecond.SECOND_3);
+			// fjd_0_buildingcomputetime
+			HBaseUtil.CreateTable_fjd_0_buildingcomputetime(Constant.agent, db, "fjd_0_buildingcomputetime");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
 
 			// ps_physical_status
 			HBaseUtil.CreateTable_ps_physical_status(Constant.agent, db, "ps_physical_status");
@@ -270,17 +329,17 @@ public class Main {
 	private static void dropTB2(String db) {
 		// TODO Auto-generated method stub
 		try {
-			Constant.agent.DDL_DropTable(db, "original_month");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
-			Constant.agent.DDL_DropTable(db, "original_present");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
-			Constant.agent.DDL_DropTable(db, "originalset");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
-			Constant.agent.DDL_DropTable(db, "originalset_present");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
-			// dy_pointlist
-			Constant.agent.DDL_DropTable(db, "dy_pointlist");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "original_month");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "original_present");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset_present");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// // dy_pointlist
+			// Constant.agent.DDL_DropTable(db, "dy_pointlist");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
 
 			// ps_physical_status
 			Constant.agent.DDL_DropTable(db, "ps_physical_status");
@@ -350,13 +409,13 @@ public class Main {
 	private static void dropTB(String db) {
 		// TODO Auto-generated method stub
 		try {
-			Constant.agent.DDL_DropTable(db, "original_month");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
-			Constant.agent.DDL_DropTable(db, "original_present");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
-			Constant.agent.DDL_DropTable(db, "originalset");
-			Thread.sleep(Constant.Millisecond.SECOND_3);
-			Constant.agent.DDL_DropTable(db, "originalset_present");
+			// Constant.agent.DDL_DropTable(db, "original_month");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "original_present");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset_present");
 			Thread.sleep(Constant.Millisecond.SECOND_3);
 			Constant.agent.DDL_DropTable(db, "collectorstatus");
 			Thread.sleep(Constant.Millisecond.SECOND_3);

+ 85 - 0
collect/src/main/java/com/saga/test/Simulator.java

@@ -0,0 +1,85 @@
+package com.saga.test;
+
+import org.zillion.util.log.LogUtil;
+
+import com.persagy.communication.entity.Packet;
+import com.persagy.communication.entity.PacketEntity;
+import com.persagy.communication.mina.udp.server.UDPServerManager;
+import com.persagy.communication.util.IServerManager;
+import com.saga.util.Constant;
+
+public class Simulator {
+
+	private static final String MAC = "AC-CF-23-E9-3C-22";
+	private static long needSend = 0l;
+	private static IServerManager simulator = new UDPServerManager("0.0.0.0", 9756, "utf-8", 1000);
+
+	private static boolean needSend() {
+		if (System.currentTimeMillis() - needSend > Constant.Millisecond.MINUTE_1) {
+			needSend = System.currentTimeMillis();
+			return true;
+		}
+		return false;
+	}
+
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+
+		simulator.Start();
+		simulator.AllowJustSend_UDP(true);
+		while (true) {
+			PacketEntity p;
+			p = simulator.PopRece();
+			returnReport(p);
+			sendReport();
+
+		}
+	}
+
+	private static void returnReport(PacketEntity p) {
+		// TODO Auto-generated method stub
+		try {
+			if ((p != null) && (p.content != null) && (p.content.packetString != null)
+					&& (p.content.packetString.length() > 0)) {
+				LogUtil.info(simulator.Name() + " PopRece:" + p.address + " " + p.content.packetString);
+				String[] rList = p.content.packetString.replace("(", "").replace(")", "").replace(" ", "").split(";");
+				// (00-50-56-C0-00-08; control;1255581314001;2;21;0;22;1;)
+				// (00-50-56-C0-00-08;controlack;1255581314001;1;)
+				if ("control".equalsIgnoreCase(rList[1])) {
+					simulator.AppendToSend(p.address,
+							new Packet("(" + rList[0] + ";" + "controlack" + ";" + rList[2] + ";1;" + ")"));
+					LogUtil.info(simulator.Name() + " AppendToSend:" + p.address + " " + "(" + rList[0] + ";"
+							+ "reportack" + ";" + rList[2] + ")");
+					// (00-50-56-C0-00-08; get;1255581314001;)
+					// (00-50-56-C0-00-08;getack;1255581314001;9;11;22.3;12;33.3;13;32.3;14;22.2;
+					// 21;1;22;0;23;0;24;0;25;1)
+				} else if ("get".equalsIgnoreCase(rList[1])) {
+					simulator.AppendToSend(p.address, new Packet("(" + rList[0] + ";" + "getack" + ";" + rList[2] + ";"
+							+ "9;11;22.3;12;33.3;13;32.3;14;22.2;21;1;22;0;23;0;24;0;25;1)"));
+					LogUtil.info(simulator.Name() + " AppendToSend:" + p.address + " " + "(" + rList[0] + ";" + "getack"
+							+ ";" + rList[2] + ";" + "9;11;22.3;12;33.3;13;32.3;14;22.2;21;1;22;0;23;0;24;0;25;1)");
+
+				}
+			}
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+			LogUtil.error("CollectSenddownThread.handlePacket.Exception: " + e.toString());
+		}
+	}
+
+	private static void sendReport() {
+		// TODO Auto-generated method stub
+		if (needSend()) {
+			StringBuffer sb = new StringBuffer();
+			sb.append("(");
+			sb.append(MAC);
+			sb.append(";report;");
+			sb.append(System.currentTimeMillis());
+			sb.append(";9;11;22.3;12;33.3;13;32.3;14;22.2;21;1;22;0;23;0;24;0;25;1)");
+			simulator.AppendToSend("/192.168.20.159:8756", new Packet(sb.toString()));
+			LogUtil.info(simulator.Name() + " AppendToSend:" + "/192.168.20.159:80" + " " + sb.toString());
+		}
+	}
+
+}

+ 502 - 136
collect/src/main/java/com/saga/test/Test.java

@@ -1,172 +1,538 @@
 package com.saga.test;
 
+import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
-import java.net.UnknownHostException;
+import java.io.FileReader;
+import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
 
 import org.apache.hadoop.conf.Configuration;
-import org.dom4j.Document;
-import org.dom4j.DocumentException;
-import org.dom4j.Element;
-import org.dom4j.Node;
-import org.dom4j.io.SAXReader;
-
 import org.zillion.util.log.LogUtil;
 
+import com.persagy.communication.mina.tcp.server.TCPServerManager;
+import com.persagy.communication.mina.udp.server.UDPServerManager;
+import com.persagy.communication.util.IServerManager;
+import com.saga.entity.EntityPoint;
+import com.saga.thread.common.CheckThread;
+import com.saga.thread.common.LoadConfigThread;
+import com.saga.thread.common.SaveThread;
+import com.saga.thread.common.StatsThread;
+import com.saga.thread.common.WTDBThread;
+import com.saga.thread.down.CollectThread;
+import com.saga.thread.up.SendThread;
 import com.saga.util.Constant;
 import com.saga.util.HBaseUtil;
 import com.zillion.database.agent.ZillionAgent;
 
 public class Test {
 
-	public Test() {
-		// TODO Auto-generated constructor stub
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+
+		// 初始化
+		init(args);
+		// 业务进程启动
+		threadStart();
 	}
 
-	public static void main(String[] args) {
+	private static void threadStart() {
 		// TODO Auto-generated method stub
-		System.out.println("1101080002;1;pointset;;66;666666;11024;0".length());
-		// String point = "8002.11201";
-		// System.out.println(point);
-		// String meter = point.split("\\.")[0];
-		// String funcid = (point.split("\\.")[1]);
-		// System.out.println(meter);
-		// System.out.println(funcid);
-
-		// getConfigFile(LogUtil.GetPath(), "config-", "port");
-		// System.out.println(Constant.pointInfo);
-		// for (Entry<String, String> pi : Constant.pointInfo.entrySet()) {
-		// System.out.println(pi.getKey() + " " + pi.getValue());
-		// }
+		// 点位状态监控
+		try {
+			new CheckThread(Constant.agent).start();
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.CheckThread: " + e.toString());
+		}
+		// 配置实时加载进程
+		try {
+			new LoadConfigThread().start();
+			Thread.sleep(3L);
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.LoadConfigThread: " + e.toString());
+		}
+		// 采集进程
+		try {
+			String ports = Constant.TypePort.get("Down");
+			new CollectThread(Integer.valueOf(ports)).start();
+			// for (String p : ports.split(",")) {
+			// new CollectThread(Integer.valueOf(p)).start();
+			// }
 
-		// try {
-		// System.out.println(System.currentTimeMillis());
-		// Constant.getLocalHostLANAddress();
-		// System.out.println(System.currentTimeMillis());
-		// } catch (UnknownHostException e) {
-		// // TODO Auto-generated catch block
-		// e.printStackTrace();
-		// }
-		// List<String> a = new ArrayList<String>();
-		// List<String> b = new ArrayList<String>();
-		// a.add("a");
-		// a.add("b");
-		// a.add("c");
-		// b.addAll(a);
-		// a = new ArrayList<String>();
-		// for (String s : b) {
-		// System.out.println(s);
-		// }
-		// try {
-		// String resource = "/hbase-site.xml";
-		// Configuration configuration = new Configuration();
-		//
-		// configuration.addResource(new
-		// FileInputStream(System.getProperty("user.dir") + resource));
-		// String zookeeper_connectionAddress =
-		// "zookeeper1:2181,zookeeper2:2181,zookeeper3:2181";
-		// int zookeeper_sessionTimeout = 300000;
-		// Constant.agent = new ZillionAgent(configuration,
-		// zookeeper_connectionAddress, zookeeper_sessionTimeout);
-		//
-		// Constant.agent.Start();
-		// // LogUtil.info("" + Constant.BuildingDB);
-		// // LogUtil.info("" + Constant.BuildingAdds);
-		// // LogUtil.info("" + Constant.BuildingConv);
-		//
-		// } catch (Exception e) {
-		// // TODO Auto-generated catch block
-		// LogUtil.error("Main: " + e.toString());
-		// }
-		// while (true) {
-		// // 加载采集器配置文件
-		// // 文件夹命名格式:config+"-"+楼号+"-"+mac序号,示例:config-1101080002-1
-		// Constant.getConfigFile(LogUtil.GetPath(), "config-", "port");
-		// // 采集器点位位置信息写入数据库
-		// for (Entry<String, String> pi : Constant.pointInfo.entrySet()) {
-		// String building = pi.getKey().split("-")[0];
-		// String mac = pi.getKey().split("-")[1];
-		// String node = mac;
-		// String point = pi.getKey().split("-")[2];
-		// String busno = pi.getValue().split("-")[0];
-		// String site = pi.getValue().split("-")[1];
-		// try {
-		// System.out.println(building + " " + node + " " + point + " " + busno
-		// + " " + site);
-		// // HBaseUtil.InsertPointInfo(Constant.agent,
-		// // Constant.BuildingDB.get(building), building, node, point,
-		// // busno, site);
-		// } catch (Exception e) {
-		// // TODO Auto-generated catch block
-		// e.printStackTrace();
-		// }
-		// // System.out.println(building + node + point + busno + site);
-		// }
-		// try {
-		// Thread.sleep(Constant.Millisecond.MINUTE_1);
-		// } catch (InterruptedException e) {
-		// // TODO Auto-generated catch block
-		// e.printStackTrace();
-		// }
-		// }
+			Thread.sleep(3L);
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.CollectThread: " + e.toString());
+		}
+		// 上传进程
+		List<IServerManager> udpserverList = new ArrayList<IServerManager>();
+		List<IServerManager> tcpserverList = new ArrayList<IServerManager>();
+		udpserverList.add(new UDPServerManager("0.0.0.0", Integer.valueOf(Constant.TypePort.get("Up")), "utf-8", 1000));
+		tcpserverList.add(new TCPServerManager("0.0.0.0", Integer.valueOf(Constant.TypePort.get("Up")), "utf-8", 1000));
+
+		for (IServerManager server : udpserverList) {
+			new SendThread(server).start();
+			try {
+				Thread.sleep(3L);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				LogUtil.error("Main.SendThread: " + e.toString());
+			}
+		}
+		for (IServerManager server : tcpserverList) {
+			new SendThread(server).start();
+			try {
+				Thread.sleep(3L);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				LogUtil.error("Main.SendThread: " + e.toString());
+			}
+		}
+		// 数据处理进程
+		try {
+			new SaveThread().start();
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.SaveThread: " + e.toString());
+		}
+		// 数据写库进程
+		try {
+			List<WTDBThread> wtdbthreadList = new ArrayList<WTDBThread>();
+			for (int i = 0; i < Constant.InsertThread; i++) {
+				WTDBThread thread = new WTDBThread(Constant.agent);
+				thread.start();
+				wtdbthreadList.add(thread);
+			}
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.WTDBThread: " + e.toString());
+		}
 
+		// 统计进程
+		try {
+			new StatsThread().start();
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.StatsThread: " + e.toString());
+		}
 	}
 
-	public static List<String> getConfigFile(String directoryPath, String directoryKey, String fileKey) {
-		List<String> list = new ArrayList<String>();
-		File baseFile = new File(directoryPath);
-		if (baseFile.isFile() || !baseFile.exists()) {
-			return list;
+	private static void init(String[] args) {
+		// TODO Auto-generated method stub
+		try {
+			// ZillionAgent 启动
+			agentStart(args);
+			// 初始化数据库环境
+			initDB();
+			// // 测试数据
+			testDB();
+			// 加载DB配置
+			loadDB();
+			//// 加载Config文件配置
+			// loadFileConfig();
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+			LogUtil.error("Main.init: " + e.toString());
+			System.exit(0);
 		}
+	}
+
+	// private static void loadFileConfig() {
+	// // TODO Auto-generated method stub
+	// // 加载采集器配置文件
+	// // 文件夹命名格式:config+"-"+楼号+"-"+mac序号,示例:config-1101080002-1
+	// Constant.getConfigFile(LogUtil.GetPath(), "config-", "port");
+	// // 采集器点位位置信息写入数据库
+	// for (Entry<String, String> pi : Constant.pointInfo.entrySet()) {
+	// String building = pi.getKey().split("-")[0];
+	// String mac = pi.getKey().split("-")[1];
+	// String node = mac;
+	// String point = pi.getKey().split("-")[2];
+	// String busno = pi.getValue().split("-")[0];
+	// String site = pi.getValue().split("-")[1];
+	// try {
+	// HBaseUtil.InsertPointInfo(Constant.agent,
+	// Constant.BuildingDB.get(building), building, node, point,
+	// busno, site);
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// }
+	// // System.out.println(building + node + point + busno + site);
+	// }
+	// }
+
+	private static void testDB() {
+		// TODO Auto-generated method stub
 		try {
-			File[] files = baseFile.listFiles();
-			SAXReader saxReader = new SAXReader();
-			for (File file : files) {
-
-				if (file.isDirectory() && file.getName().contains(directoryKey)) {
-					list.addAll(getConfigFile(file.getAbsolutePath(), directoryKey, fileKey));
-				} else {
-					if (file.getName().contains(fileKey)) {
-						list.add(file.getAbsolutePath());
-						Document document = saxReader.read(file);
-						List<Element> devList = document.selectNodes("/root/portdev/devtype");
-						// System.out.println(file.getName());
-						// System.out.println(file.getName().indexOf(fileKey));
-						// System.out.println(file.getName().indexOf(".xml"));
-						String port = file.getName().substring(file.getName().indexOf(fileKey) + fileKey.length(),
-								file.getName().indexOf(".xml"));
-						// System.out.println(port);
-						// System.out.println(file.getParentFile().getName());
-						String building = file.getParentFile().getName().replace(directoryKey, "").substring(0, 10);
-						String mac = file.getParentFile().getName().replace(directoryKey, "").substring(10);
-						// System.out.println(building);
-						// System.out.println(mac);
-
-						for (Element portdev : devList) {
-							String meter = portdev.attributeValue("meterid");
-							Iterator funcList = portdev.selectNodes("funset/funid").iterator();
-							while (funcList.hasNext()) {
-								Element func = (Element) funcList.next();
-								String point = building + "-" + mac + "-" + meter + "." + func.getText();
-								String info = port + "-" + portdev.element("addr").getText();
-								Constant.pointInfo.put(point, info);
-								// System.out.println(meter + "." +
-								// func.getText());
-							}
-						}
+			Map<String, List<EntityPoint>> buildingPointList = new HashMap<String, List<EntityPoint>>();
+			String resource = "\\testpointlist.csv";
+			LogUtil.info(System.getProperty("user.dir") + resource);
+			File csv = new File(System.getProperty("user.dir") + resource);
+			BufferedReader br = new BufferedReader(new FileReader(csv));
+			String line = "";
+			while ((line = br.readLine()) != null) {
+				EntityPoint ep = new EntityPoint();
+
+				ep.building = line.split(",")[0];
+				ep.meter = line.split(",")[1];
+				ep.funcid = Long.valueOf(line.split(",")[2]);
+				ep.system = line.split(",")[3];
+				ep.source = line.split(",")[4];
+				ep.address_1 = line.split(",")[5];
+				ep.address_2 = line.split(",")[6];
+				ep.address_3 = line.split(",")[7];
+				ep.address_4 = line.split(",")[8];
+				ep.data_type = line.split(",")[9];
+				ep.collect_cycle = line.split(",")[10];
+				ep.unit = line.split(",").length < 12 ? "" : line.split(",")[11];
 
+				if (!buildingPointList.containsKey(ep.building)) {
+					buildingPointList.put(ep.building, new ArrayList<EntityPoint>());
+				}
+				buildingPointList.get(ep.building).add(ep);
+			}
+			for (Entry<String, List<EntityPoint>> e : buildingPointList.entrySet()) {
+				for (EntityPoint p : e.getValue()) {
+					try {
+						HBaseUtil.Insert_dy_pointlist(Constant.agent, Constant.BuildingDB.get(p.building), p.building,
+								p.meter, p.funcid, p.system, p.source, p.address_1, "192.168.20.159", p.address_2,
+								p.address_3, p.address_4, p.collect_cycle, p.data_type, p.unit);
+					} catch (Exception e1) {
+						// TODO Auto-generated catch block
+						e1.printStackTrace();
 					}
 				}
+				LogUtil.info(" batch_insert dy_pointlist " + e.getKey() + " complete!");
+			}
+			try {
+				HBaseUtil.Insert_fjd_0_buildingcomputetime(Constant.agent, Constant.BuildingDB.get("1101080002"),
+						"1101080002", "20181101000000", "20181101000000");
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+		} catch (NumberFormatException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		} catch (IOException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	private static void loadDB() {
+		// TODO Auto-generated method stub
+		// // V1.0点位日志结构
+		// {
+		// loadStatus();
+		// }
+		// V2.0点位日志结构
+		{
+			// 先加载状态再加载列表,如无状态记录则根据列表生成null
+			loadStatus2();
+			loadPointList2();
+			// Constant.showStatus();
+		}
+	}
+
+	private static void loadPointList2() {
+		// TODO Auto-generated method stub
+		// 重新启动加载点位列表
+		for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+			try {
+				String project = bd.getKey();
+				String db = bd.getValue();
+				HBaseUtil.LoadDB2Cache_dy_pointlist(Constant.agent, db, "dy_pointlist", project);
+
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+				LogUtil.error("Main.loadPointList2: " + e.toString());
+			}
+		}
+	}
+
+	// private static void loadStatus() {
+	// // TODO Auto-generated method stub
+	// // 重新启动加载节点、点位状态
+	// for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+	// try {
+	// String project = bd.getKey();
+	// String db = bd.getValue();
+	// HBaseUtil.GetCollectorStatus(Constant.agent, db, "pointgatewaystatus",
+	// project);
+	// HBaseUtil.GetPointStatus(Constant.agent, db, "pointstatus", project);
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// LogUtil.error("Main.loadStatus: " + e.toString());
+	// }
+	// }
+	// }
+
+	private static void loadStatus2() {
+		// TODO Auto-generated method stub
+		// 重新启动加载节点、点位状态
+		for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+			try {
+				String project = bd.getKey();
+				String db = bd.getValue();
+				HBaseUtil.LoadDB2Cache_ps_physical_status(Constant.agent, db, "ps_physical_status", project);
+				HBaseUtil.LoadDB2Cache_ps_virtual_meter_status(Constant.agent, db, "ps_virtual_meter_status", project);
+				HBaseUtil.LoadDB2Cache_ps_virtual_point_status(Constant.agent, db, "ps_virtual_point_status", project);
+				HBaseUtil.LoadDB2Cache_ps_virtual_project_status(Constant.agent, db, "ps_virtual_project_status",
+						project);
+
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+				LogUtil.error("Main.loadStatus2: " + e.toString());
+			}
+		}
+	}
+
+	private static void initDB() {
+		// TODO Auto-generated method stub
+		Map<String, String> dbList = new HashMap<String, String>();
+		for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+			dbList.put(bd.getValue(), bd.getValue());
+		}
+
+		for (String db : dbList.keySet()) {
+			try {
+
+				createDB(db);
+
+				// // V1.0点位日志结构
+				// {
+				// Thread.sleep(Constant.Millisecond.SECOND_1);
+				// // dropTB(db);
+				// // Thread.sleep(Constant.Millisecond.SECOND_1);
+				// createTB(db);
+				// }
+
+				// V2.0点位日志结构
+				{
+					// dropTB2(db);
+					// Thread.sleep(Constant.Millisecond.SECOND_1);
+					createTB2(db);
+				}
+
+				Thread.sleep(Constant.Millisecond.SECOND_1);
+			} catch (InterruptedException e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+
 			}
+		}
+	}
+
+	private static void createTB2(String db) {
+		// TODO Auto-generated method stub
+		try {
+			HBaseUtil.CreateTable(Constant.agent, db, "original_month");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_present(Constant.agent, db, "original_present");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_set(Constant.agent, db, "originalset");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_setpresent(Constant.agent, db, "originalset_present");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			// dy_pointlist
+			HBaseUtil.CreateTable_dy_pointlist(Constant.agent, db, "dy_pointlist");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			// fjd_0_buildingcomputetime
+			HBaseUtil.CreateTable_fjd_0_buildingcomputetime(Constant.agent, db, "fjd_0_buildingcomputetime");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_physical_status
+			HBaseUtil.CreateTable_ps_physical_status(Constant.agent, db, "ps_physical_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_physical_log
+			HBaseUtil.CreateTable_ps_physical_log(Constant.agent, db, "ps_physical_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_point_status
+			HBaseUtil.CreateTable_ps_virtual_point_status(Constant.agent, db, "ps_virtual_point_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_point_log
+			HBaseUtil.CreateTable_ps_virtual_point_log(Constant.agent, db, "ps_virtual_point_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_meter_status
+			HBaseUtil.CreateTable_ps_virtual_meter_status(Constant.agent, db, "ps_virtual_meter_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_meter_log
+			HBaseUtil.CreateTable_ps_virtual_meter_log(Constant.agent, db, "ps_virtual_meter_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_project_status
+			HBaseUtil.CreateTable_ps_virtual_project_status(Constant.agent, db, "ps_virtual_project_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_project_log
+			HBaseUtil.CreateTable_ps_virtual_project_log(Constant.agent, db, "ps_virtual_project_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			Thread.sleep(Constant.Millisecond.SECOND_30);
 		} catch (Exception e) {
 			// TODO Auto-generated catch block
 			e.printStackTrace();
 		}
+	}
+
+	private static void dropTB2(String db) {
+		// TODO Auto-generated method stub
+		try {
+			// Constant.agent.DDL_DropTable(db, "original_month");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "original_present");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset_present");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// // dy_pointlist
+			// Constant.agent.DDL_DropTable(db, "dy_pointlist");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
 
-		return list;
+			// ps_physical_status
+			Constant.agent.DDL_DropTable(db, "ps_physical_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_physical_log
+			Constant.agent.DDL_DropTable(db, "ps_physical_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_point_status
+			Constant.agent.DDL_DropTable(db, "ps_virtual_point_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_point_log
+			Constant.agent.DDL_DropTable(db, "ps_virtual_point_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_meter_status
+			Constant.agent.DDL_DropTable(db, "ps_virtual_meter_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_meter_log
+			Constant.agent.DDL_DropTable(db, "ps_virtual_meter_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_project_status
+			Constant.agent.DDL_DropTable(db, "ps_virtual_project_status");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			// ps_virtual_project_log
+			Constant.agent.DDL_DropTable(db, "ps_virtual_project_log");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+
+			Thread.sleep(Constant.Millisecond.SECOND_30);
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	private static void createTB(String db) {
+		// TODO Auto-generated method stub
+		try {
+			HBaseUtil.CreateTable(Constant.agent, db, "original_month");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_present(Constant.agent, db, "original_present");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_set(Constant.agent, db, "originalset");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_setpresent(Constant.agent, db, "originalset_present");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_CollectorStatus(Constant.agent, db, "pointgatewaystatus");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_CollectorCommunicationLog(Constant.agent, db, "pointgatewaylog");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_PointStatus(Constant.agent, db, "pointstatus");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_PointCommunicationLog(Constant.agent, db, "pointlog");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			HBaseUtil.CreateTable_PointInfo(Constant.agent, db, "pointinfo");
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
 	}
-}
+
+	private static void dropTB(String db) {
+		// TODO Auto-generated method stub
+		try {
+			// Constant.agent.DDL_DropTable(db, "original_month");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "original_present");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset");
+			// Thread.sleep(Constant.Millisecond.SECOND_3);
+			// Constant.agent.DDL_DropTable(db, "originalset_present");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			Constant.agent.DDL_DropTable(db, "collectorstatus");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			Constant.agent.DDL_DropTable(db, "collectorcommunicationlog");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			Constant.agent.DDL_DropTable(db, "pointstatus");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			Constant.agent.DDL_DropTable(db, "pointcommunicationlog");
+			Thread.sleep(Constant.Millisecond.SECOND_3);
+			Constant.agent.DDL_DropTable(db, "pointinfo");
+			Thread.sleep(Constant.Millisecond.SECOND_30);
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	private static void createDB(String db) {
+		// TODO Auto-generated method stub
+		try {
+			HBaseUtil.CreateDB(Constant.agent, db);
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	private static void agentStart(String[] args) {
+		// TODO Auto-generated method stub
+		try {
+			String resource = "/hbase-site.xml";
+			Configuration configuration = new Configuration();
+
+			configuration.addResource(new FileInputStream(System.getProperty("user.dir") + resource));
+			// String zookeeper_connectionAddress =
+			// "zookeeper1:2181,zookeeper2:2181,zookeeper3:2181";
+
+			String zookeeper_connectionAddress = "zookeeper1,zookeeper2,zookeeper3";
+
+			int zookeeper_sessionTimeout = 300000;
+			Constant.agent = new ZillionAgent(configuration, zookeeper_connectionAddress, zookeeper_sessionTimeout);
+
+			Constant.agent.Start();
+			// LogUtil.info("" + Constant.BuildingDB);
+			// LogUtil.info("" + Constant.BuildingAdds);
+			// LogUtil.info("" + Constant.BuildingConv);
+
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("Main.init.agentStart: " + e.toString());
+		}
+	}
+
+}

+ 675 - 188
collect/src/main/java/com/saga/thread/common/CheckThread.java

@@ -18,8 +18,11 @@ public class CheckThread extends Thread {
 	ZillionAgent agent;
 	long realCheck = 0;
 	private String ip = null;
-	private Map<String, String> nodeStatus = new HashMap<String, String>();
+	// private Map<String, String> nodeStatus = new HashMap<String, String>();
 	private Map<String, String> pointStatus = new HashMap<String, String>();
+	private Map<String, String> meterStatus = new HashMap<String, String>();
+	private Map<String, String> projectStatus = new HashMap<String, String>();
+	private Map<String, String> physicalStatus = new HashMap<String, String>();
 
 	public CheckThread(ZillionAgent agent) {
 		try {
@@ -40,10 +43,22 @@ public class CheckThread extends Thread {
 
 	public void run() {
 		LogUtil.info("CheckThread-" + this.getName() + " Starting ...");
+		// try {
+		// Thread.sleep(Constant.Millisecond.MINUTE_1);
+		// } catch (Exception e) {
+		// // TODO Auto-generated catch block
+		// e.printStackTrace();
+		// }
+		// 初次加载生成当前状态
+		this.excuteDBStatus();
 
 		while (true) {
 			try {
-				this.realCheck();
+				// // V1.0点位日志结构
+				// this.realCheck();
+				// V2.0点位日志结构
+				this.realCheck2();
+				// Constant.showStatus();
 				Thread.sleep(Constant.Millisecond.SECOND_1);
 			} catch (Exception e) {
 				// TODO Auto-generated catch block
@@ -52,65 +67,213 @@ public class CheckThread extends Thread {
 		}
 	}
 
-	private void realCheck() {
+	private void excuteDBStatus() {
+		// TODO Auto-generated method stub
+		// for (Entry<String, Long> pns : Constant.ps_NodeStatus.entrySet()) {
+		// if (System.currentTimeMillis() - pns.getValue() <
+		// Constant.Millisecond.MINUTE_1) {
+		// this.nodeStatus.put(pns.getKey(), "conn");
+		// } else {
+		// this.nodeStatus.put(pns.getKey(), "disconn");
+		// }
+		// }
+		for (Entry<String, Long> ppsu : Constant.ps_PointStatusUp.entrySet()) {
+			if (ppsu.getValue() != null) {
+				if (System.currentTimeMillis() - ppsu.getValue() < Constant.Millisecond.MINUTE_10) {
+					this.pointStatus.put(ppsu.getKey(), "conn");
+				} else {
+					this.pointStatus.put(ppsu.getKey(), "disconn");
+				}
+			} else {
+				this.pointStatus.put(ppsu.getKey(), null);
+			}
+		}
+
+		for (Entry<String, Long> pps : Constant.ps_PhysicalStatus.entrySet()) {
+			// if (!pps.getKey().endsWith(";;;")) {
+			if (pps.getValue() != null) {
+				if (System.currentTimeMillis() - pps.getValue() < Constant.Millisecond.MINUTE_10) {
+					this.physicalStatus.put(pps.getKey(), "conn");
+				} else {
+					this.physicalStatus.put(pps.getKey(), "disconn");
+				}
+			} else {
+				this.physicalStatus.put(pps.getKey(), null);
+			}
+
+			// }
+		}
+		for (Entry<String, Long> pps : Constant.ps_ProjectStatus.entrySet()) {
+			if (pps.getValue() != null) {
+				if (System.currentTimeMillis() - pps.getValue() < Constant.Millisecond.MINUTE_10) {
+					this.projectStatus.put(pps.getKey(), "conn");
+				} else {
+					this.projectStatus.put(pps.getKey(), "disconn");
+				}
+			} else {
+				this.projectStatus.put(pps.getKey(), null);
+			}
+
+		}
+		for (Entry<String, Long> pmsu : Constant.ps_MeterStatusUp.entrySet()) {
+			if (pmsu.getValue() != null) {
+				if (System.currentTimeMillis() - pmsu.getValue() < Constant.Millisecond.MINUTE_10) {
+					this.meterStatus.put(pmsu.getKey(), "conn");
+				} else {
+					this.meterStatus.put(pmsu.getKey(), "disconn");
+				}
+			} else {
+				this.meterStatus.put(pmsu.getKey(), null);
+			}
+		}
+	}
+
+	private void realCheck2() {
 		// TODO Auto-generated method stub
 
 		if (System.currentTimeMillis() - this.realCheck > Constant.Millisecond.SECOND_3) {
 			this.realCheck = System.currentTimeMillis();
 
 			try {
-				// CollectorStatus
-				// LogUtil.info("Constant.nodeList:" + Constant.nodeList);
-				for (Entry<String, Long> node : Constant.ps_NodeStatus.entrySet()) {
-					try {
-						String building = node.getKey().split("-")[0];
-						String mac = node.getKey().split("-")[1];
-						String status = "";
-						if (node.getValue() != null
-								&& this.realCheck - node.getValue() < Constant.Millisecond.MINUTE_1) {
-							HBaseUtil.InsertCollectorStatus(Constant.agent, Constant.BuildingDB.get(building), building,
-									this.ip, mac, "conn", Constant.format(new Date(node.getValue())));
-							status = "conn";
-						} else {
-							HBaseUtil.InsertCollectorStatus(Constant.agent, Constant.BuildingDB.get(building), building,
-									this.ip, mac, "disconn", Constant.format(new Date(node.getValue())));
-							status = "disconn";
-						}
+				this.virtual_point_status_up_Check();
+				this.virtual_point_status_down_Check();
+				this.physical_status_address_Check();
 
-						if ((!this.nodeStatus.containsKey(node.getKey()))
-								|| (!status.equalsIgnoreCase(this.nodeStatus.get(node.getKey())))) {
-							// 添加Log记录
-							HBaseUtil.InsertCollectorCommunicationLog(Constant.agent, Constant.BuildingDB.get(building),
-									building, this.ip, mac, Constant.format(new Date(node.getValue())), (status));
-						}
-						// 更新Status
-						this.nodeStatus.put(node.getKey(), status);
-					} catch (Exception e) {
-						// TODO Auto-generated catch block
-						e.printStackTrace();
-						LogUtil.error("CheckThread.realCheck.CollectorStatus.Exception: " + e.toString());
-					}
+				this.virtual_meter_status_up_Check();
+				this.virtual_project_status_up_Check();
+
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+				LogUtil.error("CheckThread.realCheck2.Exception: " + e.toString());
+			}
+		}
+	}
+
+	private void virtual_project_status_up_Check() {
+		// TODO Auto-generated method stub
+		for (Entry<String, List<String>> ps : Constant.ps_ProjectList.entrySet()) {
+			String project = ps.getKey();
+			List<String> l = ps.getValue();
+			String status = this.gatherStatus(l);
+			try {
+				HBaseUtil.Insert_ps_virtual_project_status(Constant.agent, Constant.BuildingDB.get(project), project,
+						status, Constant.format(new Date()), "");
+				if (
+				// 首次
+				!this.projectStatus.containsKey(project) || null == this.projectStatus.get(project) ||
+				// 已有记录
+						!status.equalsIgnoreCase(this.projectStatus.get(project))) {
+
+					HBaseUtil.Insert_ps_virtual_project_log(Constant.agent, Constant.BuildingDB.get(project), project,
+							Constant.format(new Date()), status);
 				}
-				// PointStatus
-				// up
-				// LogUtil.info("Constant.pointListUp:" + Constant.pointListUp);
-				// synchronized (Constant.pointListUp) {
-				// if (Constant.pointListUp.size() > 0) {
-
-				// Iterator<Entry<String, Long>> pu =
-				// Constant.pointListUp.entrySet().iterator();
-				Map<String, Long> pu = new HashMap<String, Long>();
-				synchronized (Constant.ps_PointStatusUp) {
-					if (Constant.ps_PointStatusUp.size() > 0) {
-						pu.putAll(Constant.ps_PointStatusUp);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			this.projectStatus.put(project, status);
+		}
+	}
+
+	private void virtual_meter_status_up_Check() {
+		// TODO Auto-generated method stub
+		for (Entry<String, List<String>> ms : Constant.ps_MeterList.entrySet()) {
+			String pm = ms.getKey();
+			List<String> l = ms.getValue();
+			String status = this.gatherStatus(l);
+			String[] ml = pm.split("-");
+			String project = ml[0];
+			String meter = ml[1];
+			try {
+				HBaseUtil.Insert_ps_virtual_meter_status(Constant.agent, Constant.BuildingDB.get(project), project,
+						meter, "up", status, Constant.format(new Date()), "");
+				if (
+				// 首次
+				!this.meterStatus.containsKey(meter) || null == this.meterStatus.get(meter) ||
+				// 已有记录
+						!status.equalsIgnoreCase(this.meterStatus.get(meter))) {
+
+					HBaseUtil.Insert_ps_virtual_meter_log(Constant.agent, Constant.BuildingDB.get(project), project,
+							meter, "up", Constant.format(new Date()), status);
+				}
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			this.meterStatus.put(meter, status);
+		}
+	}
+
+	private void physical_status_address_Check() {
+		// TODO Auto-generated method stub
+		for (Entry<String, List<String>> ps : Constant.ps_PhysicalList.entrySet()) {
+			String physical = ps.getKey();
+			String[] pl = physical.split(";");
+			String project = pl[0];
+			String address_1 = pl.length < 2 ? "" : pl[1];
+			String address_2 = pl.length < 3 ? "" : pl[2];
+			String address_3 = pl.length < 4 ? "" : pl[3];
+			String address_4 = pl.length < 5 ? "" : pl[4];
+			String status = "disconn";
+			if (!physical.endsWith(";;;")) {
+				List<String> l = ps.getValue();
+				status = this.gatherStatus(l);
+			} else {
+				if (Constant.ps_PhysicalStatus.get(physical) != null) {
+					if (this.realCheck - Constant.ps_PhysicalStatus.get(physical) < Constant.Millisecond.MINUTE_1) {
+						status = "conn";
 					}
+				} else {
+					status = "disconn";
+				}
+			}
+			try {
+				HBaseUtil.Insert_ps_physical_status(Constant.agent, Constant.BuildingDB.get(project), project,
+						address_1, address_2, address_3, address_4, status, Constant.format(new Date()), "");
+				if (
+				// 首次
+				!this.physicalStatus.containsKey(physical) || null == this.physicalStatus.get(physical) ||
+				// 已有记录
+						!status.equalsIgnoreCase(this.physicalStatus.get(physical))) {
+
+					HBaseUtil.Insert_ps_physical_log(Constant.agent, Constant.BuildingDB.get(project), project,
+							address_1, address_2, address_3, address_4, Constant.format(new Date()), status);
 				}
-				for (Entry<String, Long> pointUp : pu.entrySet()) {
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			this.physicalStatus.put(physical, status);
+		}
+	}
+
+	private String gatherStatus(List<String> l) {
+		// TODO Auto-generated method stub
+		for (String s : l) {
+			if ("conn".equalsIgnoreCase(this.pointStatus.get(s))) {
+				return "conn";
+			}
+		}
+		return "disconn";
+	}
+
+	private void virtual_point_status_down_Check() {
+		// TODO Auto-generated method stub
+		// down
+		try {
+			List<String> nk = new ArrayList<String>();
+			List<String> ps = new ArrayList<String>();
+			synchronized (Constant.ps_ID2Time) {
+				Iterator<Entry<String, Long>> it = Constant.ps_ID2Time.entrySet().iterator();
+				while (it.hasNext()) {
 					try {
-						String building = pointUp.getKey().split("-")[0];
-						String mac = pointUp.getKey().split("-")[1];
-						String point = pointUp.getKey().split("-")[2];
-						// System.out.println(point);
+						Entry<String, Long> n = it.next();
+
+						String building = Constant.ps_ID2Point.get(n.getKey()).split("-")[0];
+						String mac = Constant.ps_ID2Point.get(n.getKey()).split("-")[1];
+						String point = Constant.ps_ID2Point.get(n.getKey()).split("-")[2];
+
 						String[] slist = point.split("\\.");
 						StringBuffer sb = new StringBuffer();
 						for (int i = 0; i < slist.length - 1; i++) {
@@ -121,166 +284,490 @@ public class CheckThread extends Thread {
 						}
 						String meter = sb.toString();
 						String funcid = slist[slist.length - 1];
-						// System.out.println(meter);
-						// System.out.println(funcid);
-						String status = "";
-
-						if (pointUp.getValue() != null
-								&& this.realCheck - pointUp.getValue() < Constant.Millisecond.MINUTE_10) {
-							HBaseUtil.InsertPointStatus(Constant.agent, Constant.BuildingDB.get(building), building,
-									this.ip, mac, meter, funcid, "up", "conn",
-									Constant.format(new Date(pointUp.getValue())), "");
-							status = "conn";
-						} else {
-							if (pointUp.getValue() != null) {
-								HBaseUtil.InsertPointStatus(Constant.agent, Constant.BuildingDB.get(building), building,
-										this.ip, mac, meter, funcid, "up", "disconn",
-										Constant.format(new Date(pointUp.getValue())), "");
-								status = "disconn";
-							}
-						}
-						if ((!this.pointStatus.containsKey(pointUp.getKey()))
-								|| (!status.equalsIgnoreCase(this.pointStatus.get(pointUp.getKey())))) {
-							// 添加Log记录
-							if (Constant.RecordAddress.containsKey(building + "-" + point.replace(".", "-"))) {
-								int address = Constant.RecordAddress.get(building + "-" + point.replace(".", "-"));
-								if (pointUp.getValue() != null) {
-									HBaseUtil.InsertPointCommunicationLog(Constant.agent,
-											Constant.BuildingDB.get(building), building, this.ip, mac, meter, funcid,
-											"up", (status), Constant.RecordList.get(address).seq,
-											Constant.format(new Date(pointUp.getValue())), "");
-								}
-							}
 
-						}
-						// 更新Status
-						this.pointStatus.put(pointUp.getKey(), status);
-					} catch (Exception e) {
-						// TODO Auto-generated catch block
-						e.printStackTrace();
-						LogUtil.error("CheckThread.realCheck.PointStatus.up.Exception:" + e.toString());
-					}
-				}
-				// down
-				try {
-					List<String> nk = new ArrayList<String>();
-					List<String> ps = new ArrayList<String>();
-					synchronized (Constant.ps_ID2Time) {
-						Iterator<Entry<String, Long>> it = Constant.ps_ID2Time.entrySet().iterator();
-						while (it.hasNext()) {
+						// String function =
+						// Constant.id2function.get(n.getKey());
+
+						if (n.getValue() != null
+								&& System.currentTimeMillis() - n.getValue() > Constant.Millisecond.SECOND_10) {
 							try {
-								Entry<String, Long> n = it.next();
-
-								String building = Constant.ps_ID2Point.get(n.getKey()).split("-")[0];
-								String mac = Constant.ps_ID2Point.get(n.getKey()).split("-")[1];
-								String point = Constant.ps_ID2Point.get(n.getKey()).split("-")[2];
-
-								String[] slist = point.split("\\.");
-								StringBuffer sb = new StringBuffer();
-								for (int i = 0; i < slist.length - 1; i++) {
-									if (i > 0) {
-										sb.append(".");
-									}
-									sb.append(slist[i]);
-								}
-								String meter = sb.toString();
-								String funcid = slist[slist.length - 1];
-
-								// String function =
-								// Constant.id2function.get(n.getKey());
-
-								if (n.getValue() != null
-										&& System.currentTimeMillis() - n.getValue() > Constant.Millisecond.SECOND_10) {
-									try {
-										if ("finish:finish:success"
-												.equalsIgnoreCase(Constant.ps_ID2Status.get(n.getKey()))) {
-											// HBaseUtil.InsertPointStatus(Constant.agent,
-											// Constant.BuildingDB.get(building),
-											// building, this.ip, mac, meter,
-											// funcid, "down", "conn",
-											// // function + ":" +
-											// Constant.format(new
-											// Date(Constant.id2time.get(n.getKey()))),
-											// Constant.id2status.get(n.getKey()));
-											ps.add(Constant.BuildingDB.get(building) + ";" + building + ";" + this.ip
-													+ ";" + mac + ";" + meter + ";" + funcid + ";" + "down" + ";"
-													+ "conn" + ";"
-													+ Constant.format(new Date(Constant.ps_ID2Time.get(n.getKey()))) + ";"
-													+ Constant.ps_ID2Status.get(n.getKey()));
-										} else {
-											// Constant.pointListDown.put(n.getKey(),
-											// null);
-											// HBaseUtil.InsertPointStatus(Constant.agent,
-											// Constant.BuildingDB.get(building),
-											// building, this.ip, mac, meter,
-											// funcid, "down", "disconn",
-											// // function + ":" +
-											// Constant.format(new
-											// Date(Constant.id2time.get(n.getKey()))),
-											// Constant.id2status.get(n.getKey()));
-											ps.add(Constant.BuildingDB.get(building) + ";" + building + ";" + this.ip
-													+ ";" + mac + ";" + meter + ";" + funcid + ";" + "down" + ";"
-													+ "disconn" + ";"
-													+ Constant.format(new Date(Constant.ps_ID2Time.get(n.getKey()))) + ";"
-													+ Constant.ps_ID2Status.get(n.getKey()));
-										}
-
-									} catch (Exception e) {
-										// TODO Auto-generated catch block
-										e.printStackTrace();
-									}
-									nk.add(n.getKey());
-									it.remove();
+								if ("finish:finish:success".equalsIgnoreCase(Constant.ps_ID2Status.get(n.getKey()))) {
+									// HBaseUtil.InsertPointStatus(Constant.agent,
+									// Constant.BuildingDB.get(building),
+									// building, this.ip, mac, meter,
+									// funcid, "down", "conn",
+									// // function + ":" +
+									// Constant.format(new
+									// Date(Constant.id2time.get(n.getKey()))),
+									// Constant.id2status.get(n.getKey()));
+									ps.add(Constant.BuildingDB.get(building) + ";" + building + ";" + this.ip + ";"
+											+ mac + ";" + meter + ";" + funcid + ";" + "down" + ";" + "conn" + ";"
+											+ Constant.format(new Date(Constant.ps_ID2Time.get(n.getKey()))) + ";"
+											+ Constant.ps_ID2Status.get(n.getKey()));
+								} else {
+									// Constant.pointListDown.put(n.getKey(),
+									// null);
+									// HBaseUtil.InsertPointStatus(Constant.agent,
+									// Constant.BuildingDB.get(building),
+									// building, this.ip, mac, meter,
+									// funcid, "down", "disconn",
+									// // function + ":" +
+									// Constant.format(new
+									// Date(Constant.id2time.get(n.getKey()))),
+									// Constant.id2status.get(n.getKey()));
+									ps.add(Constant.BuildingDB.get(building) + ";" + building + ";" + this.ip + ";"
+											+ mac + ";" + meter + ";" + funcid + ";" + "down" + ";" + "disconn" + ";"
+											+ Constant.format(new Date(Constant.ps_ID2Time.get(n.getKey()))) + ";"
+											+ Constant.ps_ID2Status.get(n.getKey()));
 								}
+
 							} catch (Exception e) {
 								// TODO Auto-generated catch block
 								e.printStackTrace();
 							}
+							nk.add(n.getKey());
+							it.remove();
 						}
+					} catch (Exception e) {
+						// TODO Auto-generated catch block
+						e.printStackTrace();
 					}
+				}
+			}
 
-					synchronized (Constant.ps_ID2Point) {
-						for (String k : nk) {
-							if (Constant.ps_ID2Point.containsKey(k)) {
-								Constant.ps_ID2Point.remove(k);
-							}
-						}
+			synchronized (Constant.ps_ID2Point) {
+				for (String k : nk) {
+					if (Constant.ps_ID2Point.containsKey(k)) {
+						Constant.ps_ID2Point.remove(k);
 					}
-					synchronized (Constant.ps_ID2Status) {
-						for (String k : nk) {
-							if (Constant.ps_ID2Status.containsKey(k)) {
-								Constant.ps_ID2Status.remove(k);
-							}
-						}
+				}
+			}
+			synchronized (Constant.ps_ID2Status) {
+				for (String k : nk) {
+					if (Constant.ps_ID2Status.containsKey(k)) {
+						Constant.ps_ID2Status.remove(k);
 					}
-					synchronized (Constant.ps_ID2Function) {
-						for (String k : nk) {
-							if (Constant.ps_ID2Function.containsKey(k)) {
-								Constant.ps_ID2Function.remove(k);
-							}
-						}
+				}
+			}
+			synchronized (Constant.ps_ID2Function) {
+				for (String k : nk) {
+					if (Constant.ps_ID2Function.containsKey(k)) {
+						Constant.ps_ID2Function.remove(k);
+					}
+				}
+			}
+
+			nk = new ArrayList<String>();
+			for (String s : ps) {
+				String[] a = s.split(";");
+				HBaseUtil.Insert_ps_virtual_point_status(Constant.agent, a[0], a[1], a[4], Long.valueOf(a[5]), a[6],
+						a[7], a[8], a[9]);
+				Thread.sleep(1L);
+			}
+			ps = new ArrayList<String>();
+
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+			LogUtil.error("CheckThread.realCheck2.PointStatus.down.Exception:" + e.toString());
+		}
+	}
+
+	private void virtual_point_status_up_Check() {
+		// TODO Auto-generated method stub
+		// PointStatus
+		// up
+		// LogUtil.info("Constant.pointListUp:" + Constant.pointListUp);
+		// synchronized (Constant.pointListUp) {
+		// if (Constant.pointListUp.size() > 0) {
+
+		// Iterator<Entry<String, Long>> pu =
+		// Constant.pointListUp.entrySet().iterator();
+		Map<String, Long> pu = new HashMap<String, Long>();
+		synchronized (Constant.ps_PointStatusUp) {
+			if (Constant.ps_PointStatusUp.size() > 0) {
+				pu.putAll(Constant.ps_PointStatusUp);
+			}
+		}
+		for (Entry<String, Long> pointUp : pu.entrySet()) {
+			try {
+				String building = pointUp.getKey().split("-")[0];
+				String point = pointUp.getKey().split("-")[1];
+
+				String[] slist = point.split("\\.");
+				StringBuffer sb = new StringBuffer();
+				for (int i = 0; i < slist.length - 1; i++) {
+					if (i > 0) {
+						sb.append(".");
 					}
+					sb.append(slist[i]);
+				}
+				String meter = sb.toString();
+				String funcid = slist[slist.length - 1];
+				String status = "";
 
-					nk = new ArrayList<String>();
-					for (String s : ps) {
-						String[] a = s.split(";");
-						HBaseUtil.InsertPointStatus(Constant.agent, a[0], a[1], a[2], a[3], a[4], a[5], a[6], a[7],
-								a[8], a[9]);
-						Thread.sleep(1L);
+				if (pointUp.getValue() != null) {
+					if (this.realCheck - pointUp.getValue() < Constant.Millisecond.MINUTE_10) {
+						status = "conn";
+					} else {
+						status = "disconn";
 					}
-					ps = new ArrayList<String>();
+				} else {
+					// 首次
+					status = "disconn";
 
-				} catch (Exception e) {
-					// TODO Auto-generated catch block
-					e.printStackTrace();
-					LogUtil.error("CheckThread.realCheck.PointStatus.down.Exception:" + e.toString());
 				}
+				this.Insert_ps_virtual_point_status(building, meter, Long.valueOf(funcid), "up", status,
+						pointUp.getValue(), "");
+				if (
+				// 首次
+				!this.pointStatus.containsKey(pointUp.getKey()) || null == this.pointStatus.get(pointUp.getKey()))
+
+				{
+					this.Insert_ps_virtual_point_log(building, meter, Long.valueOf(funcid), "up",
+							System.currentTimeMillis(), 0l, status, "");
+				} else if (
+				// 已有记录
+				(!status.equalsIgnoreCase(this.pointStatus.get(pointUp.getKey())))) {
+					// 添加Log记录
+					if (Constant.RecordAddress.containsKey(building + "-" + point.replace(".", "-"))) {
+						if (Constant.RecordAddress.get(building + "-" + point.replace(".", "-")) != null) {
+							int address = Constant.RecordAddress.get(building + "-" + point.replace(".", "-"));
+							this.Insert_ps_virtual_point_log(building, meter, Long.valueOf(funcid), "up",
+									pointUp.getValue(), Long.valueOf(Constant.RecordList.get(address).seq), status, "");
+						}
+					}
+				}
+				// 更新Status
+				this.pointStatus.put(pointUp.getKey(), status);
 			} catch (Exception e) {
 				// TODO Auto-generated catch block
 				e.printStackTrace();
-				LogUtil.error("CheckThread.realCheck.Exception: " + e.toString());
+				LogUtil.error("CheckThread.virtual_point_status_up_Check.Exception:" + e.toString());
+			}
+		}
+	}
+
+	private void Insert_ps_virtual_point_log(String building, String meter, Long funcid, String updown, Long time,
+			Long seq, String status, String feedback) {
+		// TODO Auto-generated method stub
+		try {
+			if (time == null) {
+				HBaseUtil.Insert_ps_virtual_point_log(Constant.agent, Constant.BuildingDB.get(building), building,
+						meter, Long.valueOf(funcid), updown, Constant.format(new Date()), seq, status, feedback);
+			} else {
+				HBaseUtil.Insert_ps_virtual_point_log(Constant.agent, Constant.BuildingDB.get(building), building,
+						meter, Long.valueOf(funcid), updown, Constant.format(new Date(time)), seq, status, feedback);
+			}
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	// private void physical_status_address_1_Check() {
+	// // TODO Auto-generated method stub
+	// // physical_status
+	// // address_1
+	// for (Entry<String, Long> node : Constant.ps_NodeStatus.entrySet()) {
+	// try {
+	// String building = node.getKey().split("-")[0];
+	// String mac = node.getKey().split("-")[1];
+	// String status = "";
+	// if (node.getValue() != null && this.realCheck - node.getValue() <
+	// Constant.Millisecond.MINUTE_1) {
+	// status = "conn";
+	// HBaseUtil.Insert_ps_physical_status(Constant.agent,
+	// Constant.BuildingDB.get(building), building,
+	// mac, "", "", "", status, Constant.format(new Date(node.getValue())), "");
+	// } else {
+	// status = "disconn";
+	// HBaseUtil.Insert_ps_physical_status(Constant.agent,
+	// Constant.BuildingDB.get(building), building,
+	// mac, "", "", "", status, Constant.format(new Date(node.getValue())), "");
+	// }
+	//
+	// if ((!this.nodeStatus.containsKey(node.getKey()))
+	// || (!status.equalsIgnoreCase(this.nodeStatus.get(node.getKey())))) {
+	// // 添加Log记录
+	// HBaseUtil.Insert_ps_physical_log(Constant.agent,
+	// Constant.BuildingDB.get(building), building, mac,
+	// "", "", "", Constant.format(new Date(node.getValue())), (status));
+	// }
+	// // 更新Status
+	// this.nodeStatus.put(node.getKey(), status);
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// LogUtil.error("CheckThread.realCheck.CollectorStatus.Exception: " +
+	// e.toString());
+	// }
+	// }
+	// }
+
+	private void Insert_ps_virtual_point_status(String building, String meter, Long funcid, String updown,
+			String status, Long time, String detail) {
+		// TODO Auto-generated method stub
+		try {
+			if (time == null) {
+				HBaseUtil.Insert_ps_virtual_point_status(Constant.agent, Constant.BuildingDB.get(building), building,
+						meter, Long.valueOf(funcid), "up", status, Constant.format(new Date()), "");
+
+			} else {
+				HBaseUtil.Insert_ps_virtual_point_status(Constant.agent, Constant.BuildingDB.get(building), building,
+						meter, Long.valueOf(funcid), "up", status, Constant.format(new Date(time)), "");
 			}
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
 		}
 	}
 
+	// private void realCheck() {
+	// // TODO Auto-generated method stub
+	//
+	// if (System.currentTimeMillis() - this.realCheck >
+	// Constant.Millisecond.SECOND_3) {
+	// this.realCheck = System.currentTimeMillis();
+	//
+	// try {
+	// // CollectorStatus
+	// // LogUtil.info("Constant.nodeList:" + Constant.nodeList);
+	// for (Entry<String, Long> node : Constant.ps_NodeStatus.entrySet()) {
+	// try {
+	// String building = node.getKey().split("-")[0];
+	// String mac = node.getKey().split("-")[1];
+	// String status = "";
+	// if (node.getValue() != null
+	// && this.realCheck - node.getValue() < Constant.Millisecond.MINUTE_1) {
+	// HBaseUtil.InsertCollectorStatus(Constant.agent,
+	// Constant.BuildingDB.get(building), building,
+	// this.ip, mac, "conn", Constant.format(new Date(node.getValue())));
+	// status = "conn";
+	// } else {
+	// HBaseUtil.InsertCollectorStatus(Constant.agent,
+	// Constant.BuildingDB.get(building), building,
+	// this.ip, mac, "disconn", Constant.format(new Date(node.getValue())));
+	// status = "disconn";
+	// }
+	//
+	// if ((!this.nodeStatus.containsKey(node.getKey()))
+	// || (!status.equalsIgnoreCase(this.nodeStatus.get(node.getKey())))) {
+	// // 添加Log记录
+	// HBaseUtil.InsertCollectorCommunicationLog(Constant.agent,
+	// Constant.BuildingDB.get(building),
+	// building, this.ip, mac, Constant.format(new Date(node.getValue())),
+	// (status));
+	// }
+	// // 更新Status
+	// this.nodeStatus.put(node.getKey(), status);
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// LogUtil.error("CheckThread.realCheck.CollectorStatus.Exception: " +
+	// e.toString());
+	// }
+	// }
+	// // PointStatus
+	// // up
+	// // LogUtil.info("Constant.pointListUp:" + Constant.pointListUp);
+	// // synchronized (Constant.pointListUp) {
+	// // if (Constant.pointListUp.size() > 0) {
+	//
+	// // Iterator<Entry<String, Long>> pu =
+	// // Constant.pointListUp.entrySet().iterator();
+	// Map<String, Long> pu = new HashMap<String, Long>();
+	// synchronized (Constant.ps_PointStatusUp) {
+	// if (Constant.ps_PointStatusUp.size() > 0) {
+	// pu.putAll(Constant.ps_PointStatusUp);
+	// }
+	// }
+	// for (Entry<String, Long> pointUp : pu.entrySet()) {
+	// try {
+	// String building = pointUp.getKey().split("-")[0];
+	// String mac = pointUp.getKey().split("-")[1];
+	// String point = pointUp.getKey().split("-")[2];
+	// // System.out.println(point);
+	// String[] slist = point.split("\\.");
+	// StringBuffer sb = new StringBuffer();
+	// for (int i = 0; i < slist.length - 1; i++) {
+	// if (i > 0) {
+	// sb.append(".");
+	// }
+	// sb.append(slist[i]);
+	// }
+	// String meter = sb.toString();
+	// String funcid = slist[slist.length - 1];
+	// // System.out.println(meter);
+	// // System.out.println(funcid);
+	// String status = "";
+	//
+	// if (pointUp.getValue() != null
+	// && this.realCheck - pointUp.getValue() < Constant.Millisecond.MINUTE_10)
+	// {
+	// HBaseUtil.InsertPointStatus(Constant.agent,
+	// Constant.BuildingDB.get(building), building,
+	// this.ip, mac, meter, funcid, "up", "conn",
+	// Constant.format(new Date(pointUp.getValue())), "");
+	// status = "conn";
+	// } else {
+	// if (pointUp.getValue() != null) {
+	// HBaseUtil.InsertPointStatus(Constant.agent,
+	// Constant.BuildingDB.get(building), building,
+	// this.ip, mac, meter, funcid, "up", "disconn",
+	// Constant.format(new Date(pointUp.getValue())), "");
+	// status = "disconn";
+	// }
+	// }
+	// if ((!this.pointStatus.containsKey(pointUp.getKey()))
+	// || (!status.equalsIgnoreCase(this.pointStatus.get(pointUp.getKey())))) {
+	// // 添加Log记录
+	// if (Constant.RecordAddress.containsKey(building + "-" +
+	// point.replace(".", "-"))) {
+	// int address = Constant.RecordAddress.get(building + "-" +
+	// point.replace(".", "-"));
+	// if (pointUp.getValue() != null) {
+	// HBaseUtil.InsertPointCommunicationLog(Constant.agent,
+	// Constant.BuildingDB.get(building), building, this.ip, mac, meter, funcid,
+	// "up", (status), Constant.RecordList.get(address).seq,
+	// Constant.format(new Date(pointUp.getValue())), "");
+	// }
+	// }
+	//
+	// }
+	// // 更新Status
+	// this.pointStatus.put(pointUp.getKey(), status);
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// LogUtil.error("CheckThread.realCheck.PointStatus.up.Exception:" +
+	// e.toString());
+	// }
+	// }
+	// // down
+	// try {
+	// List<String> nk = new ArrayList<String>();
+	// List<String> ps = new ArrayList<String>();
+	// synchronized (Constant.ps_ID2Time) {
+	// Iterator<Entry<String, Long>> it =
+	// Constant.ps_ID2Time.entrySet().iterator();
+	// while (it.hasNext()) {
+	// try {
+	// Entry<String, Long> n = it.next();
+	//
+	// String building = Constant.ps_ID2Point.get(n.getKey()).split("-")[0];
+	// String mac = Constant.ps_ID2Point.get(n.getKey()).split("-")[1];
+	// String point = Constant.ps_ID2Point.get(n.getKey()).split("-")[2];
+	//
+	// String[] slist = point.split("\\.");
+	// StringBuffer sb = new StringBuffer();
+	// for (int i = 0; i < slist.length - 1; i++) {
+	// if (i > 0) {
+	// sb.append(".");
+	// }
+	// sb.append(slist[i]);
+	// }
+	// String meter = sb.toString();
+	// String funcid = slist[slist.length - 1];
+	//
+	// // String function =
+	// // Constant.id2function.get(n.getKey());
+	//
+	// if (n.getValue() != null
+	// && System.currentTimeMillis() - n.getValue() >
+	// Constant.Millisecond.SECOND_10) {
+	// try {
+	// if ("finish:finish:success"
+	// .equalsIgnoreCase(Constant.ps_ID2Status.get(n.getKey()))) {
+	// // HBaseUtil.InsertPointStatus(Constant.agent,
+	// // Constant.BuildingDB.get(building),
+	// // building, this.ip, mac, meter,
+	// // funcid, "down", "conn",
+	// // // function + ":" +
+	// // Constant.format(new
+	// // Date(Constant.id2time.get(n.getKey()))),
+	// // Constant.id2status.get(n.getKey()));
+	// ps.add(Constant.BuildingDB.get(building) + ";" + building + ";" + this.ip
+	// + ";" + mac + ";" + meter + ";" + funcid + ";" + "down" + ";"
+	// + "conn" + ";"
+	// + Constant.format(new Date(Constant.ps_ID2Time.get(n.getKey())))
+	// + ";" + Constant.ps_ID2Status.get(n.getKey()));
+	// } else {
+	// // Constant.pointListDown.put(n.getKey(),
+	// // null);
+	// // HBaseUtil.InsertPointStatus(Constant.agent,
+	// // Constant.BuildingDB.get(building),
+	// // building, this.ip, mac, meter,
+	// // funcid, "down", "disconn",
+	// // // function + ":" +
+	// // Constant.format(new
+	// // Date(Constant.id2time.get(n.getKey()))),
+	// // Constant.id2status.get(n.getKey()));
+	// ps.add(Constant.BuildingDB.get(building) + ";" + building + ";" + this.ip
+	// + ";" + mac + ";" + meter + ";" + funcid + ";" + "down" + ";"
+	// + "disconn" + ";"
+	// + Constant.format(new Date(Constant.ps_ID2Time.get(n.getKey())))
+	// + ";" + Constant.ps_ID2Status.get(n.getKey()));
+	// }
+	//
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// }
+	// nk.add(n.getKey());
+	// it.remove();
+	// }
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// }
+	// }
+	// }
+	//
+	// synchronized (Constant.ps_ID2Point) {
+	// for (String k : nk) {
+	// if (Constant.ps_ID2Point.containsKey(k)) {
+	// Constant.ps_ID2Point.remove(k);
+	// }
+	// }
+	// }
+	// synchronized (Constant.ps_ID2Status) {
+	// for (String k : nk) {
+	// if (Constant.ps_ID2Status.containsKey(k)) {
+	// Constant.ps_ID2Status.remove(k);
+	// }
+	// }
+	// }
+	// synchronized (Constant.ps_ID2Function) {
+	// for (String k : nk) {
+	// if (Constant.ps_ID2Function.containsKey(k)) {
+	// Constant.ps_ID2Function.remove(k);
+	// }
+	// }
+	// }
+	//
+	// nk = new ArrayList<String>();
+	// for (String s : ps) {
+	// String[] a = s.split(";");
+	// HBaseUtil.InsertPointStatus(Constant.agent, a[0], a[1], a[2], a[3], a[4],
+	// a[5], a[6], a[7],
+	// a[8], a[9]);
+	// Thread.sleep(1L);
+	// }
+	// ps = new ArrayList<String>();
+	//
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// LogUtil.error("CheckThread.realCheck.PointStatus.down.Exception:" +
+	// e.toString());
+	// }
+	// } catch (Exception e) {
+	// // TODO Auto-generated catch block
+	// e.printStackTrace();
+	// LogUtil.error("CheckThread.realCheck.Exception: " + e.toString());
+	// }
+	// }
+	// }
+
 }

+ 114 - 82
collect/src/main/java/com/saga/thread/common/LoadConfigThread.java

@@ -29,99 +29,131 @@ public class LoadConfigThread extends Thread {
 
 	public void run() {
 		LogUtil.info("LoadConfigThread" + this.getName() + " Starting ...");
-
+		try {
+			Thread.sleep(Constant.Millisecond.MINUTE_30);
+		} catch (InterruptedException e1) {
+			// TODO Auto-generated catch block
+			e1.printStackTrace();
+		}
 		while (true) {
 			try {
 				// 加载config配置文件
+				this.loadConfig();
+				// 加载DB点位列表
+				this.loadPointList2();
+				// 加载采集器配置文件
+				// this.loadCJQ();
+				Thread.sleep(Constant.Millisecond.MINUTE_5);
 
-				String resource = "/config.json";
-				String path = null;
-				JsonParser config = new JsonParser();
-
-				try {
-					LogUtil.info("System.get " + System.getProperty("user.dir") + resource);
-					LogUtil.info("LogUtil.get " + LogUtil.GetPath() + resource);
-					path = LogUtil.GetPath() + resource;
-
-					JsonObject object = (JsonObject) config.parse(new FileReader(path));
-
-					JsonArray db = object.get("DB").getAsJsonArray();
-					for (int i = 0; i < db.size(); i++) {
-						JsonObject subObject = db.get(i).getAsJsonObject();
-						if (subObject.get("building") != null) {
-							String building = subObject.get("building").getAsString();
-							// if(Constant.BuildingDB.containsKey(building)){
-							// continue;
-							// }
-							Constant.BuildingDB.put(building, subObject.get("database").getAsString());
-
-							if (subObject.get("buildingadds") != null) {
-								JsonArray buildingAdds = subObject.get("buildingadds").getAsJsonArray();
-								for (int j = 0; j < buildingAdds.size(); j++) {
-									JsonObject addObject = buildingAdds.get(j).getAsJsonObject();
-									String builing_new = addObject.get("building").getAsString();
-									if (!Constant.BuildingAdds.containsKey(building)) {
-										Constant.BuildingAdds.put(building, new ArrayList<String>());
-									}
-									if (!exist(builing_new)) {
-										Constant.BuildingAdds.get(building).add(builing_new);
-										Constant.BuildingConv.put(builing_new, building);
-									}
-								}
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				LogUtil.error("LoadConfigThread.run.Exception: " + e.toString());
+			}
+		}
+	}
 
+	private void loadPointList2() {
+		// TODO Auto-generated method stub
+		for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+			try {
+				String project = bd.getKey();
+				String db = bd.getValue();
+				HBaseUtil.LoadDB2Cache_dy_pointlist(Constant.agent, db, "dy_pointlist", project);
+
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+				LogUtil.error("LoadConfigThread.loadPointList2: " + e.toString());
+			}
+		}
+	}
+
+	private void loadCJQ() {
+		// TODO Auto-generated method stub
+		// 文件夹命名格式:config+"-"+楼号+"-"+mac序号,示例:config-1101080002-1
+		Constant.getConfigFile(LogUtil.GetPath(), "config-", "port");
+		// 采集器点位位置信息写入数据库
+		for (Entry<String, String> pi : Constant.pointInfo.entrySet()) {
+			String building = pi.getKey().split("-")[0];
+			String mac = pi.getKey().split("-")[1];
+			String node = mac;
+			String point = pi.getKey().split("-")[2];
+			String busno = pi.getValue().split("-")[0];
+			String site = pi.getValue().split("-")[1];
+			try {
+				LogUtil.info(building + " " + node + " " + point + " " + busno + " " + site);
+				HBaseUtil.InsertPointInfo(Constant.agent, Constant.BuildingDB.get(building), building, node, point,
+						busno, site);
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+			// System.out.println(building + node + point + busno +
+			// site);
+		}
+		try {
+			Thread.sleep(Constant.Millisecond.MINUTE_1);
+		} catch (InterruptedException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	private void loadConfig() {
+		// TODO Auto-generated method stub
+		String resource = "/config.json";
+		String path = null;
+		JsonParser config = new JsonParser();
+
+		try {
+			LogUtil.info("System.get " + System.getProperty("user.dir") + resource);
+			LogUtil.info("LogUtil.get " + LogUtil.GetPath() + resource);
+			path = LogUtil.GetPath() + resource;
+
+			JsonObject object = (JsonObject) config.parse(new FileReader(path));
+
+			JsonArray db = object.get("DB").getAsJsonArray();
+			for (int i = 0; i < db.size(); i++) {
+				JsonObject subObject = db.get(i).getAsJsonObject();
+				if (subObject.get("building") != null) {
+					String building = subObject.get("building").getAsString();
+					// if(Constant.BuildingDB.containsKey(building)){
+					// continue;
+					// }
+					Constant.BuildingDB.put(building, subObject.get("database").getAsString());
+
+					if (subObject.get("buildingadds") != null) {
+						JsonArray buildingAdds = subObject.get("buildingadds").getAsJsonArray();
+						for (int j = 0; j < buildingAdds.size(); j++) {
+							JsonObject addObject = buildingAdds.get(j).getAsJsonObject();
+							String builing_new = addObject.get("building").getAsString();
+							if (!Constant.BuildingAdds.containsKey(building)) {
+								Constant.BuildingAdds.put(building, new ArrayList<String>());
+							}
+							if (!exist(builing_new)) {
+								Constant.BuildingAdds.get(building).add(builing_new);
+								Constant.BuildingConv.put(builing_new, building);
 							}
-						} else if (subObject.get("insertthread") != null) {
-							Constant.InsertThread = subObject.get("insertthread").getAsInt();
 						}
-					}
-					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());
-						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) {
-					// TODO Auto-generated catch block
-					e.printStackTrace();
-				}
-
-				// 加载采集器配置文件
-				// 文件夹命名格式:config+"-"+楼号+"-"+mac序号,示例:config-1101080002-1
-				Constant.getConfigFile(LogUtil.GetPath(), "config-", "port");
-				// 采集器点位位置信息写入数据库
-				for (Entry<String, String> pi : Constant.pointInfo.entrySet()) {
-					String building = pi.getKey().split("-")[0];
-					String mac = pi.getKey().split("-")[1];
-					String node = mac;
-					String point = pi.getKey().split("-")[2];
-					String busno = pi.getValue().split("-")[0];
-					String site = pi.getValue().split("-")[1];
-					try {
-						LogUtil.info(building + " " + node + " " + point + " " + busno + " " + site);
-						HBaseUtil.InsertPointInfo(Constant.agent, Constant.BuildingDB.get(building), building, node,
-								point, busno, site);
-					} catch (Exception e) {
-						// TODO Auto-generated catch block
-						e.printStackTrace();
 					}
-					// System.out.println(building + node + point + busno +
-					// site);
+				} else if (subObject.get("insertthread") != null) {
+					Constant.InsertThread = subObject.get("insertthread").getAsInt();
 				}
-				try {
-					Thread.sleep(Constant.Millisecond.MINUTE_1);
-				} catch (InterruptedException e) {
-					// TODO Auto-generated catch block
-					e.printStackTrace();
-				}
-
-			} catch (Exception e) {
-				// TODO Auto-generated catch block
-				LogUtil.error("LoadConfigThread.run.Exception: " + e.toString());
 			}
+			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());
+				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) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
 		}
 	}
 

+ 2 - 2
collect/src/main/java/com/saga/thread/common/SaveThread.java

@@ -145,7 +145,7 @@ public class SaveThread extends Thread {
 												// synchronized
 												// (Constant.pointListUp) {
 												Constant.ps_PointStatusUp.put(
-														building + "-" + mac + "-" + meter + "." + record.funcID,
+														building + "-" + meter + "." + record.funcID,
 														System.currentTimeMillis());
 												// }
 
@@ -157,7 +157,7 @@ public class SaveThread extends Thread {
 												// synchronized
 												// (Constant.pointListUp) {
 												Constant.ps_PointStatusUp.put(
-														building + "-" + mac + "-" + meter + "." + record.funcID,
+														building + "-" + meter + "." + record.funcID,
 														System.currentTimeMillis());
 
 												// }

File diff suppressed because it is too large
+ 449 - 49
collect/src/main/java/com/saga/thread/down/CollectThread.java


+ 283 - 9
collect/src/main/java/com/saga/thread/up/SendThread.java

@@ -54,13 +54,25 @@ public class SendThread extends Thread {
 				// Constant.setlog = new File(
 				// System.getProperty("user.dir") + "/setlog" +
 				// Constant.format_day(new Date()) + ".log");
-
-				// 返pointset结果包
-				this.returnPointSet();
-				// 处理PopRece包
-				PacketEntity packageRece = this.uploadServer.PopRece();
-				if (packageRece != null) {
-					this.handlePopRece(packageRece);
+				// // V1.0点位日志结构
+				// {
+				// // 返pointset结果包
+				// this.returnPointSet();
+				// // 处理PopRece包
+				// PacketEntity packageRece = this.uploadServer.PopRece();
+				// if (packageRece != null) {
+				// this.handlePopRece(packageRece);
+				// }
+				// }
+				// V2.0点位日志结构
+				{
+					// 返pointset结果包
+					this.returnPointSet2();
+					// 处理PopRece包
+					PacketEntity packageRece = this.uploadServer.PopRece();
+					if (packageRece != null) {
+						this.handlePopRece2(packageRece);
+					}
 				}
 
 				Thread.sleep(1L);
@@ -186,6 +198,268 @@ public class SendThread extends Thread {
 		}
 	}
 
+	private void handlePopRece2(PacketEntity packageRece) {
+		// TODO Auto-generated method stub
+		if ((packageRece != null) && (packageRece.content.packetString != null)
+				&& (packageRece.content.packetString.length() > 0)) {
+			LogUtil.info(this.uploadServer.Name() + " PopRece: " + packageRece.address + " "
+					+ packageRece.content.packetString);
+			Constant.sendRec += packageRece.content.packetString.length();
+			String results = packageRece.content.packetString.replace("(", "").replace(")", "").replace(" ", "");
+			String[] resultArray = results.split("&");
+
+			for (String result : resultArray) {
+				String[] rList = result.split(";");
+				if (rList.length > 6) {
+					String MAC = rList[1];
+
+					if (Constant.pointMAC.containsKey(rList[0] + "-" + rList[5])) {
+						MAC = Constant.pointMAC.get(rList[0] + "-" + rList[5]);
+						LogUtil.info("MAC:" + (rList[0] + "-" + rList[5]) + " " + rList[1] + "-->" + MAC);
+					}
+					String id = rList[0] + "-" + MAC + "-" + rList[4];
+					String key = rList[0] + "-" + MAC + "-" + rList[5] + "." + rList[6];
+					// Constant.id2time.put(id, System.currentTimeMillis());
+					// Constant.id2point.put(id, key);
+					// Constant.id2function.put(id, rList[2]);
+					// Constant.id2status.put(id, "processing:rcvd");
+
+					try {
+						HBaseUtil.Insert_ps_virtual_point_log(Constant.agent, Constant.BuildingDB.get(rList[0]),
+								rList[0], rList[5], Long.valueOf(rList[6]), this.getupdown(rList[2]),
+								Constant.format(new Date()), Long.valueOf(rList[4]), rList[2], "processing:rcvd");
+					} catch (UnknownHostException e1) {
+						// TODO Auto-generated catch block
+						e1.printStackTrace();
+					} catch (Exception e1) {
+						// TODO Auto-generated catch block
+						e1.printStackTrace();
+					}
+
+					if ("pointset".equalsIgnoreCase(rList[2])) {
+						// 1101070037;1;pointset;;123;1001;11;3.1
+						synchronized (Constant.ps_ID2Time) {
+							Constant.ps_ID2Time.put(id, System.currentTimeMillis());
+						}
+						synchronized (Constant.ps_ID2Point) {
+							Constant.ps_ID2Point.put(id, key);
+						}
+						synchronized (Constant.ps_ID2Function) {
+							Constant.ps_ID2Function.put(id, rList[2]);
+						}
+						synchronized (Constant.ps_ID2Status) {
+							Constant.ps_ID2Status.put(id, "processing:rcvd");
+						}
+
+						PointSet ps = new PointSet();
+						ps.uploadName = this.uploadServer.Name();
+						ps.address = packageRece.address;
+						ps.content = rList;
+						ps.sendTime = 0;
+						ps.recTime = 0;
+						ps.buildingSign = rList[0];
+						ps.buildingSignUP = rList[0];
+						// 楼号有转换规则,且转换后的楼-表有数据才进行转换
+						// 解决一楼多楼号时总有部分仪表无法下发的问题
+						if (Constant.BuildingConv.containsKey(ps.buildingSign) && Constant.Meter_Collector_Control
+								.containsKey(Constant.BuildingConv.get(ps.buildingSign) + "-" + ps.meterSign)) {
+							ps.buildingSign = Constant.BuildingConv.get(ps.buildingSign);
+						}
+
+						ps.ID = Integer.valueOf(rList[4]);
+						ps.meterSign = rList[5];
+						ps.mac = MAC;
+						this.pointMac_UP.put(ps.buildingSignUP + "-" + ps.meterSign, rList[1]);
+						ps.funcID = Integer.valueOf(rList[6]);
+						ps.dataSet = Double.valueOf(rList[7]);
+						// String adress = packageRece.address;
+						// String cmd = ps.buildingSign + ";" + ps.ID + ";";
+						// +ps.meterSign + ";" + ps.funcID + ";"
+						// + ps.dataSet;
+
+						// Constant.CollectorPointSetList.put(cmd, adress);
+						synchronized (Constant.pointSetList) {
+							Constant.pointSetList.add(ps);
+						}
+
+					} else if (rList.length > 6 && "pointread".equalsIgnoreCase(rList[2])) {
+						try {
+							StringBuffer sb = new StringBuffer();
+							sb = this.getCacheData(rList);
+							if (sb != null && sb.toString().length() > 0) {
+								this.uploadServer.AppendToSend(packageRece.address, new Packet(sb.toString()));
+								LogUtil.info(this.uploadServer.Name() + " AppendToSend: " + packageRece.address + " "
+										+ sb.toString());
+								// Constant.id2status.put(id, "finish:finish:" +
+								// "success");
+								HBaseUtil.Insert_ps_virtual_point_log(Constant.agent, Constant.BuildingDB.get(rList[0]),
+										rList[0], rList[5], Long.valueOf(rList[6]), this.getupdown(rList[2]),
+										Constant.format(new Date()), Long.valueOf(rList[4]), "pointreadack",
+										"finish:finish:" + "success");
+
+							} else {
+								// Constant.id2status.put(id, "finish:finish:" +
+								// "fail");
+								HBaseUtil.Insert_ps_virtual_point_log(Constant.agent, Constant.BuildingDB.get(rList[0]),
+										rList[0], rList[5], Long.valueOf(rList[6]), this.getupdown(rList[2]),
+										Constant.format(new Date()), Long.valueOf(rList[4]), "pointreadack",
+										"finish:finish:" + "fail");
+							}
+
+						} catch (Exception e) {
+							e.printStackTrace();
+							LogUtil.error("SendThread.handlePopRece2.pointread.Exception: " + e.toString());
+						}
+
+					} else if (rList.length > 6 && "realtimepointdata".equalsIgnoreCase(rList[2])) {
+						try {
+							StringBuffer sb = new StringBuffer();
+							sb = this.getCacheData2(rList);
+							if (sb != null && sb.toString().length() > 0) {
+								this.uploadServer.AppendToSend(packageRece.address, new Packet(sb.toString()));
+								LogUtil.info(this.uploadServer.Name() + " AppendToSend: " + packageRece.address + " "
+										+ sb.toString());
+								// Constant.id2status.put(id, "finish:finish:" +
+								// "success");
+								HBaseUtil.Insert_ps_virtual_point_log(Constant.agent, Constant.BuildingDB.get(rList[0]),
+										rList[0], rList[5], Long.valueOf(rList[6]), this.getupdown(rList[2]),
+										Constant.format(new Date()), Long.valueOf(rList[4]), "realtimepointdataack",
+										"finish:finish:" + "success");
+
+							} else {
+								// Constant.id2status.put(id, "finish:finish:" +
+								// "fail");
+								HBaseUtil.Insert_ps_virtual_point_log(Constant.agent, Constant.BuildingDB.get(rList[0]),
+										rList[0], rList[5], Long.valueOf(rList[6]), this.getupdown(rList[2]),
+										Constant.format(new Date()), Long.valueOf(rList[4]), "realtimepointdataack",
+										"finish:finish:" + "fail");
+							}
+
+						} catch (Exception e) {
+							e.printStackTrace();
+							LogUtil.error("SendThread.handlePopRece2.realtimepointdata.Exception: " + e.toString());
+						}
+					}
+				}
+			}
+		}
+	}
+
+	private void returnPointSet2() {
+		// TODO Auto-generated method stub
+		try {
+			synchronized (Constant.pointSetList) {
+				if (Constant.pointSetList.size() > 0) {
+					Iterator<PointSet> it = Constant.pointSetList.iterator();
+					while (it.hasNext()) {
+						StringBuffer sb = new StringBuffer();
+						PointSet ps = it.next();
+						// String address =
+						// Constant.CollectorPointSetList.get(ps.buildingSign +
+						// ";"
+						// + ps.ID + ";");
+						// String address = ps.address;
+
+						if (!ps.uploadName.equalsIgnoreCase(this.uploadServer.Name())) {
+							break;
+						}
+
+						if (ps.address == null) {
+							break;
+						}
+						String MAC = ps.mac;
+						if (this.pointMac_UP.containsKey(ps.buildingSignUP + "-" + ps.meterSign)) {
+							MAC = this.pointMac_UP.get(ps.buildingSignUP + "-" + ps.meterSign);
+						}
+						// 已得到控制设备的反馈状态,移除
+						if (ps.recTime > 0 && ps.sendTime > 0 && (ps.status != null && ps.status.length() > 0)) {
+							sb.append(ps.buildingSignUP).append(";").append(MAC).append(";").append("pointsetack")
+									.append(";").append(Constant.format(new Date())).append(";").append(ps.ID)
+									.append(";").append(ps.meterSign).append(";").append(ps.funcID).append(";")
+									.append(ps.status).append(";");
+							this.uploadServer.AppendToSend(ps.address, new Packet(sb.toString()));
+							Constant.sendSend += sb.toString().length();
+							HBaseUtil.BatchInsert_Set2(Constant.agent, Constant.BuildingDB.get(ps.buildingSign), ps);
+							LogUtil.info(
+									this.uploadServer.Name() + " AppendToSend: " + ps.address + " " + sb.toString());
+
+							String id = ps.buildingSignUP + "-" + ps.mac + "-" + ps.ID;
+							HBaseUtil.Insert_ps_virtual_point_log(Constant.agent,
+									Constant.BuildingDB.get(ps.buildingSignUP), ps.buildingSignUP, ps.meterSign,
+									Long.valueOf(ps.funcID), this.getupdown("pointsetack"), Constant.format(new Date()),
+									Long.valueOf(ps.ID), "pointsetack", "finish:finish:" + ps.status);
+
+							synchronized (Constant.ps_ID2Status) {
+								Constant.ps_ID2Status.put(id, "finish:finish:" + ps.status);
+							}
+
+							// CSVUtil.insert(Constant.setlog, new
+							// StringBuffer(Constant.format_Log(new Date())
+							// + ": "
+							// + ps.meterSign + "." + ps.funcID + " set " +
+							// ps.dataSet + " " + ps.status + "!"
+							// + " Consuming:" + (ps.recTime - ps.sendTime <
+							// 0 ? 0 : ps.recTime - ps.sendTime)
+							// + "ms!" + "\r\n"));
+							it.remove();
+
+						} // 超时未得到设备的反馈状态,移除
+						else if (ps.recTime == 0 && ps.sendTime > 0
+								&& (System.currentTimeMillis() - ps.sendTime > Constant.getWaitingtime())) {
+							sb.append(ps.buildingSignUP).append(";").append(MAC).append(";").append("pointsetack")
+									.append(";").append(Constant.format(new Date())).append(";").append(ps.ID)
+									.append(";").append(ps.meterSign).append(";").append(ps.funcID).append(";")
+									.append("fail:timeout").append(";");
+							this.uploadServer.AppendToSend(ps.address, new Packet(sb.toString()));
+							Constant.sendSend += sb.toString().length();
+							HBaseUtil.BatchInsert_Set3(Constant.agent, Constant.BuildingDB.get(ps.buildingSign), ps);
+							LogUtil.info(
+									this.uploadServer.Name() + " AppendToSend: " + ps.address + " " + sb.toString());
+							String id = ps.buildingSignUP + "-" + ps.mac + "-" + ps.ID;
+
+							// LogUtil.info("InsertPointCommunicationLog" + "
+							// begin");
+							HBaseUtil.Insert_ps_virtual_point_log(Constant.agent,
+									Constant.BuildingDB.get(ps.buildingSignUP), ps.buildingSignUP, ps.meterSign,
+									Long.valueOf(ps.funcID), this.getupdown("pointsetack"), Constant.format(new Date()),
+									Long.valueOf(ps.ID), "pointsetack", "finish:finish:" + "fail:timeout");
+
+							synchronized (Constant.ps_ID2Status) {
+								Constant.ps_ID2Status.put(id, "finish:finish:" + "fail:timeout");
+							}
+							// LogUtil.info("InsertPointCommunicationLog" + "
+							// end");
+
+							// CSVUtil.insert(Constant.setlog,
+							// new StringBuffer(
+							// Constant.format_Log(new Date()) + ": " +
+							// ps.meterSign + "." + ps.funcID
+							// + " set " + ps.dataSet + " fail(timeout) " +
+							// "!" + " Consuming:"
+							// + (System.currentTimeMillis() - ps.sendTime)
+							// + "ms!" + "\r\n"));
+							it.remove();
+						}
+						if (sb.length() > 0) {
+							// Constant.CollectorPointSetList.remove(ps.buildingSign
+							// + ";" + ps.ID + ";");
+							break;
+							// CSVUtil.insert(Constant.setlog, new
+							// StringBuffer(Constant.format_Log(new Date())
+							// + ": "
+							// + this.uploadClient.Name() + " AppendToSend:
+							// " + sb.toString() + "\r\n"));
+						}
+					}
+				}
+			}
+
+		} catch (Exception e) {
+			// TODO Auto-generated catch block
+			LogUtil.error("SendThread.returnPointSet2.Exception: " + e.toString());
+		}
+	}
+
 	private void handlePopRece(PacketEntity packageRece) {
 		// TODO Auto-generated method stub
 		if ((packageRece != null) && (packageRece.content.packetString != null)
@@ -247,8 +521,8 @@ public class SendThread extends Thread {
 						ps.recTime = 0;
 						ps.buildingSign = rList[0];
 						ps.buildingSignUP = rList[0];
-						//楼号有转换规则,且转换后的楼-表有数据才进行转换
-						//解决一楼多楼号时总有部分仪表无法下发的问题
+						// 楼号有转换规则,且转换后的楼-表有数据才进行转换
+						// 解决一楼多楼号时总有部分仪表无法下发的问题
 						if (Constant.BuildingConv.containsKey(ps.buildingSign) && Constant.Meter_Collector_Control
 								.containsKey(Constant.BuildingConv.get(ps.buildingSign) + "-" + ps.meterSign)) {
 							ps.buildingSign = Constant.BuildingConv.get(ps.buildingSign);

+ 53 - 5
collect/src/main/java/com/saga/util/Constant.java

@@ -14,6 +14,7 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 
 import org.dom4j.Document;
 import org.dom4j.Element;
@@ -53,15 +54,18 @@ public class Constant {
 	public static Map<String, Integer> RecordAddress = new HashMap<String, Integer>();
 	public static List<Record> RecordList = new ArrayList<Record>();
 
-	public static Map<String, Long> ps_NodeStatus = new HashMap<String, Long>();
+	// public static Map<String, Long> ps_NodeStatus = new HashMap<String,
+	// Long>();
 	public static Map<String, Long> ps_PointStatusUp = new HashMap<String, Long>();
 	public static Map<String, Long> ps_PhysicalStatus = new HashMap<String, Long>();
 	public static Map<String, Long> ps_ProjectStatus = new HashMap<String, Long>();
 	public static Map<String, Long> ps_MeterStatusUp = new HashMap<String, Long>();
+	public static Map<String, Long> ps_PointStatusDown = new HashMap<String, Long>();
+	public static Map<String, Long> ps_MeterStatusDown = new HashMap<String, Long>();
 
-	public static Map<String, String> ps_PhysicalList = new HashMap<String, String>();
-	public static Map<String, String> ps_ProjectList = new HashMap<String, String>();
-	public static Map<String, String> ps_MeterListUp = new HashMap<String, String>();
+	public static Map<String, List<String>> ps_PhysicalList = new HashMap<String, List<String>>();
+	public static Map<String, List<String>> ps_ProjectList = new HashMap<String, List<String>>();
+	public static Map<String, List<String>> ps_MeterList = new HashMap<String, List<String>>();
 
 	public static Map<String, String> pointInfo = new HashMap<String, String>();
 	public static Map<String, Data> nodeCache = new HashMap<String, Data>();
@@ -83,7 +87,7 @@ public class Constant {
 	public static Map<String, String> pointMAC = new HashMap<String, String>();
 
 	static {
-		String resource = "/config.json";
+		String resource = "\\config.json";
 		String path = null;
 		JsonParser config = new JsonParser();
 
@@ -257,6 +261,9 @@ public class Constant {
 	}
 
 	public synchronized static String format(Date value) {
+		if (value == null) {
+			return "null";
+		}
 		return SIMPLEDATEFORMAT.format(value);
 	}
 
@@ -272,6 +279,47 @@ public class Constant {
 		return SIMPLEDATEFORMAT_DAY.parse(value);
 	}
 
+	public static void showStatus() {
+		// TODO Auto-generated method stub
+		LogUtil.info("ps_PointStatusUp:");
+		for (Entry<String, Long> d : Constant.ps_PointStatusUp.entrySet()) {
+			LogUtil.info(d.getKey() + " " + Constant.format(new Date(d.getValue())));
+		}
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_PhysicalStatus:");
+		for (Entry<String, Long> d : Constant.ps_PhysicalStatus.entrySet()) {
+			LogUtil.info(d.getKey() + " " + Constant.format(new Date(d.getValue())));
+		}
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_ProjectStatus:");
+		for (Entry<String, Long> d : Constant.ps_ProjectStatus.entrySet()) {
+			LogUtil.info(d.getKey() + " " + Constant.format(new Date(d.getValue())));
+		}
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_MeterStatusUp:");
+		for (Entry<String, Long> d : Constant.ps_MeterStatusUp.entrySet()) {
+			LogUtil.info(d.getKey() + " " + Constant.format(new Date(d.getValue())));
+		}
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_PointStatusDown:");
+		for (Entry<String, Long> d : Constant.ps_PointStatusDown.entrySet()) {
+			LogUtil.info(d.getKey() + " " + Constant.format(new Date(d.getValue())));
+		}
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_MeterStatusDown:");
+		for (Entry<String, Long> d : Constant.ps_MeterStatusDown.entrySet()) {
+			LogUtil.info(d.getKey() + " " + Constant.format(new Date(d.getValue())));
+		}
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_PhysicalList:\n" + Constant.ps_PhysicalList);
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_ProjectList:\n" + Constant.ps_ProjectList);
+		LogUtil.info("-----------------------------------------------------");
+		LogUtil.info("ps_MeterList:\n" + Constant.ps_MeterList);
+		LogUtil.info("-----------------------------------------------------");
+
+	}
+
 	public static List<String> getConfigFile(String directoryPath, String directoryKey, String fileKey) {
 		List<String> list = new ArrayList<String>();
 		File baseFile = new File(directoryPath);

+ 324 - 52
collect/src/main/java/com/saga/util/HBaseUtil.java

@@ -31,6 +31,7 @@ public class HBaseUtil {
 		// 10 collect_cycle String 采集频率 采集器上传频率
 		// 11 data_type String 点位类型 Acc:累积变量;Inst:瞬时变量;Boolv:通断变量;
 		// Enumv:状态变量;Step:阶跃变量
+		// 12 unit String 单位
 		JSONObject Criteria = new JSONObject();
 		Criteria.put("building", project);
 
@@ -40,6 +41,7 @@ public class HBaseUtil {
 		wrapperObject.put("Datatable", tb);
 		wrapperObject.put("Criteria", Criteria);
 
+		LogUtil.info("Reload Project:" + project + "  dy_pointlist Begin...");
 		JSONObject queryResult = agent.Query(wrapperObject);
 		JSONArray resultContent = (JSONArray) queryResult.get("Content");
 		for (int i = 0; i < resultContent.size(); i++) {
@@ -50,14 +52,80 @@ public class HBaseUtil {
 			String address_2 = (String) dataItem.get("address_2");
 			String address_3 = (String) dataItem.get("address_3");
 			String address_4 = (String) dataItem.get("address_4");
-			String key = project + ";" + address_1 + ";" + address_2 + ";" + address_3 + ";" + address_4;
+			String key4 = project + ";" + address_1 + ";" + address_2 + ";" + address_3 + ";" + address_4;
+			String key3 = project + ";" + address_1 + ";" + address_2 + ";" + address_3 + ";" + "";
+			String key2 = project + ";" + address_1 + ";" + address_2 + ";" + "" + ";" + "";
+			String key1 = project + ";" + address_1 + ";" + "" + ";" + "" + ";" + "";
+
 			String value = project + "-" + meter + "." + funcid;
-			Constant.ps_PhysicalList.put(key, value);
-			Constant.ps_MeterListUp.put(meter, value);
-			Constant.ps_ProjectList.put(project, value);
+			if (!Constant.ps_PhysicalList.containsKey(key4)) {
+				Constant.ps_PhysicalList.put(key4, new ArrayList<String>());
+			}
+			if (!Constant.ps_PhysicalList.containsKey(key3)) {
+				Constant.ps_PhysicalList.put(key3, new ArrayList<String>());
+			}
+			if (!Constant.ps_PhysicalList.containsKey(key2)) {
+				Constant.ps_PhysicalList.put(key2, new ArrayList<String>());
+			}
+			if (!Constant.ps_PhysicalList.containsKey(key1)) {
+				Constant.ps_PhysicalList.put(key1, new ArrayList<String>());
+			}
 
+			if (!Constant.ps_MeterList.containsKey(project + "-" + meter)) {
+				Constant.ps_MeterList.put(project + "-" + meter, new ArrayList<String>());
+			}
+			if (!Constant.ps_ProjectList.containsKey(project)) {
+				Constant.ps_ProjectList.put(project, new ArrayList<String>());
+			}
+			if (!Constant.ps_PhysicalList.get(key4).contains(value)) {
+				Constant.ps_PhysicalList.get(key4).add(value);
+			}
+			if (!Constant.ps_PhysicalList.get(key3).contains(value)) {
+				Constant.ps_PhysicalList.get(key3).add(value);
+			}
+			if (!Constant.ps_PhysicalList.get(key2).contains(value)) {
+				Constant.ps_PhysicalList.get(key2).add(value);
+			}
+			if (!Constant.ps_PhysicalList.get(key1).contains(value)) {
+				Constant.ps_PhysicalList.get(key1).add(value);
+			}
+
+			Constant.ps_MeterList.get(project + "-" + meter).add(value);
+			Constant.ps_ProjectList.get(project).add(value);
+			checkStatus(project, meter, funcid, address_1, address_2, address_3, address_4);
+		}
+
+		LogUtil.info("Reload Project:" + project + "  dy_pointlist End...");
+	}
+
+	private static void checkStatus(String project, String meter, long funcid, String address_1, String address_2,
+			String address_3, String address_4) {
+		// TODO Auto-generated method stub
+		if (!Constant.ps_PointStatusUp.containsKey(project + "-" + meter + "." + funcid)) {
+			Constant.ps_PointStatusUp.put((project + "-" + meter + "." + funcid), null);
+		}
+		if (!Constant.ps_MeterStatusUp.containsKey(project + "-" + meter)) {
+			Constant.ps_MeterStatusUp.put((project + "-" + meter), null);
+		}
+		if (!Constant.ps_ProjectStatus.containsKey(project)) {
+			Constant.ps_ProjectStatus.put((project), null);
+		}
+		String key4 = project + ";" + address_1 + ";" + address_2 + ";" + address_3 + ";" + address_4;
+		String key3 = project + ";" + address_1 + ";" + address_2 + ";" + address_3 + ";" + "";
+		String key2 = project + ";" + address_1 + ";" + address_2 + ";" + "" + ";" + "";
+		String key1 = project + ";" + address_1 + ";" + "" + ";" + "" + ";" + "";
+		if (!Constant.ps_PhysicalStatus.containsKey(key4)) {
+			Constant.ps_PhysicalStatus.put((key4), null);
+		}
+		if (!Constant.ps_PhysicalStatus.containsKey(key3)) {
+			Constant.ps_PhysicalStatus.put((key3), null);
+		}
+		if (!Constant.ps_PhysicalStatus.containsKey(key2)) {
+			Constant.ps_PhysicalStatus.put((key2), null);
+		}
+		if (!Constant.ps_PhysicalStatus.containsKey(key1)) {
+			Constant.ps_PhysicalStatus.put((key1), null);
 		}
-		LogUtil.info("Reload Project:" + project + "  CollectorStatus End...");
 	}
 
 	public static void LoadDB2Cache_ps_physical_status(ZillionAgent agent, String db, String tb, String project)
@@ -90,19 +158,25 @@ public class HBaseUtil {
 			String address_4 = (String) dataItem.get("address_4");
 			String key = project + ";" + address_1 + ";" + address_2 + ";" + address_3 + ";" + address_4;
 			String receivetime = (String) dataItem.get("receivetime");
-			Constant.ps_NodeStatus.put(key, Constant.parse(receivetime).getTime());
+			Constant.ps_PhysicalStatus.put(key, Constant.parse(receivetime).getTime());
 
 		}
 		LogUtil.info("Reload Project:" + project + "  ps_physical_status End...");
 
 	}
 
-	public static void GetCollectorStatus(ZillionAgent agent, String db, String tb, String project) throws Exception {
-
+	public static void LoadDB2Cache_ps_virtual_point_status(ZillionAgent agent, String db, String tb, String project)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 funcid Long 功能号 主键
+		// 4 updown String 上/下行 主键
+		// 5 status String 状态
+		// 6 receivetime String 时间
+		// 7 detail String 详情
 		JSONObject Criteria = new JSONObject();
 		Criteria.put("project", project);
-		Criteria.put("collector",
-				Constant.getLocalHostLANAddress().getHostAddress() + ":" + Constant.TypePort.get("Down"));
 
 		JSONObject wrapperObject = new JSONObject();
 		wrapperObject.put("QueryType", "select");
@@ -111,26 +185,36 @@ public class HBaseUtil {
 		wrapperObject.put("Criteria", Criteria);
 		JSONObject queryResult = agent.Query(wrapperObject);
 		JSONArray resultContent = (JSONArray) queryResult.get("Content");
-		LogUtil.info("Reload Project:" + project + "  CollectorStatus Begin...");
+		LogUtil.info("Reload Project:" + project + " ps_virtual_point_status Begin...");
+
 		for (int i = 0; i < resultContent.size(); i++) {
 			JSONObject dataItem = (JSONObject) resultContent.get(i);
-			String node = (String) dataItem.get("gateway");
-			String key = project + "-" + node;
-			String receivetime = (String) dataItem.get("receivetime");
-			Constant.ps_NodeStatus.put(key, Constant.parse(receivetime).getTime());
-			LogUtil.info("project " + project + " collector " + Constant.getLocalHostLANAddress().getHostAddress() + ":"
-					+ Constant.TypePort.get("Down") + " gateway " + node + " " + System.currentTimeMillis() + " "
-					+ Constant.parse(receivetime).getTime());
+			String meter = (String) dataItem.get("meter");
+			long funcid = (long) dataItem.get("funcid");
+			String updown = (String) dataItem.get("updown");
+			if ("up".equalsIgnoreCase(updown)) {
+				String key = project + "-" + meter + "." + funcid;
+				String receivetime = (String) dataItem.get("receivetime");
+				Constant.ps_PointStatusUp.put(key, Constant.parse(receivetime).getTime());
+				LogUtil.info("project " + project + " meter " + meter + " funcid " + funcid + " "
+						+ System.currentTimeMillis() + " " + Constant.parse(receivetime).getTime());
+			}
 		}
-		LogUtil.info("Reload Project:" + project + "  CollectorStatus End...");
+		// }
+		LogUtil.info("Reload Project:" + project + "  ps_virtual_point_status End...");
 	}
 
-	public static void GetPointStatus(ZillionAgent agent, String db, String tb, String project) throws Exception {
-
+	public static void LoadDB2Cache_ps_virtual_meter_status(ZillionAgent agent, String db, String tb, String project)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 updown String 上/下行 主键
+		// 4 status String 状态
+		// 5 receivetime String 时间
+		// 6 detail String 详情
 		JSONObject Criteria = new JSONObject();
 		Criteria.put("project", project);
-		Criteria.put("collector",
-				Constant.getLocalHostLANAddress().getHostAddress() + ":" + Constant.TypePort.get("Down"));
 
 		JSONObject wrapperObject = new JSONObject();
 		wrapperObject.put("QueryType", "select");
@@ -139,27 +223,128 @@ public class HBaseUtil {
 		wrapperObject.put("Criteria", Criteria);
 		JSONObject queryResult = agent.Query(wrapperObject);
 		JSONArray resultContent = (JSONArray) queryResult.get("Content");
-		LogUtil.info("Reload Project:" + project + " PointStatus Begin...");
-		// synchronized (Constant.pointListUp) {
+		LogUtil.info("Reload Project:" + project + " ps_virtual_meter_status Begin...");
+
 		for (int i = 0; i < resultContent.size(); i++) {
 			JSONObject dataItem = (JSONObject) resultContent.get(i);
-			String node = (String) dataItem.get("gateway");
 			String meter = (String) dataItem.get("meter");
-			String funcid = (String) dataItem.get("funcid");
 			String updown = (String) dataItem.get("updown");
 			if ("up".equalsIgnoreCase(updown)) {
-				String key = project + "-" + node + "-" + meter + "." + funcid;
+				String key = project + "-" + meter;
 				String receivetime = (String) dataItem.get("receivetime");
-				Constant.ps_PointStatusUp.put(key, Constant.parse(receivetime).getTime());
-				LogUtil.info("project " + project + " collector " + Constant.getLocalHostLANAddress().getHostAddress()
-						+ ":" + Constant.TypePort.get("Down") + " gateway " + node + " meter " + meter + " funcid "
-						+ funcid + " " + System.currentTimeMillis() + " " + Constant.parse(receivetime).getTime());
+				Constant.ps_MeterStatusUp.put(key, Constant.parse(receivetime).getTime());
+				LogUtil.info("project " + project + " meter " + meter + " " + System.currentTimeMillis() + " "
+						+ Constant.parse(receivetime).getTime());
 			}
 		}
-		// }
-		LogUtil.info("Reload Project:" + project + "  PointStatus End...");
+		LogUtil.info("Reload Project:" + project + "  ps_virtual_meter_status End...");
+	}
+
+	public static void LoadDB2Cache_ps_virtual_project_status(ZillionAgent agent, String db, String tb, String project)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 status String 状态
+		// 3 receivetime String 时间
+		// 4 detail String 详情
+		JSONObject Criteria = new JSONObject();
+		Criteria.put("project", project);
+
+		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");
+		LogUtil.info("Reload Project:" + project + " ps_virtual_project_status Begin...");
+
+		for (int i = 0; i < resultContent.size(); i++) {
+			JSONObject dataItem = (JSONObject) resultContent.get(i);
+
+			String key = project;
+			String receivetime = (String) dataItem.get("receivetime");
+			Constant.ps_ProjectStatus.put(key, Constant.parse(receivetime).getTime());
+			LogUtil.info("project " + project + " " + System.currentTimeMillis() + " "
+					+ Constant.parse(receivetime).getTime());
+
+		}
+		LogUtil.info("Reload Project:" + project + "  ps_virtual_project_status End...");
 	}
 
+	// public static void GetCollectorStatus(ZillionAgent agent, String db,
+	// String tb, String project) throws Exception {
+	//
+	// JSONObject Criteria = new JSONObject();
+	// Criteria.put("project", project);
+	// Criteria.put("collector",
+	// Constant.getLocalHostLANAddress().getHostAddress() + ":" +
+	// Constant.TypePort.get("Down"));
+	//
+	// 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");
+	// LogUtil.info("Reload Project:" + project + " CollectorStatus Begin...");
+	// for (int i = 0; i < resultContent.size(); i++) {
+	// JSONObject dataItem = (JSONObject) resultContent.get(i);
+	// String node = (String) dataItem.get("gateway");
+	// String key = project + "-" + node;
+	// String receivetime = (String) dataItem.get("receivetime");
+	// Constant.ps_NodeStatus.put(key, Constant.parse(receivetime).getTime());
+	// LogUtil.info("project " + project + " collector " +
+	// Constant.getLocalHostLANAddress().getHostAddress() + ":"
+	// + Constant.TypePort.get("Down") + " gateway " + node + " " +
+	// System.currentTimeMillis() + " "
+	// + Constant.parse(receivetime).getTime());
+	// }
+	// LogUtil.info("Reload Project:" + project + " CollectorStatus End...");
+	// }
+
+	// public static void GetPointStatus(ZillionAgent agent, String db, String
+	// tb, String project) throws Exception {
+	//
+	// JSONObject Criteria = new JSONObject();
+	// Criteria.put("project", project);
+	// Criteria.put("collector",
+	// Constant.getLocalHostLANAddress().getHostAddress() + ":" +
+	// Constant.TypePort.get("Down"));
+	//
+	// 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");
+	// LogUtil.info("Reload Project:" + project + " PointStatus Begin...");
+	// // synchronized (Constant.pointListUp) {
+	// for (int i = 0; i < resultContent.size(); i++) {
+	// JSONObject dataItem = (JSONObject) resultContent.get(i);
+	// String node = (String) dataItem.get("gateway");
+	// String meter = (String) dataItem.get("meter");
+	// String funcid = (String) dataItem.get("funcid");
+	// String updown = (String) dataItem.get("updown");
+	// if ("up".equalsIgnoreCase(updown)) {
+	// String key = project + "-" + node + "-" + meter + "." + funcid;
+	// String receivetime = (String) dataItem.get("receivetime");
+	// Constant.ps_PointStatusUp.put(key,
+	// Constant.parse(receivetime).getTime());
+	// LogUtil.info("project " + project + " collector " +
+	// Constant.getLocalHostLANAddress().getHostAddress()
+	// + ":" + Constant.TypePort.get("Down") + " gateway " + node + " meter " +
+	// meter + " funcid "
+	// + funcid + " " + System.currentTimeMillis() + " " +
+	// Constant.parse(receivetime).getTime());
+	// }
+	// }
+	// // }
+	// LogUtil.info("Reload Project:" + project + " PointStatus End...");
+	// }
+
 	public static Map<String, List<Integer>> GetDBPointList(ZillionAgent agent, String db, String tb, String building)
 			throws Exception {
 
@@ -491,6 +676,30 @@ public class HBaseUtil {
 		}
 	}
 
+	public static void CreateTable_fjd_0_buildingcomputetime(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 building String 楼号 主键
+		// 2 compute_time String 计算时间
+		// 3 first_compute_time String 首次计算时间
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("building", "String"));
+			Columns.add(CreateJSON("compute_time", "String"));
+			Columns.add(CreateJSON("first_compute_time", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("building");
+
+			JSONObject Definition = new JSONObject();
+			Definition.put("Columns", Columns);
+			Definition.put("Key", Key);
+			JSONArray Indexes = new JSONArray();
+			agent.DDL_AddTable(DB, tableName, Definition, Indexes);
+		}
+	}
+
 	public static void CreateTable_dy_pointlist(ZillionAgent agent, String DB, String tableName) throws Exception {
 		// 序号 名称 数据类型 释义 备注
 		// 1 building String 楼号 主键
@@ -499,12 +708,17 @@ public class HBaseUtil {
 		// 4 system String 系统
 		// 5 source String 数据来源
 		// 6 address_1 String 一级地址 采集器网关ID
-		// 7 address_2 String 二级地址 总线端口号
-		// 8 address_3 String 三级地址 仪表地址
-		// 9 address_4 String 四级地址
-		// 10 collect_cycle String 采集频率 采集器上传频率
-		// 11 data_type String 点位类型
-		// Acc:累积变量;Inst:瞬时变量;Boolv:通断变量;Enumv:状态变量;Step:阶跃变量
+		// 7 ip_port String 一级地址IP_PORT 如无端口,可缺省
+		// 8 address_2 String 二级地址 总线端口号
+		// 9 address_3 String 三级地址 仪表地址
+		// 10 address_4 String 四级地址
+		// 11 collect_cycle String 采集频率 采集器上传频率
+		// 12 data_type String 点位类型 Acc:累积变量;Inst:瞬时变量;Boolv:通断变量;
+		// Enumv:状态变量;Step:阶跃变量
+		// 13 unit String 单位
+		// 14 remark1 String 备注1 预留
+		// 15 remark2 String 备注2 预留
+		// 16 remark3 String 备注3 预留
 
 		if (!tableExist(agent, DB, tableName)) {
 			JSONArray Columns = new JSONArray();
@@ -514,11 +728,16 @@ public class HBaseUtil {
 			Columns.add(CreateJSON("system", "String"));
 			Columns.add(CreateJSON("source", "String"));
 			Columns.add(CreateJSON("address_1", "String"));
+			Columns.add(CreateJSON("ip_port", "String"));
 			Columns.add(CreateJSON("address_2", "String"));
 			Columns.add(CreateJSON("address_3", "String"));
 			Columns.add(CreateJSON("address_4", "String"));
 			Columns.add(CreateJSON("collect_cycle", "String"));
 			Columns.add(CreateJSON("data_type", "String"));
+			Columns.add(CreateJSON("unit", "String"));
+			Columns.add(CreateJSON("remark1", "String"));
+			Columns.add(CreateJSON("remark2", "String"));
+			Columns.add(CreateJSON("remark3", "String"));
 
 			JSONArray Key = new JSONArray();
 			Key.add("building");
@@ -529,6 +748,14 @@ public class HBaseUtil {
 			Definition.put("Columns", Columns);
 			Definition.put("Key", Key);
 			JSONArray Indexes = new JSONArray();
+
+			// Indexes.add("building");
+			// Indexes.add("address_1");
+			// Indexes.add("address_2");
+			// Indexes.add("address_3");
+			// Indexes.add("address_4");
+			// Indexes.add("meter");
+			// Indexes.add("funcid");
 			agent.DDL_AddTable(DB, tableName, Definition, Indexes);
 		}
 	}
@@ -1098,9 +1325,45 @@ public class HBaseUtil {
 		}
 	}
 
+	public static void Insert_fjd_0_buildingcomputetime(ZillionAgent agent, String DB, String building,
+			String compute_time, String first_compute_time) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 building String 楼号 主键
+		// 2 compute_time String 计算时间
+		// 3 first_compute_time String 首次计算时间
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("building", building);
+				InsertObject.put("compute_time", compute_time);
+				InsertObject.put("first_compute_time", first_compute_time);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_put");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "fjd_0_buildingcomputetime");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_fjd_0_buildingcomputetime.Exception: " + e.toString());
+		}
+	}
+
 	public static void Insert_dy_pointlist(ZillionAgent agent, String DB, String building, String meter, long funcid,
-			String system, String source, String address_1, String address_2, String address_3, String address_4,
-			String collect_cycle, String data_type) throws Exception {
+			String system, String source, String address_1, String ip_port, String address_2, String address_3,
+			String address_4, String collect_cycle, String data_type, String unit) throws Exception {
 		// 序号 名称 数据类型 释义 备注
 		// 1 building String 楼号 主键
 		// 2 meter String 表号 主键
@@ -1108,12 +1371,17 @@ public class HBaseUtil {
 		// 4 system String 系统
 		// 5 source String 数据来源
 		// 6 address_1 String 一级地址 采集器网关ID
-		// 7 address_2 String 二级地址 总线端口号
-		// 8 address_3 String 三级地址 仪表地址
-		// 9 address_4 String 四级地址
-		// 10 collect_cycle String 采集频率 采集器上传频率
-		// 11 data_type String 点位类型 Acc:累积变量;Inst:瞬时变量;Boolv:通断变量;
+		// 7 ip_port String 一级地址IP_PORT 如无端口,可缺省
+		// 8 address_2 String 二级地址 总线端口号
+		// 9 address_3 String 三级地址 仪表地址
+		// 10 address_4 String 四级地址
+		// 11 collect_cycle String 采集频率 采集器上传频率
+		// 12 data_type String 点位类型 Acc:累积变量;Inst:瞬时变量;Boolv:通断变量;
 		// Enumv:状态变量;Step:阶跃变量
+		// 13 unit String 单位
+		// 14 remark1 String 备注1 预留
+		// 15 remark2 String 备注2 预留
+		// 16 remark3 String 备注3 预留
 
 		JSONArray InsertObjects = new JSONArray();
 		try {
@@ -1129,18 +1397,20 @@ public class HBaseUtil {
 				InsertObject.put("system", system);
 				InsertObject.put("source", source);
 				InsertObject.put("address_1", address_1);
+				InsertObject.put("ip_port", ip_port);
 				InsertObject.put("address_2", address_2);
 				InsertObject.put("address_3", address_3);
 				InsertObject.put("address_4", address_4);
 				InsertObject.put("collect_cycle", collect_cycle);
 				InsertObject.put("data_type", data_type);
+				InsertObject.put("unit", unit);
 
 				InsertObjects.add(InsertObject);
 
 				JSONObject wrapperObject = new JSONObject();
 
 				wrapperObject = new JSONObject();
-				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("QueryType", "batch_put");
 				wrapperObject.put("Database", DB);
 				wrapperObject.put("Datatable", "dy_pointlist");
 				wrapperObject.put("InsertObjects", InsertObjects);
@@ -1187,7 +1457,7 @@ public class HBaseUtil {
 				JSONObject wrapperObject = new JSONObject();
 
 				wrapperObject = new JSONObject();
-				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("QueryType", "batch_put");
 				wrapperObject.put("Database", DB);
 				wrapperObject.put("Datatable", "ps_physical_status");
 				wrapperObject.put("InsertObjects", InsertObjects);
@@ -1276,7 +1546,7 @@ public class HBaseUtil {
 				JSONObject wrapperObject = new JSONObject();
 
 				wrapperObject = new JSONObject();
-				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("QueryType", "batch_put");
 				wrapperObject.put("Database", DB);
 				wrapperObject.put("Datatable", "ps_virtual_point_status");
 				wrapperObject.put("InsertObjects", InsertObjects);
@@ -1334,6 +1604,7 @@ public class HBaseUtil {
 				wrapperObject.put("Datatable", "ps_virtual_point_log");
 				wrapperObject.put("InsertObjects", InsertObjects);
 				agent.Query(wrapperObject);
+				LogUtil.info(wrapperObject.toJSONString());
 			}
 
 		} catch (Exception e) {
@@ -1371,7 +1642,7 @@ public class HBaseUtil {
 				JSONObject wrapperObject = new JSONObject();
 
 				wrapperObject = new JSONObject();
-				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("QueryType", "batch_put");
 				wrapperObject.put("Database", DB);
 				wrapperObject.put("Datatable", "ps_virtual_meter_status");
 				wrapperObject.put("InsertObjects", InsertObjects);
@@ -1449,7 +1720,7 @@ public class HBaseUtil {
 				JSONObject wrapperObject = new JSONObject();
 
 				wrapperObject = new JSONObject();
-				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("QueryType", "batch_put");
 				wrapperObject.put("Database", DB);
 				wrapperObject.put("Datatable", "ps_virtual_project_status");
 				wrapperObject.put("InsertObjects", InsertObjects);
@@ -1490,6 +1761,7 @@ public class HBaseUtil {
 				wrapperObject.put("Datatable", "ps_virtual_project_log");
 				wrapperObject.put("InsertObjects", InsertObjects);
 				agent.Query(wrapperObject);
+				LogUtil.info(wrapperObject.toJSONString());
 			}
 
 		} catch (Exception e) {