Просмотр исходного кода

修复report写库失败正常返回ack的bug。

Jay Lee 6 лет назад
Родитель
Сommit
24c1d40667

+ 2 - 0
collect/src/main/java/com/saga/entity/Report.java

@@ -4,5 +4,7 @@ public class Report {
 	public String UploadName;
 	public int ReportID;
 	public long resend;
+	public String ReportType;
+	public String mac;
 	public Record record;
 }

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

@@ -613,9 +613,11 @@ public class Main {
 
 			List<String> qrlist = new ArrayList<String>();
 			// qrlist.add("zookeeper1:2181,zookeeper2:2181,zookeeper3:2181");
+
 			qrlist.add("zookeeper1,zookeeper2,zookeeper3");
 			qrlist.add("zookeeper1.saga");
-			// qrlist.add("zookeeper1");
+
+			// qrlist.add("zookeeper1.test");
 
 			for (String qr : qrlist) {
 				try {

+ 153 - 9
collect/src/main/java/com/saga/test/Test.java

@@ -35,21 +35,165 @@ public class Test {
 	public static void main(String[] args) throws InterruptedException {
 		// TODO Auto-generated method stub
 
+		Constant.ackList.put("192.168.20.159:8081", new ArrayList<String>());
+		Constant.ackList.put("192.168.20.159:8082", new ArrayList<String>());
+		Constant.ackList.get("192.168.20.159:8081").add("ack1;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack2;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack3;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack4;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack5;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack6;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack7;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack8;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack9;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack10;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack11;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack12;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack13;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack14;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack15;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack16;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack17;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack18;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack19;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack20;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack21;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack22;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack23;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack24;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack25;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack26;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack27;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack28;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack29;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack30;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack31;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack32;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack33;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack34;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack35;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack36;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack37;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack38;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack39;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack40;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack41;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack42;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack43;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack44;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack45;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack46;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack47;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack48;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack49;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack50;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack51;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack52;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack53;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack54;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack55;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack56;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack57;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack58;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack59;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack60;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack61;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack62;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack63;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack64;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack65;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack66;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack67;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack68;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack69;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack70;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack71;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack72;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack73;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack74;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack75;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack76;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack77;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack78;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack79;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack80;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack81;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack82;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack83;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack84;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack85;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack86;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack87;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack88;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack89;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack90;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack91;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack92;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack93;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack94;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack95;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack96;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack97;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack98;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack99;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack100;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack101;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack102;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack103;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack104;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack105;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack106;");
+		Constant.ackList.get("192.168.20.159:8081").add("ack107;");
+
+		for (Entry<String, List<String>> a : excuteAckList().entrySet()) {
+			for (String v : a.getValue()) {
+				System.out.println(a.getKey() + "");
+				System.out.println(v + "");
+			}
+
+		}
+
 		// ³õʼ»¯
 		// init(args);
 		// ÒµÎñ½ø³ÌÆô¶¯
 		// threadStart();
-		agentsStart();
-		while (true) {
-			Constant.AgentsLoadDB2Cache_dy_pointlist();
-			for (Entry<String, List<String>> p : Constant.ps_ProjectList.entrySet()) {
-				LogUtil.info(p.getKey() + " " + p.getValue());
-			}
-			for (Entry<String, List<String>> m : Constant.ps_MeterList.entrySet()) {
-				LogUtil.info(m.getKey() + " " + m.getValue());
+		// agentsStart();
+		// while (true) {
+		// Constant.AgentsLoadDB2Cache_dy_pointlist();
+		// for (Entry<String, List<String>> p :
+		// Constant.ps_ProjectList.entrySet()) {
+		// LogUtil.info(p.getKey() + " " + p.getValue());
+		// }
+		// for (Entry<String, List<String>> m :
+		// Constant.ps_MeterList.entrySet()) {
+		// LogUtil.info(m.getKey() + " " + m.getValue());
+		// }
+		// Thread.sleep(Constant.Millisecond.MINUTE_1);
+		// }
+	}
+
+	private static 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("&");
+					}
+				}
 			}
-			Thread.sleep(Constant.Millisecond.MINUTE_1);
 		}
+		Constant.ackList.clear();
+		return ackListResult;
+
 	}
 
 	private static void agentsStart() {

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

@@ -2,19 +2,24 @@ package com.saga.thread.common;
 
 import java.util.ArrayList;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 
 import org.zillion.util.log.LogUtil;
 
 import com.saga.entity.Record;
+import com.saga.entity.Report;
 import com.saga.util.Constant;
 
 public class SaveThread extends Thread {
 	// private List<String> pointCommunicationList = new ArrayList<String>();
 	// private String ip = null;
-	private List<String> streportList = new ArrayList<String>();
-	private List<Record> strecordList = new ArrayList<Record>();
+
+	private Map<String, List<String>> streportList = new HashMap<String, List<String>>();
+	private List<Report> strecordList = new ArrayList<Report>();
 
 	public SaveThread() {
 		try {
@@ -68,7 +73,15 @@ public class SaveThread extends Thread {
 			synchronized (Constant.CollectReportList) {
 				if (Constant.CollectReportList.size() > 0) {
 					LogUtil.info("Constant.CollectReportList:" + Constant.CollectReportList.size());
-					this.streportList.addAll(Constant.CollectReportList);
+					for (Entry<String, List<String>> cr : Constant.CollectReportList.entrySet()) {
+						if (!this.streportList.containsKey(cr.getKey())) {
+							this.streportList.put(cr.getKey(), new ArrayList<String>());
+						}
+						this.streportList.get(cr.getKey()).addAll(cr.getValue());
+						Constant.collectReportCount += cr.getValue().size();
+						LogUtil.info("ReportList." + cr.getKey() + ":" + cr.getValue().size());
+					}
+
 					Constant.CollectReportList.clear();
 					// Constant.CollectReportList = new ArrayList<String>();
 					LogUtil.info("Constant.CollectReportList:" + Constant.CollectReportList.size());
@@ -82,128 +95,142 @@ public class SaveThread extends Thread {
 					return;
 				}
 			}
-			Constant.collectReportCount += this.streportList.size();
-			LogUtil.info("ReportList:" + this.streportList.size());
-			int seq = 0;
 
-			while (this.streportList.size() > 0) {
-				seq++;
-				LogUtil.info("ReportList:" + this.streportList.size() + " seq:" + seq);
-				int no = 5000 > this.streportList.size() ? this.streportList.size() : 5000;
-				if (no == 5000 || seq > 1) {
-					LogUtil.info("ReportList:" + this.streportList.size() + " seq:" + seq);
-				}
-				String b = this.streportList.get(0).replace("(", "").replace(")", "").replace("", "");
-				if (b == null || b.length() == 0) {
-					this.streportList.remove(0);
-					return;
-				}
-				b = this.streportList.get(0).replace("(", "").replace(")", "").replace(" ", "").split(";")[0];
+			int seq = 0;
+			Map<String, String> recvlist = new HashMap<String, String>();
+			for (Entry<String, List<String>> rl : this.streportList.entrySet()) {
+				recvlist.put(rl.getKey(), rl.getKey());
+			}
+			for (Entry<String, String> r : recvlist.entrySet()) {
+				while (this.streportList.get(r.getKey()).size() > 0) {
+					seq++;
+					LogUtil.info("ReportList:" + this.streportList.get(r.getKey()).size() + " seq:" + seq);
+					int no = 5000 > this.streportList.get(r.getKey()).size() ? this.streportList.get(r.getKey()).size()
+							: 5000;
+					if (no == 5000 || seq > 1) {
+						LogUtil.info("ReportList:" + this.streportList.get(r.getKey()).size() + " seq:" + seq);
+					}
+					String b = this.streportList.get(r.getKey()).get(0).replace("(", "").replace(")", "").replace("",
+							"");
+					if (b == null || b.length() == 0) {
+						this.streportList.get(r.getKey()).remove(0);
+						return;
+					}
+					b = this.streportList.get(r.getKey()).get(0).replace("(", "").replace(")", "").replace(" ", "")
+							.split(";")[0];
 
-				while (Constant.RecordListToDB.size() > 0) {
-					if (Constant.RecordListToDB.size() == 0) {
-						break;
-					} else {
-						Thread.sleep(3L);
+					while (Constant.RecordListToDB.size() > 0) {
+						if (Constant.RecordListToDB.size() == 0) {
+							break;
+						} else {
+							Thread.sleep(3L);
+						}
+						Thread.sleep(1L);
 					}
-					Thread.sleep(1L);
-				}
 
-				// recordList = new ArrayList<Record>();
-				if (this.strecordList.size() > 0) {
-					LogUtil.warn("SaveThread.recordList " + this.strecordList.size());
-					this.strecordList.clear();
-					System.gc();
-				}
-				Iterator<String> it = this.streportList.iterator();
-				int c = 0;
-				synchronized (Constant.ps_PointStatusUp) {
-					while (it.hasNext()) {
-						try {
-							String report = it.next().replace("(", "").replace(")", "").replace(" ", "");
-							if (c < no) {
-								if (report != null) {
-									String[] rList = report.replace("(", "").replace(")", "").replace(" ", "")
-											.split(";");
-									if (rList != null && rList.length > 8) {
-										String building = rList[0];
-										if (!building.equalsIgnoreCase(b)) {
-											continue;
-										}
-										//String mac = rList[1];
-										String function = rList[2];
-										Date time = Constant.parse(rList[3]);
-										String meter = rList[5];
-										int count = Integer.valueOf(rList[6]);
-										// String id = rList[4];
+					// recordList = new ArrayList<Record>();
+					if (this.strecordList.size() > 0) {
+						LogUtil.warn("SaveThread.recordList " + this.strecordList.size());
+						this.strecordList.clear();
+						System.gc();
+					}
+					Iterator<String> it = this.streportList.get(r.getKey()).iterator();
+					int c = 0;
+					synchronized (Constant.ps_PointStatusUp) {
+						while (it.hasNext()) {
+							try {
+								String report = it.next().replace("(", "").replace(")", "").replace(" ", "");
+								if (c < no) {
+									if (report != null) {
+										String[] rList = report.replace("(", "").replace(")", "").replace(" ", "")
+												.split(";");
+										if (rList != null && rList.length > 8) {
+											String building = rList[0];
+											if (!building.equalsIgnoreCase(b)) {
+												continue;
+											}
+											String mac = rList[1];
+											String function = rList[2];
+											Date time = Constant.parse(rList[3]);
+											String meter = rList[5];
+											int count = Integer.valueOf(rList[6]);
+											int id = Integer.valueOf(rList[4]);
 
-										for (int i = 0; i < count; i++) {
-											Record record = new Record();
-											if ("report".equalsIgnoreCase(function)) {
-												record.funcID = Integer.valueOf(rList[7 + 2 * i]);
-												record.data = Double.valueOf(rList[8 + 2 * i]);
-												// synchronized
-												// (Constant.pointListUp) {
-												Constant.ps_PointStatusUp.put(
-														building + "-" + meter + "." + record.funcID,
-														System.currentTimeMillis());
-												// }
+											for (int i = 0; i < count; i++) {
+												Report report2 = new Report();
+												Record record = new Record();
+												if ("report".equalsIgnoreCase(function)) {
+													record.funcID = Integer.valueOf(rList[7 + 2 * i]);
+													record.data = Double.valueOf(rList[8 + 2 * i]);
+													// synchronized
+													// (Constant.pointListUp) {
+													Constant.ps_PointStatusUp.put(
+															building + "-" + meter + "." + record.funcID,
+															System.currentTimeMillis());
+													// }
 
-											} else if ("reportaddtion".equalsIgnoreCase(function)) {
-												record.funcID = Integer.valueOf(rList[7 + 3 * i]);
-												record.data = Double.valueOf(rList[8 + 3 * i]);
-												record.addtion = "null".equalsIgnoreCase(rList[9 + i * 3]) ? ""
-														: (rList[9 + i * 3]);
-												// synchronized
-												// (Constant.pointListUp) {
-												Constant.ps_PointStatusUp.put(
-														building + "-" + meter + "." + record.funcID,
-														System.currentTimeMillis());
+												} else if ("reportaddtion".equalsIgnoreCase(function)) {
+													record.funcID = Integer.valueOf(rList[7 + 3 * i]);
+													record.data = Double.valueOf(rList[8 + 3 * i]);
+													record.addtion = "null".equalsIgnoreCase(rList[9 + i * 3]) ? ""
+															: (rList[9 + i * 3]);
+													// synchronized
+													// (Constant.pointListUp) {
+													Constant.ps_PointStatusUp.put(
+															building + "-" + meter + "." + record.funcID,
+															System.currentTimeMillis());
 
-												// }
-											}
+													// }
+												}
 
-											record.buildingSign = building;
-											record.meterSign = meter;
-											record.receivetime = time;
+												record.buildingSign = building;
+												record.meterSign = meter;
+												record.receivetime = time;
+												report2.record = record;
 
-											this.strecordList.add(record);
+												report2.ReportID = id;
+												report2.mac = mac;
+												report2.UploadName = r.getKey();
+												report2.ReportType = function;
+												this.strecordList.add(report2);
 
+											}
+											it.remove();
+											// LogUtil.info("HBase.BatchInsert "
+											// +
+											// report);
+										} else {
+											it.remove();
 										}
-										it.remove();
-										// LogUtil.info("HBase.BatchInsert "
-										// +
-										// report);
+
 									} else {
 										it.remove();
 									}
-
 								} else {
-									it.remove();
+									break;
 								}
-							} else {
-								break;
+								c++;
+								// Thread.sleep(1L);
+							} catch (Exception e) {
+								// TODO Auto-generated catch block
+								it.remove();
+								LogUtil.error("SaveThread.run.Exception: " + e.toString());
 							}
-							c++;
-							// Thread.sleep(1L);
-						} catch (Exception e) {
-							// TODO Auto-generated catch block
-							it.remove();
-							LogUtil.error("SaveThread.run.Exception: " + e.toString());
 						}
 					}
-				}
 
-				Constant.RecordListToDB.addAll(this.strecordList);
-				Constant.report2DBCount += this.strecordList.size();
-				LogUtil.info("Constant.RecordListToDB add " + this.strecordList.size());
-				// recordList = new ArrayList<Record>();
-				this.strecordList.clear();
+					Constant.RecordListToDB.addAll(this.strecordList);
+					Constant.report2DBCount += this.strecordList.size();
+					LogUtil.info("Constant.RecordListToDB add " + this.strecordList.size());
+					// recordList = new ArrayList<Record>();
+					this.strecordList.clear();
 
-				Thread.sleep(1L);
+					Thread.sleep(1L);
 
-				// Thread.sleep(Constant.Millisecond.SECOND_1);
+					// Thread.sleep(Constant.Millisecond.SECOND_1);
+				}
 			}
+
 			if (this.streportList.size() > 0) {
 				LogUtil.warn("SaveThread.reportList " + streportList.size());
 				this.streportList.clear();

+ 22 - 4
collect/src/main/java/com/saga/thread/common/WTDBThread.java

@@ -11,13 +11,14 @@ import java.util.List;
 import org.zillion.util.log.LogUtil;
 
 import com.saga.entity.Record;
+import com.saga.entity.Report;
 import com.saga.util.Constant;
 
 public class WTDBThread extends Thread {
 	// ZillionAgent agent;
 	long realCheck = 0;
 	// private String ip = null;
-	private List<Record> wtdbrecordList = new ArrayList<Record>();
+	private List<Report> wtdbrecordList = new ArrayList<Report>();
 	// private Map<String, String> nodeStatus = new HashMap<String, String>();
 	// private Map<String, String> pointStatus = new HashMap<String, String>();
 
@@ -54,7 +55,7 @@ public class WTDBThread extends Thread {
 						// + Constant.RecordListToDB.size());
 						wtdbrecordList.addAll(Constant.RecordListToDB);
 						LogUtil.info("Constant.RecordListToDB remove " + wtdbrecordList.size());
-						Constant.RecordListToDB = new ArrayList<Record>();
+						Constant.RecordListToDB = new ArrayList<Report>();
 
 						// LogUtil.info(
 						// this.getName() + " Constant.RecordListToDB.size(): "
@@ -62,18 +63,19 @@ public class WTDBThread extends Thread {
 					}
 				}
 				if (wtdbrecordList.size() > 0) {
-					String b = wtdbrecordList.get(0).buildingSign;
+					String b = wtdbrecordList.get(0).record.buildingSign;
 					if (Constant.BuildingConv.containsKey(b)) {
 						b = Constant.BuildingConv.get(b);
 					}
 					// HBaseUtil.BatchInsert(Constant.agent,
 					// Constant.BuildingDB.get(b), wtdbrecordList);
 					Constant.AgentsBatchInsert(Constant.BuildingDB.get(b), wtdbrecordList);
+					this.addAck(wtdbrecordList);
 					// LogUtil.info(this.getName() + " recordList size: " +
 					// recordList.size());
 					// LogUtil.info(this.getName() + " HBaseUtil.BatchInsert
 					// size: " + recordList.size());
-					wtdbrecordList = new ArrayList<Record>();
+					wtdbrecordList = new ArrayList<Report>();
 					// LogUtil.info(this.getName() + " recordList size: " +
 					// recordList.size());
 				}
@@ -86,6 +88,22 @@ public class WTDBThread extends Thread {
 		}
 	}
 
+	private void addAck(List<Report> wtdbrecordList2) {
+		// TODO Auto-generated method stub
+		synchronized (Constant.ackList) {
+			for (Report r : wtdbrecordList2) {
+				if (!Constant.ackList.containsKey(r.UploadName)) {
+					Constant.ackList.put(r.UploadName, new ArrayList<String>());
+				}
+				// 1101070037;1;reportack;;123
+				StringBuffer sb = new StringBuffer();
+				sb.append(r.record.buildingSign).append(";").append(r.mac).append(";").append(r.ReportType)
+						.append("ack").append(";;").append(r.ReportID);
+				Constant.ackList.get(r.UploadName).add(sb.toString());
+			}
+		}
+	}
+
 	// private void insertCollectorCommunicationLog(String building, String mac)
 	// {
 	// // TODO Auto-generated method stub

+ 79 - 18
collect/src/main/java/com/saga/thread/down/CollectThread.java

@@ -1,5 +1,6 @@
 package com.saga.thread.down;
 
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
@@ -101,6 +102,7 @@ public class CollectThread extends Thread {
 
 					Thread.sleep(1L);
 				}
+				this.ack();
 				// Thread.sleep(3L);
 			} catch (Exception e) {
 				// TODO Auto-generated catch block
@@ -109,6 +111,48 @@ public class CollectThread extends Thread {
 		}
 	}
 
+	private void ack() {
+		// 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()) {
+					try {
+						server.getValue().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("&");
+					}
+				}
+			}
+			Constant.ackList.clear();
+		}
+
+		return ackListResult;
+
+	}
+
 	private void handlePointSet2() {
 		// TODO Auto-generated method stub
 		try {
@@ -282,7 +326,8 @@ public class CollectThread extends Thread {
 							if (Constant.BuildingAdds.containsKey(rList[0])) {
 								for (String building : Constant.BuildingAdds.get(rList[0])) {
 									String newData = s.replace(rList[0], building);
-									Constant.CollectReportList.add(newData);
+									this.addReport(p.address, newData);
+									// Constant.CollectReportList.add(newData);
 									// LogUtil.info("Constant.CollectReportList
 									// add:" + newData);
 									// String key = building + "-" + rList[1] +
@@ -291,7 +336,8 @@ public class CollectThread extends Thread {
 									// System.currentTimeMillis());
 								}
 							} else {
-								Constant.CollectReportList.add(s);
+								this.addReport(p.address, s);
+								// Constant.CollectReportList.add(s);
 								// LogUtil.info("rList[0]:" + rList[0]);
 								// String key = rList[0] + "-" + rList[1] + "-"
 								// + rList[4];
@@ -311,13 +357,15 @@ 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]);
@@ -332,7 +380,8 @@ public class CollectThread extends Thread {
 							if (Constant.BuildingAdds.containsKey(rList[0])) {
 								for (String building : Constant.BuildingAdds.get(rList[0])) {
 									String newData = s.replace(rList[0], building);
-									Constant.CollectReportList.add(newData);
+									this.addReport(p.address, newData);
+									// Constant.CollectReportList.add(newData);
 									// LogUtil.info("Constant.CollectReportList
 									// add:" + newData);
 									// String key = building + "-" + rList[1] +
@@ -341,7 +390,8 @@ public class CollectThread extends Thread {
 									// System.currentTimeMillis());
 								}
 							} else {
-								Constant.CollectReportList.add(s);
+								this.addReport(p.address, s);
+								// Constant.CollectReportList.add(s);
 								// LogUtil.info("rList[0]:" + rList[0]);
 								// String key = rList[0] + "-" + rList[1] + "-"
 								// + rList[4];
@@ -363,13 +413,15 @@ 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])) {
@@ -567,6 +619,15 @@ public class CollectThread extends Thread {
 		}
 	}
 
+	private void addReport(String address, String data) {
+		// TODO Auto-generated method stub
+
+		if (!Constant.CollectReportList.containsKey(address)) {
+			Constant.CollectReportList.put(address, new ArrayList<String>());
+		}
+		Constant.CollectReportList.get(address).add(data);
+	}
+
 	private void getPointList(String address, String[] rList) {
 		// TODO Auto-generated method stub
 		String key = rList[0] + "-" + rList[1] + "-" + address;

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

@@ -26,6 +26,7 @@ import com.google.gson.JsonParser;
 import com.saga.entity.Data;
 import com.saga.entity.PointSet;
 import com.saga.entity.Record;
+import com.saga.entity.Report;
 import com.zillion.database.agent.ZillionAgent;
 
 public class Constant {
@@ -36,9 +37,9 @@ public class Constant {
 	public static boolean DBConn = true;
 	public static String WaitingTimeControl = "5Second";
 	public static File setlog = new File(System.getProperty("user.dir") + "/setlog.log");
-	public static List<String> CollectReportList = new ArrayList<String>();
+	public static Map<String, List<String>> CollectReportList = new HashMap<String, List<String>>();
 
-	public static List<Record> RecordListToDB = new ArrayList<Record>();
+	public static List<Report> RecordListToDB = new ArrayList<Report>();
 	public static Map<String, String> Meter_Collector = new HashMap<String, String>();
 	public static Map<String, String> Meter_Collector_Control = new HashMap<String, String>();
 	public static List<PointSet> pointSetList = new ArrayList<PointSet>();
@@ -85,6 +86,7 @@ public class Constant {
 	public static int collectReportCount = 0;
 	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>>();
 
 	static {
 		String resource = "\\config.json";
@@ -574,7 +576,7 @@ public class Constant {
 		}
 	}
 
-	public static void AgentsBatchInsert(String db, List<Record> wtdbrecordList) {
+	public static void AgentsBatchInsert(String db, List<Report> wtdbrecordList) {
 		// TODO Auto-generated method stub
 		for (ZillionAgent agent : Constant.agents) {
 			try {

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

@@ -11,6 +11,7 @@ import org.zillion.util.log.LogUtil;
 
 import com.saga.entity.PointSet;
 import com.saga.entity.Record;
+import com.saga.entity.Report;
 import com.sagacloud.json.JSONArray;
 import com.sagacloud.json.JSONObject;
 import com.zillion.database.agent.ZillionAgent;
@@ -1875,7 +1876,7 @@ public class HBaseUtil {
 	// }
 	// }
 
-	public static void BatchInsert(ZillionAgent agent, String DB, List<Record> dataList) throws Exception {
+	public static void BatchInsert(ZillionAgent agent, String DB, List<Report> dataList) throws Exception {
 		JSONArray InsertObjects = new JSONArray();
 		try {
 			boolean history = false;
@@ -1892,7 +1893,7 @@ public class HBaseUtil {
 			if (DB != null && dataList.size() > 0) {
 				for (int i = 0; i < dataList.size(); i++) {
 
-					Record data = dataList.get(i);
+					Record data = dataList.get(i).record;
 
 					JSONObject InsertObject = new JSONObject();
 					InsertObject.put("building", data.buildingSign);