浏览代码

有问题,需退至1.2号版本。

Jay Lee 6 年之前
父节点
当前提交
63d3fdaca4

+ 3 - 0
collect/src/main/java/com/saga/main/Main.java

@@ -15,6 +15,7 @@ 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.thread.common.CheckThread;
+import com.saga.thread.common.ExcuteAckThread;
 import com.saga.thread.common.LoadConfigThread;
 import com.saga.thread.common.SaveThread;
 import com.saga.thread.common.StatsThread;
@@ -62,6 +63,7 @@ public class Main {
 			// TODO Auto-generated catch block
 			LogUtil.error("Main.threadStart.CollectThread: " + e.toString());
 		}
+		// new ExcuteAckThread().start();
 		// ÉÏ´«½ø³Ì
 		List<IServerManager> udpserverList = new ArrayList<IServerManager>();
 		List<IServerManager> tcpserverList = new ArrayList<IServerManager>();
@@ -86,6 +88,7 @@ public class Main {
 				LogUtil.error("Main.threadStart.SendThread-TCP: " + e.toString());
 			}
 		}
+
 		// Êý¾Ý´¦Àí½ø³Ì
 		try {
 			new SaveThread().start();

+ 31 - 0
collect/src/main/java/com/saga/test/TestClient.java

@@ -0,0 +1,31 @@
+package com.saga.test;
+
+import java.util.Date;
+
+import com.persagy.communication.entity.Packet;
+import com.persagy.communication.mina.udp.client.UDPClientManager;
+import com.persagy.communication.util.IClientManager;
+import com.saga.util.Constant;
+
+public class TestClient {
+
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+		IClientManager client = new UDPClientManager("192.168.20.159", 8089, "192.168.20.159", 8080, "utf-8", false,
+				1000);
+		while (true) {
+			StringBuffer sb = new StringBuffer();
+			sb.append("1234567890").append(";").append(1).append(";").append("report").append(";")
+					.append(Constant.format(new Date())).append(";").append(Constant.getID()).append(";").append("1001")
+					.append(";1;").append(1111).append(";").append(3.33).append(";");
+			client.AppendToSend(new Packet(sb.toString()));
+			try {
+				Thread.sleep(1L);
+			} catch (InterruptedException e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+		}
+	}
+
+}

+ 38 - 0
collect/src/main/java/com/saga/test/TestServer.java

@@ -0,0 +1,38 @@
+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;
+
+public class TestServer {
+
+	public static void main(String[] args) {
+		// TODO Auto-generated method stub
+		IServerManager server = new UDPServerManager("0.0.0.0", 8080, "utf-8", false, 10000);
+		server.Start();
+		while (true) {
+
+			PacketEntity p = server.PopRece();
+			if (p != null)
+
+			{
+				handlePacket2(server, p);
+			}
+			try {
+				Thread.sleep(2L);
+			} catch (InterruptedException e) {
+				// TODO Auto-generated catch block
+				e.printStackTrace();
+			}
+		}
+	}
+
+	private static void handlePacket2(IServerManager server, PacketEntity p) {
+		// TODO Auto-generated method stub
+		LogUtil.info(server.Name() + " PopRece:" + p.address + " " + p.content.packetString);
+
+	}
+}

+ 72 - 9
collect/src/main/java/com/saga/thread/common/CheckThread.java

@@ -11,6 +11,8 @@ import java.util.Map.Entry;
 import org.zillion.util.log.LogUtil;
 
 import com.saga.util.Constant;
+import com.saga.util.HBaseUtil;
+import com.zillion.database.agent.ZillionAgent;
 
 public class CheckThread extends Thread {
 	// ZillionAgent agent;
@@ -22,6 +24,11 @@ public class CheckThread extends Thread {
 	private Map<String, String> projectStatus = new HashMap<String, String>();
 	private Map<String, String> physicalStatus = new HashMap<String, String>();
 
+	private Map<String, String> dbPointStatus = new HashMap<String, String>();
+	private Map<String, String> dbMeterStatus = new HashMap<String, String>();
+	private Map<String, String> dbProjectStatus = new HashMap<String, String>();
+	private Map<String, String> dbPhysicalStatus = new HashMap<String, String>();
+
 	public CheckThread() {
 		try {
 			// this.agent = agent;
@@ -40,14 +47,18 @@ public class CheckThread extends Thread {
 		LogUtil.info("CheckThread-" + this.getName() + " Starting ...");
 		// 初次加载生成当前状态
 		this.excuteDBStatus();
+		// 获取DB当前状态
+		this.loadDBStatus();
 
 		while (true) {
 			try {
+
 				// // V1.0点位日志结构
 				// this.realCheck();
 				// V2.0点位日志结构
 				this.realCheck2();
 				// Constant.showStatus();
+				this.DBStatus();
 				Thread.sleep(Constant.Millisecond.SECOND_1);
 			} catch (Exception e) {
 				// TODO Auto-generated catch block
@@ -56,6 +67,50 @@ public class CheckThread extends Thread {
 		}
 	}
 
+	private void loadDBStatus() {
+		// TODO Auto-generated method stub
+		// 重新启动加载节点、点位状态
+		this.dbPointStatus = new HashMap<String, String>();
+		this.dbMeterStatus = new HashMap<String, String>();
+		this.dbProjectStatus = new HashMap<String, String>();
+		this.dbPhysicalStatus = new HashMap<String, String>();
+		for (ZillionAgent agent : Constant.agents) {
+			for (Entry<String, String> bd : Constant.BuildingDB.entrySet()) {
+				try {
+					String project = bd.getKey();
+					String db = bd.getValue();
+					this.dbPhysicalStatus
+							.putAll(HBaseUtil.ps_physical_status(agent, db, "ps_physical_status", project));
+					this.dbMeterStatus
+							.putAll(HBaseUtil.ps_virtual_meter_status(agent, db, "ps_virtual_meter_status", project));
+					this.dbPointStatus
+							.putAll(HBaseUtil.ps_virtual_point_status(agent, db, "ps_virtual_point_status", project));
+					this.dbProjectStatus.putAll(
+							HBaseUtil.ps_virtual_project_status(agent, db, "ps_virtual_project_status", project));
+
+				} catch (Exception e) {
+					// TODO Auto-generated catch block
+					e.printStackTrace();
+					LogUtil.error("CheckThread.loadDBStatus: " + e.toString());
+				}
+			}
+			return;
+		}
+	}
+
+	private void DBStatus() {
+		// TODO Auto-generated method stub
+
+		this.dbPhysicalStatus = this.physicalStatus;
+
+		this.dbMeterStatus = this.meterStatus;
+
+		this.dbPointStatus = this.pointStatus;
+
+		this.dbProjectStatus = this.projectStatus;
+
+	}
+
 	private void excuteDBStatus() {
 		// TODO Auto-generated method stub
 		// for (Entry<String, Long> pns : Constant.ps_NodeStatus.entrySet()) {
@@ -68,7 +123,7 @@ public class CheckThread extends Thread {
 		// }
 		for (Entry<String, Long> ppsu : Constant.ps_PointStatusUp.entrySet()) {
 			if (ppsu.getValue() != null) {
-				if (System.currentTimeMillis() - ppsu.getValue() < Constant.Millisecond.MINUTE_10) {
+				if (System.currentTimeMillis() - ppsu.getValue() < Constant.Millisecond.MINUTE_15) {
 					this.pointStatus.put(ppsu.getKey(), "conn");
 				} else {
 					this.pointStatus.put(ppsu.getKey(), "disconn");
@@ -81,7 +136,7 @@ public class CheckThread extends Thread {
 		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) {
+				if (System.currentTimeMillis() - pps.getValue() < Constant.Millisecond.MINUTE_15) {
 					this.physicalStatus.put(pps.getKey(), "conn");
 				} else {
 					this.physicalStatus.put(pps.getKey(), "disconn");
@@ -94,7 +149,7 @@ public class CheckThread extends Thread {
 		}
 		for (Entry<String, Long> pps : Constant.ps_ProjectStatus.entrySet()) {
 			if (pps.getValue() != null) {
-				if (System.currentTimeMillis() - pps.getValue() < Constant.Millisecond.MINUTE_10) {
+				if (System.currentTimeMillis() - pps.getValue() < Constant.Millisecond.MINUTE_15) {
 					this.projectStatus.put(pps.getKey(), "conn");
 				} else {
 					this.projectStatus.put(pps.getKey(), "disconn");
@@ -106,7 +161,7 @@ public class CheckThread extends Thread {
 		}
 		for (Entry<String, Long> pmsu : Constant.ps_MeterStatusUp.entrySet()) {
 			if (pmsu.getValue() != null) {
-				if (System.currentTimeMillis() - pmsu.getValue() < Constant.Millisecond.MINUTE_10) {
+				if (System.currentTimeMillis() - pmsu.getValue() < Constant.Millisecond.MINUTE_15) {
 					this.meterStatus.put(pmsu.getKey(), "conn");
 				} else {
 					this.meterStatus.put(pmsu.getKey(), "disconn");
@@ -156,7 +211,9 @@ public class CheckThread extends Thread {
 					// 首次
 					!this.projectStatus.containsKey(project) || null == this.projectStatus.get(project) ||
 					// 已有记录且状态发生变化
-							!status.equalsIgnoreCase(this.projectStatus.get(project))) {
+							!status.equalsIgnoreCase(this.projectStatus.get(project)) ||
+							// DB同步
+							!status.equalsIgnoreCase(this.dbProjectStatus.get(project))) {
 						Constant.Insert_agents_ps_virtual_project_log(project, Constant.format(new Date()), status);
 						if ("disconn".equalsIgnoreCase(status)) {
 							Constant.Insert_agents_ps_virtual_project_status(project, status,
@@ -192,7 +249,9 @@ public class CheckThread extends Thread {
 					// 首次
 					!this.meterStatus.containsKey(pm) || null == this.meterStatus.get(pm) ||
 					// 已有记录且状态发生变化
-							!status.equalsIgnoreCase(this.meterStatus.get(pm))) {
+							!status.equalsIgnoreCase(this.meterStatus.get(pm))
+							// DB同步
+							|| !status.equalsIgnoreCase(this.dbMeterStatus.get(pm))) {
 						Constant.Insert_agents_ps_virtual_meter_log(project, meter, "up", Constant.format(new Date()),
 								status);
 						if ("disconn".equalsIgnoreCase(status)) {
@@ -246,7 +305,9 @@ public class CheckThread extends Thread {
 					// 首次
 					!this.physicalStatus.containsKey(physical) || null == this.physicalStatus.get(physical) ||
 					// 已有记录且状态发生变化
-							!status.equalsIgnoreCase(this.physicalStatus.get(physical))) {
+							!status.equalsIgnoreCase(this.physicalStatus.get(physical)) ||
+							// DB同步
+							!status.equalsIgnoreCase(this.dbPhysicalStatus.get(physical))) {
 						Constant.Insert_agents_ps_physical_log(project, address_1, address_2, address_3, address_4,
 								Constant.format(new Date()), status);
 
@@ -465,7 +526,9 @@ public class CheckThread extends Thread {
 					}
 				} else if (
 				// 已有记录且状态发生变化
-				(!status.equalsIgnoreCase(this.pointStatus.get(pointUp.getKey())))) {
+				(!status.equalsIgnoreCase(this.pointStatus.get(pointUp.getKey()))) ||
+				// DB同步
+						(!status.equalsIgnoreCase(this.dbPointStatus.get(pointUp.getKey())))) {
 					// 添加Log记录
 					if (Constant.RecordAddress.containsKey(building + "-" + point.replace(".", "-"))) {
 						if (Constant.RecordAddress.get(building + "-" + point.replace(".", "-")) != null) {
@@ -662,7 +725,7 @@ public class CheckThread extends Thread {
 	// String status = "";
 	//
 	// if (pointUp.getValue() != null
-	// && this.realCheck - pointUp.getValue() < Constant.Millisecond.MINUTE_10)
+	// && this.realCheck - pointUp.getValue() < Constant.Millisecond.MINUTE_15)
 	// {
 	// HBaseUtil.InsertPointStatus(Constant.agent,
 	// Constant.BuildingDB.get(building), building,

+ 49 - 0
collect/src/main/java/com/saga/thread/common/ExcuteAckThread.java

@@ -0,0 +1,49 @@
+package com.saga.thread.common;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+
+import org.zillion.util.log.LogUtil;
+
+import com.saga.util.Constant;
+
+public class ExcuteAckThread extends Thread {
+	public void run() {
+		LogUtil.info("ExcuteAckThread-" + this.getName() + " Starting ...");
+
+		while (true) {
+			try {
+				excuteAckList();
+				Thread.sleep(10L);
+
+			} catch (Exception e) {
+				// TODO Auto-generated catch block
+				LogUtil.error("ExcuteAckThread.run.Exception: " + e.toString());
+			}
+		}
+	}
+
+	private void excuteAckList() {
+		synchronized (Constant.ackList) {
+			for (Entry<String, List<String>> ack : Constant.ackList.entrySet()) {
+				StringBuffer cmd = new StringBuffer();
+				for (int i = 0; i < ack.getValue().size(); i++) {
+					cmd.append(ack.getValue().get(i));
+					if ((i > 0) && (i % 50 == 0 || i == ack.getValue().size() - 1)) {
+						synchronized (Constant.ackListResult) {
+							if (!Constant.ackListResult.containsKey(ack.getKey())) {
+								Constant.ackListResult.put(ack.getKey(), new ArrayList<String>());
+							}
+							Constant.ackListResult.get(ack.getKey()).add(cmd.toString());
+						}
+						cmd = new StringBuffer();
+					} else {
+						cmd.append("&");
+					}
+				}
+			}
+			Constant.ackList.clear();
+		}
+	}
+}

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

@@ -130,7 +130,7 @@ public class SaveThread extends Thread {
 
 					// recordList = new ArrayList<Record>();
 					if (this.strecordList.size() > 0) {
-						LogUtil.warn("SaveThread.recordList " + this.strecordList.size());
+						LogUtil.info("SaveThread.recordList " + this.strecordList.size());
 						this.strecordList.clear();
 						System.gc();
 					}
@@ -232,10 +232,10 @@ public class SaveThread extends Thread {
 			}
 
 			if (this.streportList.size() > 0) {
-				LogUtil.warn("SaveThread.reportList " + streportList.size());
+				LogUtil.info("SaveThread.reportList " + streportList.size());
 				this.streportList.clear();
 				System.gc();
-				LogUtil.warn("SaveThread.reportList " + streportList.size());
+				LogUtil.info("SaveThread.reportList " + streportList.size());
 			}
 		} catch (Exception e) {
 			// TODO Auto-generated catch block

+ 32 - 60
collect/src/main/java/com/saga/thread/down/CollectThread.java

@@ -41,24 +41,24 @@ public class CollectThread extends Thread {
 			// }
 			this.port = port;
 			if (Constant.PortCompress == null || !Constant.PortCompress.containsKey(this.port)) {
-				this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", false, 1000));
+				this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", false, 10000));
 				this.serverList.put("TCP",
-						new TCPServerManager("0.0.0.0", this.port, "utf-8", false, 1000, (byte) '(', (byte) ')'));
+						new TCPServerManager("0.0.0.0", this.port, "utf-8", false, 10000, (byte) '(', (byte) ')'));
 			} else {
 				if ("true".equalsIgnoreCase(Constant.PortCompress.get(this.port))) {
 					// this.serverList.put("UDP", new
 					// UDPServerManager("0.0.0.0",
 					// this.port, "utf-8", 1000));
-					this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", true, 1000));
+					this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", true, 10000));
 					this.serverList.put("TCP",
-							new TCPServerManager("0.0.0.0", this.port, "utf-8", true, 1000, (byte) '(', (byte) ')'));
+							new TCPServerManager("0.0.0.0", this.port, "utf-8", true, 10000, (byte) '(', (byte) ')'));
 				} else {
 					// this.serverList.put("UDP", new
 					// UDPServerManager("0.0.0.0",
 					// this.port, "utf-8", 1000));
-					this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", false, 1000));
+					this.serverList.put("UDP", new UDPServerManager("0.0.0.0", this.port, "utf-8", false, 10000));
 					this.serverList.put("TCP",
-							new TCPServerManager("0.0.0.0", this.port, "utf-8", false, 1000, (byte) '(', (byte) ')'));
+							new TCPServerManager("0.0.0.0", this.port, "utf-8", false, 10000, (byte) '(', (byte) ')'));
 				}
 			}
 
@@ -97,12 +97,15 @@ public class CollectThread extends Thread {
 					{
 						this.handlePacket2(server.getValue(), p);
 						this.handlePointSet2();
+						// this.ack(server.getValue());
 
 					}
-
-					Thread.sleep(1L);
 				}
-				this.ack();
+				Thread.sleep(1L);
+				// synchronized (Constant.ackListResult) {
+				// Constant.ackListResult.clear();
+				// }
+
 				// Thread.sleep(3L);
 			} catch (Exception e) {
 				// TODO Auto-generated catch block
@@ -111,46 +114,19 @@ public class CollectThread extends Thread {
 		}
 	}
 
-	private void ack() {
+	private void ack(IServerManager server) {
 		// TODO Auto-generated method stub
-		Map<String, List<String>> acks = this.excuteAckList();
-
-		for (Entry<String, List<String>> ack : acks.entrySet()) {
-			for (String cmd : ack.getValue()) {
-				for (Entry<String, IServerManager> server : this.serverList.entrySet()) {
+		synchronized (Constant.ackListResult) {
+			for (Entry<String, List<String>> ack : Constant.ackListResult.entrySet()) {
+				for (String cmd : ack.getValue()) {
 					try {
-						server.getValue().AppendToSend(ack.getKey(), new Packet(cmd));
+						server.AppendToSend(ack.getKey(), new Packet(cmd));
 					} catch (Exception e) {
-						e.printStackTrace();
-					}
-				}
-			}
-		}
-	}
-
-	private Map<String, List<String>> excuteAckList() {
-		Map<String, List<String>> ackListResult = new HashMap<String, List<String>>();
-		synchronized (Constant.ackList) {
-			for (Entry<String, List<String>> ack : Constant.ackList.entrySet()) {
-				StringBuffer cmd = new StringBuffer();
-				for (int i = 0; i < ack.getValue().size(); i++) {
-					cmd.append(ack.getValue().get(i));
-					if ((i > 0) && (i % 50 == 0 || i == ack.getValue().size() - 1)) {
-						if (!ackListResult.containsKey(ack.getKey())) {
-							ackListResult.put(ack.getKey(), new ArrayList<String>());
-						}
-						ackListResult.get(ack.getKey()).add(cmd.toString());
-						cmd = new StringBuffer();
-					} else {
-						cmd.append("&");
+						// e.printStackTrace();
 					}
 				}
 			}
-			Constant.ackList.clear();
 		}
-
-		return ackListResult;
-
 	}
 
 	private void handlePointSet2() {
@@ -357,15 +333,13 @@ public class CollectThread extends Thread {
 						// LogUtil.info("handlePacket updateMeterCollector
 						// end...");
 						// if (this.needAck(server)) {
-						// if (sb.length() > 0) {
-						// sb.append("&" + rList[0] + ";" + rList[1] + ";" +
-						// "reportack" + ";"
-						// + Constant.format(new Date()) + ";" + rList[4]);
-						// } else {
-						// sb.append(rList[0] + ";" + rList[1] + ";" +
-						// "reportack" + ";" + Constant.format(new Date())
-						// + ";" + rList[4]);
-						// }
+						if (sb.length() > 0) {
+							sb.append("&" + rList[0] + ";" + rList[1] + ";" + "reportack" + ";"
+									+ Constant.format(new Date()) + ";" + rList[4]);
+						} else {
+							sb.append(rList[0] + ";" + rList[1] + ";" + "reportack" + ";" + Constant.format(new Date())
+									+ ";" + rList[4]);
+						}
 						// }
 					} else if ("reportaddtion".equalsIgnoreCase(rList[2]) && rList.length > 7) {
 						Constant.pointMAC.put(rList[0] + "-" + rList[5], rList[1]);
@@ -413,15 +387,13 @@ public class CollectThread extends Thread {
 						// end...");
 						// LogUtil.info("handlePacket Ack begin...");
 						// if (this.needAck(server)) {
-						// if (sb.length() > 0) {
-						// sb.append("&" + rList[0] + ";" + rList[1] + ";" +
-						// "reportaddtionack" + ";"
-						// + Constant.format(new Date()) + ";" + rList[4]);
-						// } else {
-						// sb.append(rList[0] + ";" + rList[1] + ";" +
-						// "reportaddtionack" + ";"
-						// + Constant.format(new Date()) + ";" + rList[4]);
-						// }
+						if (sb.length() > 0) {
+							sb.append("&" + rList[0] + ";" + rList[1] + ";" + "reportaddtionack" + ";"
+									+ Constant.format(new Date()) + ";" + rList[4]);
+						} else {
+							sb.append(rList[0] + ";" + rList[1] + ";" + "reportaddtionack" + ";"
+									+ Constant.format(new Date()) + ";" + rList[4]);
+						}
 						// }
 						// LogUtil.info("handlePacket Ack end...");
 					} else if ("senddownsetack".equalsIgnoreCase(rList[2])) {

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

@@ -34,7 +34,7 @@ public class Constant {
 	public static int InsertThread = 1;
 	// public static ZillionAgent agent;
 	public static List<ZillionAgent> agents = new ArrayList<ZillionAgent>();
-	public static boolean DBConn = true;
+	// public static boolean DBConn = true;
 	public static String WaitingTimeControl = "5Second";
 	public static File setlog = new File(System.getProperty("user.dir") + "/setlog.log");
 	public static Map<String, List<String>> CollectReportList = new HashMap<String, List<String>>();
@@ -87,6 +87,7 @@ public class Constant {
 	public static int report2DBCount = 0;
 	public static Map<String, String> pointMAC = new HashMap<String, String>();
 	public static Map<String, List<String>> ackList = new HashMap<String, List<String>>();
+	public static Map<String, List<String>> ackListResult = new HashMap<String, List<String>>();
 
 	static {
 		String resource = "\\config.json";

+ 141 - 1
collect/src/main/java/com/saga/util/HBaseUtil.java

@@ -1934,7 +1934,7 @@ public class HBaseUtil {
 
 		} catch (Exception e) {
 			LogUtil.error(agent.getFull_name() + " HBaseUtil.BatchInsert.Exception: " + e.toString());
-			Constant.DBConn = false;
+			// Constant.DBConn = false;
 		}
 	}
 
@@ -2080,4 +2080,144 @@ public class HBaseUtil {
 		}
 	}
 
+	public static Map<String, String> ps_physical_status(ZillionAgent agent, String db, String tb, String project) 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 详情
+		Map<String, String> dbPhysicalStatus = new HashMap<String, 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");
+		for (int i = 0; i < resultContent.size(); i++) {
+			JSONObject dataItem = (JSONObject) resultContent.get(i);
+			String address_1 = (String) dataItem.get("address_1");
+			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 status = (String) dataItem.get("status");
+			dbPhysicalStatus.put(key, status);
+
+		}
+		return dbPhysicalStatus;
+	}
+
+	public static Map<String, String> 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 详情
+		Map<String, String> dbPointStatus = new HashMap<String, 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");
+
+		for (int i = 0; i < resultContent.size(); i++) {
+			JSONObject dataItem = (JSONObject) resultContent.get(i);
+			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 status = (String) dataItem.get("status");
+				dbPointStatus.put(key, status);
+			}
+		}
+		return dbPointStatus;
+	}
+
+	public static Map<String, String> 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 详情
+
+		Map<String, String> dbMeterStatus = new HashMap<String, 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");
+
+		for (int i = 0; i < resultContent.size(); i++) {
+			JSONObject dataItem = (JSONObject) resultContent.get(i);
+			String meter = (String) dataItem.get("meter");
+			String updown = (String) dataItem.get("updown");
+			if ("up".equalsIgnoreCase(updown)) {
+				String key = project + "-" + meter;
+				String status = (String) dataItem.get("status");
+				dbMeterStatus.put(key, status);
+
+			}
+		}
+		return dbMeterStatus;
+	}
+
+	public static Map<String, String> 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 详情
+
+		Map<String, String> dbProjectStatus = new HashMap<String, 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");
+
+		for (int i = 0; i < resultContent.size(); i++) {
+			JSONObject dataItem = (JSONObject) resultContent.get(i);
+
+			String key = project;
+			String status = (String) dataItem.get("status");
+			dbProjectStatus.put(key, status);
+
+		}
+		return dbProjectStatus;
+	}
 }