summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--nms_sync/src/com/nms/interceptor/SyncMissionResultStatisticalInterceptor.java65
-rw-r--r--nms_sync/src/com/nms/main/SyncData.java34
-rw-r--r--nms_sync/src/com/nms/thread/SyncSlaveToMasterThread.java14
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")) {
generated by cgit v1.2.3 (git 2.39.1) at 2025-12-30 17:38:29 +0000