gfwleak/nms-nmssync
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
default
2018-11-02 09:46:35 +08:00
parent 63974e7ad6
commit 0ce15c7c7d
3 changed files with 101 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
nms_sync/src/com/nms/interceptor/SyncMissionResultStatisticalInterceptor.java
View File

@@ -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) {
}
}

34
nms_sync/src/com/nms/main/SyncData.java
View File

@@ -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();
}
}

14
nms_sync/src/com/nms/thread/SyncSlaveToMasterThread.java
View File

@@ -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")) {