Browse Source

V1.0版本:
仅含网关节点通讯日志及点位通讯日志。

Jay Lee 6 years ago
parent
commit
8672caa834

+ 190 - 103
collect/src/main/java/com/saga/main/Main.java

@@ -28,109 +28,16 @@ public class Main {
 
 	public static void main(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";
-			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());
-		}
-		try {
-			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()) {
-				HBaseUtil.CreateDB(Constant.agent, db);
-				Thread.sleep(Constant.Millisecond.SECOND_1);
-
-				// 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);
-				// System.exit(0);
-
-				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");
-			}
-
-			// 重新启动加载节点、点位历史列表
-			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: " + e.toString());
-				}
-			}
-
-			// 加载采集器配置文件
-			// 文件夹命名格式: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];
-				HBaseUtil.InsertPointInfo(Constant.agent, Constant.BuildingDB.get(building), building, node, point,
-						busno, site);
-				// System.out.println(building + node + point + busno + site);
-			}
+		// 初始化
+		init(args);
+		// 业务进程启动
+		threadStart();
+	}
 
-		} catch (Exception e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-			LogUtil.error("Main: " + e.toString());
-			System.exit(0);
-		}
+	private static void threadStart() {
+		// TODO Auto-generated method stub
+		// 配置实时加载进程
 		try {
 			new LoadConfigThread().start();
 			Thread.sleep(3L);
@@ -138,7 +45,7 @@ public class Main {
 			// TODO Auto-generated catch block
 			LogUtil.error("Main.LoadConfigThread: " + e.toString());
 		}
-
+		// 采集进程
 		try {
 			String ports = Constant.TypePort.get("Down");
 			new CollectThread(Integer.valueOf(ports)).start();
@@ -151,6 +58,7 @@ public class Main {
 			// 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));
@@ -174,13 +82,14 @@ public class Main {
 				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++) {
@@ -192,12 +101,14 @@ 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();
 		} catch (Exception e) {
@@ -206,4 +117,180 @@ public class Main {
 		}
 	}
 
+	private static void init(String[] args) {
+		// TODO Auto-generated method stub
+		try {
+			// ZillionAgent 启动
+			agentStart(args);
+			// 初始化数据库环境
+			initDB();
+			// 加载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 loadDB() {
+		// 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.loadDB: " + 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);
+				Thread.sleep(Constant.Millisecond.SECOND_1);
+				// dropTB(db);
+				// Thread.sleep(Constant.Millisecond.SECOND_1);
+				createTB(db);
+				Thread.sleep(Constant.Millisecond.SECOND_1);
+			} catch (InterruptedException 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";
+
+			if ("test".equalsIgnoreCase(args[0])) {
+				// zookeeper_connectionAddress =
+				// "zk01.td.com:2181,zk02.td.com:2181,zk03.td.com:2181";
+				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());
+		}
+	}
+
 }

+ 17 - 7
collect/src/main/java/com/saga/thread/down/CollectThread.java

@@ -21,6 +21,7 @@ import com.saga.util.HBaseUtil;
 
 public class CollectThread extends Thread {
 	private int port = 0;
+	private long getPointList = 0;
 	private String ip = null;
 	// private List<IServerManager> serverList = new
 	// ArrayList<IServerManager>();
@@ -351,14 +352,14 @@ public class CollectThread extends Thread {
 						// Date()));
 						Constant.nodeList.put(rList[0] + "-" + rList[1], System.currentTimeMillis());
 						// this.insertCollectorCommunicationLog(rList);
+						if (this.needGetPointList()) {
+							int id = Constant.getID();
+							server.AppendToSend(p.address,
+									new Packet(rList[0] + ";" + rList[1] + ";" + "pointcount" + ";" + "" + ";" + id));
+							LogUtil.info(server.Name() + " AppendToSend:" + p.address + " " + rList[0] + ";" + rList[1]
+									+ ";" + "pointcount" + ";" + "" + ";" + id);
+						}
 
-						// int id = Constant.getID();
-						// server.AppendToSend(p.address,
-						// new Packet(rList[0] + ";" + rList[1] + ";" +
-						// "pointcount" + ";" + "" + ";" + id));
-						// LogUtil.info(server.Name() + " AppendToSend:" +
-						// p.address + " " + rList[0] + ";" + rList[1]
-						// + ";" + "pointcount" + ";" + "" + ";" + id);
 					} else if ("pointcountack".equalsIgnoreCase(rList[2])) {
 						int count = Integer.valueOf(rList[5]);
 						int loop = (count / 50) + 1;
@@ -470,6 +471,15 @@ public class CollectThread extends Thread {
 		}
 	}
 
+	private boolean needGetPointList() {
+		// TODO Auto-generated method stub
+		if (System.currentTimeMillis() - this.getPointList > Constant.Millisecond.MINUTE_3) {
+			this.getPointList = System.currentTimeMillis();
+			return true;
+		}
+		return false;
+	}
+
 	// private void insertCollectorStatus(String[] rList) {
 	// // TODO Auto-generated method stub
 	// try {

+ 715 - 2
collect/src/main/java/com/saga/util/HBaseUtil.java

@@ -1,6 +1,5 @@
 package com.saga.util;
 
-import java.io.File;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.HashMap;
@@ -99,7 +98,9 @@ public class HBaseUtil {
 			JSONObject dataItem = (JSONObject) resultContent.get(i);
 
 			String meter = (String) dataItem.get("meter");
-			Integer funcid = (Integer) dataItem.get("funcid");
+			Long f = (long) dataItem.get("funcid");
+
+			Integer funcid = f.intValue();
 			if (!dbPointList.containsKey(meter)) {
 				dbPointList.put(meter, new ArrayList<Integer>());
 			}
@@ -409,6 +410,320 @@ public class HBaseUtil {
 		}
 	}
 
+	public static void CreateTable_dy_pointlist(ZillionAgent agent, String DB, String tableName) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 building String 楼号 主键
+		// 2 meter String 表号 主键
+		// 3 funcid Long 功能号 主键
+		// 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:阶跃变量
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("building", "String"));
+			Columns.add(CreateJSON("meter", "String"));
+			Columns.add(CreateJSON("funcid", "Long"));
+			Columns.add(CreateJSON("system", "String"));
+			Columns.add(CreateJSON("source", "String"));
+			Columns.add(CreateJSON("address_1", "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"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("building");
+			Key.add("meter");
+			Key.add("funcid");
+
+			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_ps_physical_status(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 address_1 String 一级地址 主键
+		// 3 address_2 String 二级地址 主键
+		// 4 address_3 String 三级地址 主键
+		// 5 address_4 String 四级地址 主键
+		// 6 status String 状态 conn/disconn
+		// 7 receivetime String 时间 最近通讯时间
+		// 8 detail String 详情
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("address_1", "String"));
+			Columns.add(CreateJSON("address_2", "String"));
+			Columns.add(CreateJSON("address_3", "String"));
+			Columns.add(CreateJSON("address_4", "String"));
+			Columns.add(CreateJSON("status", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("detail", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+			Key.add("address_1");
+			Key.add("address_2");
+			Key.add("address_3");
+			Key.add("address_4");
+
+			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_ps_physical_log(ZillionAgent agent, String DB, String tableName) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 address_1 String 一级地址 主键
+		// 3 address_2 String 二级地址 主键
+		// 4 address_3 String 三级地址 主键
+		// 5 address_4 String 四级地址 主键
+		// 6 receivetime String 时间 主键,触发时间
+		// 7 function String 通讯功能 主键,conn/disconn
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("address_1", "String"));
+			Columns.add(CreateJSON("address_2", "String"));
+			Columns.add(CreateJSON("address_3", "String"));
+			Columns.add(CreateJSON("address_4", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("function", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+			Key.add("address_1");
+			Key.add("address_2");
+			Key.add("address_3");
+			Key.add("address_4");
+			Key.add("receivetime");
+			Key.add("function");
+
+			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_ps_virtual_point_status(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 funcid Long 功能号 主键
+		// 4 updown String 上/下行 主键
+		// 5 status String 状态
+		// 6 receivetime String 时间
+		// 7 detail String 详情
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("meter", "String"));
+			Columns.add(CreateJSON("funcid", "Long"));
+			Columns.add(CreateJSON("updown", "String"));
+			Columns.add(CreateJSON("status", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("detail", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+			Key.add("meter");
+			Key.add("funcid");
+			Key.add("updown");
+
+			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_ps_virtual_point_log(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 funcid Long 功能号 主键
+		// 4 updown String 上/下行 主键
+		// 5 receivetime String 时间 主键
+		// 6 seq Long 报文序号 主键
+		// 7 function String 通讯功能 主键
+		// 8 feedback String 反馈 processing:rcvd(采集软件已收到)
+		// processing:sent(采集软件已发出)
+		// processing:finish:finish:+采集节点反馈标识(采集软件已收到反馈)
+		// finish:finish:+采集节点反馈标识(采集软件已反馈)
+		// 其中采集节点反馈标识,分两级,分别为:
+		// 1、success/fail
+		// 2、如success,该级空;如fail,则该级描述fail原因
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("meter", "String"));
+			Columns.add(CreateJSON("funcid", "Long"));
+			Columns.add(CreateJSON("updown", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("seq", "Long"));
+			Columns.add(CreateJSON("function", "String"));
+			Columns.add(CreateJSON("feedback", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+			Key.add("meter");
+			Key.add("funcid");
+			Key.add("updown");
+			Key.add("receivetime");
+			Key.add("seq");
+			Key.add("function");
+
+			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_ps_virtual_meter_status(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 updown String 上/下行 主键
+		// 4 status String 状态
+		// 5 receivetime String 时间
+		// 6 detail String 详情
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("meter", "String"));
+			Columns.add(CreateJSON("updown", "String"));
+			Columns.add(CreateJSON("status", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("detail", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+			Key.add("meter");
+			Key.add("updown");
+
+			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_ps_virtual_meter_log(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 updown String 上/下行 主键
+		// 4 receivetime String 时间 主键
+		// 5 function String 通讯功能 主键
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("meter", "String"));
+			Columns.add(CreateJSON("updown", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("function", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+			Key.add("meter");
+			Key.add("updown");
+			Key.add("receivetime");
+			Key.add("function");
+
+			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_ps_virtual_project_status(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 status String 状态
+		// 3 receivetime String 时间
+		// 4 detail String 详情
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("status", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("detail", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+
+			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_ps_virtual_project_log(ZillionAgent agent, String DB, String tableName)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 receivetime String 时间 主键
+		// 3 function String 通讯功能 主键
+
+		if (!tableExist(agent, DB, tableName)) {
+			JSONArray Columns = new JSONArray();
+			Columns.add(CreateJSON("project", "String"));
+			Columns.add(CreateJSON("receivetime", "String"));
+			Columns.add(CreateJSON("function", "String"));
+
+			JSONArray Key = new JSONArray();
+			Key.add("project");
+			Key.add("receivetime");
+			Key.add("function");
+
+			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_PointStatus(ZillionAgent agent, String DB, String tableName) throws Exception {
 		if (!tableExist(agent, DB, tableName)) {
 
@@ -702,6 +1017,404 @@ public class HBaseUtil {
 		}
 	}
 
+	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 {
+		// 序号 名称 数据类型 释义 备注
+		// 1 building String 楼号 主键
+		// 2 meter String 表号 主键
+		// 3 funcid Long 功能号 主键
+		// 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:阶跃变量
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("building", building);
+				InsertObject.put("meter", meter);
+				InsertObject.put("funcid", funcid);
+				InsertObject.put("system", system);
+				InsertObject.put("source", source);
+				InsertObject.put("address_1", address_1);
+				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);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "dy_pointlist");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_dy_pointlist.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_physical_status(ZillionAgent agent, String DB, String project, String address_1,
+			String address_2, String address_3, String address_4, String status, String receivetime, String detail)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 address_1 String 一级地址 主键
+		// 3 address_2 String 二级地址 主键
+		// 4 address_3 String 三级地址 主键
+		// 5 address_4 String 四级地址 主键
+		// 6 status String 状态 conn/disconn
+		// 7 receivetime String 时间 最近通讯时间
+		// 8 detail String 详情
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("address_1", address_1);
+				InsertObject.put("address_2", address_2);
+				InsertObject.put("address_3", address_3);
+				InsertObject.put("address_4", address_4);
+				InsertObject.put("status", status);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("detail", detail);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_physical_status");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_physical_status.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_physical_log(ZillionAgent agent, String DB, String project, String address_1,
+			String address_2, String address_3, String address_4, String receivetime, String function)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 address_1 String 一级地址 主键
+		// 3 address_2 String 二级地址 主键
+		// 4 address_3 String 三级地址 主键
+		// 5 address_4 String 四级地址 主键
+		// 6 receivetime String 时间 主键,触发时间
+		// 7 function String 通讯功能 主键,conn/disconn
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("address_1", address_1);
+				InsertObject.put("address_2", address_2);
+				InsertObject.put("address_3", address_3);
+				InsertObject.put("address_4", address_4);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("function", function);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_physical_log");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_physical_log.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_virtual_point_status(ZillionAgent agent, String DB, String project, String meter,
+			long funcid, String updown, String status, String receivetime, String detail) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 funcid Long 功能号 主键
+		// 4 updown String 上/下行 主键
+		// 5 status String 状态
+		// 6 receivetime String 时间
+		// 7 detail String 详情
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("meter", meter);
+				InsertObject.put("funcid", funcid);
+				InsertObject.put("updown", updown);
+				InsertObject.put("status", status);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("detail", detail);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_virtual_point_status");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_virtual_point_status.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_virtual_point_log(ZillionAgent agent, String DB, String project, String meter,
+			long funcid, String updown, String receivetime, long seq, String function, String feedback)
+			throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 funcid Long 功能号 主键
+		// 4 updown String 上/下行 主键
+		// 5 receivetime String 时间 主键
+		// 6 seq Long 报文序号 主键
+		// 7 function String 通讯功能 主键
+		// 8 feedback String 反馈 processing:rcvd(采集软件已收到)
+		// processing:sent(采集软件已发出)
+		// processing:finish:finish:+采集节点反馈标识(采集软件已收到反馈)
+		// finish:finish:+采集节点反馈标识(采集软件已反馈)
+		// 其中采集节点反馈标识,分两级,分别为:
+		// 1、success/fail
+		// 2、如success,该级空;如fail,则该级描述fail原因
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("meter", meter);
+				InsertObject.put("funcid", funcid);
+				InsertObject.put("updown", updown);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("seq", seq);
+				InsertObject.put("function", function);
+				InsertObject.put("feedback", feedback);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_virtual_point_log");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_virtual_point_log.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_virtual_meter_status(ZillionAgent agent, String DB, String project, String meter,
+			String updown, String status, String receivetime, String detail) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 updown String 上/下行 主键
+		// 4 status String 状态
+		// 5 receivetime String 时间
+		// 6 detail String 详情
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("meter", meter);
+				InsertObject.put("updown", updown);
+				InsertObject.put("status", status);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("detail", detail);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_virtual_meter_status");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_virtual_meter_status.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_virtual_meter_log(ZillionAgent agent, String DB, String project, String meter,
+			String updown, String receivetime, String function) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 meter String 仪表 主键
+		// 3 updown String 上/下行 主键
+		// 4 receivetime String 时间 主键
+		// 5 function String 通讯功能 主键
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("meter", meter);
+				InsertObject.put("updown", updown);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("function", function);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_virtual_meter_log");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_virtual_meter_log.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_virtual_project_status(ZillionAgent agent, String DB, String project, String status,
+			String receivetime, String detail) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 status String 状态
+		// 3 receivetime String 时间
+		// 4 detail String 详情
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("status", status);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("detail", detail);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_virtual_project_status");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_virtual_project_status.Exception: " + e.toString());
+		}
+	}
+
+	public static void Insert_ps_virtual_project_log(ZillionAgent agent, String DB, String project, String receivetime,
+			String function) throws Exception {
+		// 序号 名称 数据类型 释义 备注
+		// 1 project String 项目名 主键
+		// 2 receivetime String 时间 主键
+		// 3 function String 通讯功能 主键
+
+		JSONArray InsertObjects = new JSONArray();
+		try {
+			if (DB == null) {
+				DB = Constant.defaultDB;
+			}
+			if (DB != null) {
+
+				JSONObject InsertObject = new JSONObject();
+				InsertObject.put("project", project);
+				InsertObject.put("receivetime", receivetime);
+				InsertObject.put("function", function);
+
+				InsertObjects.add(InsertObject);
+
+				JSONObject wrapperObject = new JSONObject();
+
+				wrapperObject = new JSONObject();
+				wrapperObject.put("QueryType", "batch_insert");
+				wrapperObject.put("Database", DB);
+				wrapperObject.put("Datatable", "ps_virtual_project_log");
+				wrapperObject.put("InsertObjects", InsertObjects);
+				agent.Query(wrapperObject);
+			}
+
+		} catch (Exception e) {
+			LogUtil.error("HBaseUtil.Insert_ps_virtual_project_log.Exception: " + e.toString());
+		}
+	}
 	// public static void InsertPointCommunicationLog(List<String> list) throws
 	// Exception {
 	// // TODO Auto-generated method stub