update

3 files changed, 101 insertions, 12 deletions

diff --git a/nms_sync/src/com/nms/interceptor/SyncMissionResultStatisticalInterceptor.java b/nms_sync/src/com/nms/interceptor/SyncMissionResultStatisticalInterceptor.java
index d9c2e56..d4eabc2 100644
--- a/nms_sync/src/com/nms/interceptor/SyncMissionResultStatisticalInterceptor.java
+++ b/nms_sync/src/com/nms/interceptor/SyncMissionResultStatisticalInterceptor.java
@@ -13,6 +13,7 @@ import com.jfinal.aop.Interceptor;
 import com.jfinal.aop.Invocation;

 import com.jfinal.plugin.activerecord.Db;

 import com.jfinal.plugin.activerecord.Record;

+import com.nms.main.SyncData;

 import com.nms.model.SyncDbInfo;

 import com.nms.thread.SyncSlaveToMasterThread;

 

@@ -24,8 +25,8 @@ public class SyncMissionResultStatisticalInterceptor implements Interceptor{
 	@Override

 	public void intercept(Invocation inv) {

 		try{

-			logger.info("---------SyncMissionResultStatisticalInterceptor拦截器拦截开始------------");

 			

+			/*

 			//创建一个任务id集合 存储任务结果改变的任务id 这些任务统一走新的统计方法 修改mission_state_table状态

 			

 			//新增或者修改的结果可能是多条 但是任务id相同 用set进行去重

@@ -55,10 +56,9 @@ public class SyncMissionResultStatisticalInterceptor implements Interceptor{
 			

 			Record beforeUpdateMissionResultTable6 = Db.use(url.toString()).findFirst("select * from table_sync_info where table_name='mission_result_table6' and event=2 and db_id=-1");

 			logger.info("获取mission_result_table1表中最后一次同步id的数据信息为 "+JSON.toJSONString(beforeUpdateMissionResultTable6));

+			*/

 			

-			

-			inv.invoke();

-			

+			/*

 			

 			Record afterInsertMissionResultTable1 = Db.use(url.toString()).findFirst("select * from table_sync_info where table_name='mission_result_table1' and event=1 and db_id=-1");

 			logger.info("获取mission_result_table1表中最后一次同步id的数据信息为 "+JSON.toJSONString(afterInsertMissionResultTable1));

@@ -176,16 +176,43 @@ public class SyncMissionResultStatisticalInterceptor implements Interceptor{
 						}

 					}

 				}

-			}

+			}*/

+			

+			logger.info("---------SyncMissionResultStatisticalInterceptor拦截器拦截开始------------");

+			inv.invoke();

+			

+			/**

+			 * 处理非周期任务的统计功能

+			 */

+			Set<Long> insertMissionIds = SyncData.getThreadlocalInsertMissionIds();

+			logger.info("即将进行统计的所有修改任务的mission_id为:"+JSON.toJSONString(insertMissionIds));

+			

+			Set<Long> updateMissionIds = SyncData.getThreadlocalUpdateMissionIds();

+			logger.info("即将进行统计的所有修改任务的mission_id为:"+JSON.toJSONString(updateMissionIds));

+			

 			

-			logger.info("即将进行统计的所有任务的mission_id为:"+JSON.toJSONString(missionIds));

 			

 			//根据统计结果更新mission_state_talbe表对应任务的状态

-			if(missionIds.size()>0) {

-				for (Long missionId : missionIds) {

-					StatisticalHandle(missionId);

+			if(insertMissionIds.size()>0) {

+				for (Long missionId : insertMissionIds) {

+					StatisticalHandle(missionId,true);

 				}

 			}

+			

+			if(updateMissionIds.size()>0) {

+				for (Long missionId : updateMissionIds) {

+					StatisticalHandle(missionId,false);

+				}

+			}

+			

+			/**

+			 * 处理周期任务的统计功能

+			 */

+			

+			

+			// 统计结束后  清空threadLocal中的值

+			SyncData.removeThreadlocalUpdateMissionIds();

+			SyncData.removeThreadlocalInsertMissionIds();

 			logger.info("--------SyncMissionResultStatisticalInterceptor拦截器拦截结束------------");

 		}catch(Exception e){

 			e.printStackTrace();

@@ -194,7 +221,10 @@ public class SyncMissionResultStatisticalInterceptor implements Interceptor{
 		

 	}

 	

-	public void StatisticalHandle(Long missionId) {

+	/**

+	 * 处理非周期任务的统计功能

+	 */

+	public void StatisticalHandle(Long missionId,boolean isInsert) {

 		logger.info("根据当前任务id为:"+missionId+"开始统计");

 		//  mission_state_table 状态值

 		Integer status = null;

@@ -244,6 +274,13 @@ public class SyncMissionResultStatisticalInterceptor implements Interceptor{
 			logger.info("统计完成 修改mission_state_table状态为:? 总个数:？执行数:? 成功:? 失败:?");

 			

 			String missionStateAutoDesc=null;

+			if(isInsert) {

+				Record missionStateTable = Db.use().findFirst("select * from mission_state_table where mission_id = ?",missionId);

+				if(null!=missionStateTable&&null!=missionStateTable.get("auto_desc")&&missionStateTable.getStr("auto_desc").length()>0) {

+					return;

+				}

+				missionStateAutoDesc=format.format(System.currentTimeMillis())+"i18n_server.MissionConstants.NOTICE_TASK_RUNNING_n81i </br>";

+			}

 			if(okCount+failCount==total) {

 				missionStateAutoDesc=format.format(System.currentTimeMillis())+" i18n_sserver.UpgradeService.sql.complate_n81i "+total+" i18n_sserver.UpgradeService.sql.executeNode_n81i "+okCount+" i18n_sserver.UpgradeService.sql.failed_n81i "+failCount;

 			}

@@ -261,4 +298,12 @@ public class SyncMissionResultStatisticalInterceptor implements Interceptor{
 		}

 		logger.info("修改mission_state_table信息完成");

 	}

+	

+	

+	/**

+	 * 处理周期任务的统计功能

+	 */

+	public void StatisticalLoopHandle(Long missionId) {

+		

+	}

 }

diff --git a/nms_sync/src/com/nms/main/SyncData.java b/nms_sync/src/com/nms/main/SyncData.java
index 8589302..37a1a03 100644
--- a/nms_sync/src/com/nms/main/SyncData.java
+++ b/nms_sync/src/com/nms/main/SyncData.java
@@ -1,5 +1,6 @@
 package com.nms.main;
 
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -25,8 +26,9 @@ import com.nms.thread.SyncThread;
  *
  */
 public class SyncData{
+	private static final ThreadLocal<Set<Long>> threadLocalUpdateMissionIds = new ThreadLocal<Set<Long>>();
+	private static final ThreadLocal<Set<Long>> threadLocalInsertMissionIds = new ThreadLocal<Set<Long>>();
 	
-	private static final ThreadLocal<Set<Long>> threadLocal = new ThreadLocal<Set<Long>>();
 	
 	public static void main(String[] args) {
 		Logger logger = Logger.getLogger(SyncData.class);
@@ -68,4 +70,34 @@ public class SyncData{
 			logger.info("获取同步记录信息失败 请检查数据库数据信息");
 		}
 	}
+
+	public static Set<Long> getThreadlocalUpdateMissionIds() {
+		Set<Long> set = threadLocalUpdateMissionIds.get();
+		if(null==set) {
+			set=new HashSet<Long>();
+		}
+		return set;
+	}
+	public static void setThreadlocalUpdateMissionIds(Set<Long> set) {
+		threadLocalUpdateMissionIds.set(set);
+	}
+	
+	public static void removeThreadlocalUpdateMissionIds() {
+		threadLocalUpdateMissionIds.remove();
+	}
+	
+	public static Set<Long> getThreadlocalInsertMissionIds() {
+		Set<Long> set = threadLocalInsertMissionIds.get();
+		if(null==set) {
+			set=new HashSet<Long>();
+		}
+		return set;
+	}
+	public static void setThreadlocalInsertMissionIds(Set<Long> set) {
+		threadLocalInsertMissionIds.set(set);
+	}
+	
+	public static void removeThreadlocalInsertMissionIds() {
+		threadLocalInsertMissionIds.remove();
+	}
 }
diff --git a/nms_sync/src/com/nms/thread/SyncSlaveToMasterThread.java b/nms_sync/src/com/nms/thread/SyncSlaveToMasterThread.java
index de50297..edd1782 100644
--- a/nms_sync/src/com/nms/thread/SyncSlaveToMasterThread.java
+++ b/nms_sync/src/com/nms/thread/SyncSlaveToMasterThread.java
@@ -4,9 +4,11 @@ import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.Set;
 
 import org.apache.log4j.Logger;
 import com.nms.interceptor.SyncMissionResultStatisticalInterceptor;
+import com.nms.main.SyncData;
 import com.alibaba.fastjson.JSON;
 import com.jfinal.aop.Before;
 import com.jfinal.plugin.activerecord.Db;
@@ -67,7 +69,14 @@ public class SyncSlaveToMasterThread implements Runnable{
 								//logger.info("分库同步到主库新增数据信息为"+JSON.toJSONString(data));
 								if (data != null && data.size() > 0) {
 									final Object lastInsertId = data.get(data.size() - 1).get(record.getStr("id_name"));
-									if(record.getInt("mode").equals(2)) {
+									if(record.getInt("mode").equals(2)&&record.getStr("table_name").contains("mission_result_table")) {
+										Set<Long> set = SyncData.getThreadlocalInsertMissionIds();
+										for(Record entity:data) {
+											set.add(entity.getLong("mission_id"));
+											entity.remove(record.getStr("id_name"));
+										}
+										SyncData.setThreadlocalInsertMissionIds(set);
+									}else {
 										for(Record entity:data) {
 											entity.remove(record.getStr("id_name"));
 										}
@@ -245,9 +254,12 @@ public class SyncSlaveToMasterThread implements Runnable{
 															}
 															Db.use("masterDataSource").batchUpdate(record.getStr("table_name"), "DETECTION_SET_INFO_ID,SEQ_ID",updateDatas, record.getInt("batch_size"));
 														}else if(record.getStr("table_name").contains("mission_result_table")){
+															Set<Long> missionIds = SyncData.getThreadlocalUpdateMissionIds();
 															for(Record updateData:updateDatas) {
+																missionIds.add(updateData.getLong("mission_id"));
 																updateData.remove(record.getStr("id_name"));
 															}
+															SyncData.setThreadlocalUpdateMissionIds(missionIds);
 															Db.use("masterDataSource").batchUpdate(record.getStr("table_name"), "mission_id,seq_id",updateDatas, record.getInt("batch_size"));
 														}else {
 															if(record.getStr("table_name").equals("detection_info_new")) {