Browse Source

fix adm 数据迁移 新增sql脚本 方式 数据量大的时候oom

lvxianyun 3 years ago
parent
commit
72c687027d

+ 4 - 3
adm-business/adm-middleware/src/main/java/com/persagy/proxy/migration/service/Impl/ObjectDigitalMigration.java

@@ -103,14 +103,14 @@ public class ObjectDigitalMigration extends MigrationAbstractServiceImpl<ObjectD
     public String migrateForSql(InstanceUrlParam context) {
         StringBuffer sqls = new StringBuffer(" \n -- 数据迁移  dt_object ------ \n");
         //获取已经采集的数据
-        StringBuffer sqls_temp = new StringBuffer();
+        String sqls_temp = "";
         for(MigrationObjType objType : MigrationObjType.values()){
             QueryCriteria queryCriteria = getQueryCriteria(objType.getCode());
             List<ObjectNode> admData = getAdmData(context,queryCriteria);
             if(CollUtil.isEmpty(admData)){
                 continue;
             }
-            appendValues(admData, sqls_temp);
+            sqls_temp += appendValues(admData).toString();
         }
         if(sqls_temp.length() == 0){
             return "";
@@ -124,7 +124,8 @@ public class ObjectDigitalMigration extends MigrationAbstractServiceImpl<ObjectD
         return sql;
     }
 
-    private StringBuffer appendValues(List<ObjectNode> admData, StringBuffer sqls){
+    private StringBuffer appendValues(List<ObjectNode> admData){
+        StringBuffer sqls = new StringBuffer();
         admData.forEach(obj -> {
             ObjectDigital objectDigital = JsonNodeUtils.toEntity(obj, ObjectDigital.class, "infos");
             sqls.append("(");

+ 65 - 4
adm-business/adm-middleware/src/main/java/com/persagy/proxy/migration/service/Impl/ObjectRelationMigration.java

@@ -8,8 +8,8 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.persagy.dmp.basic.model.QueryCriteria;
 import com.persagy.dmp.basic.utils.JsonNodeUtils;
 import com.persagy.dmp.common.constant.ResponseCode;
-import com.persagy.dmp.common.model.entity.AuditableEntity;
 import com.persagy.dmp.common.model.entity.BaseEntity;
+import com.persagy.dmp.common.model.response.CommonResult;
 import com.persagy.dmp.define.client.DigitalGraphDefineFacade;
 import com.persagy.dmp.define.client.DigitalRelationDefineFacade;
 import com.persagy.dmp.define.entity.GraphDefine;
@@ -20,6 +20,7 @@ import com.persagy.dmp.digital.entity.ObjectRelation;
 import com.persagy.dmp.rwd.migrate.entity.SyncData;
 import com.persagy.proxy.adm.constant.AdmCommonConstant;
 import com.persagy.proxy.adm.request.AdmResponse;
+import com.persagy.proxy.common.entity.DmpResult;
 import com.persagy.proxy.common.entity.InstanceUrlParam;
 import com.persagy.proxy.migration.constant.*;
 import com.persagy.proxy.migration.model.DataMigrationExcel;
@@ -136,7 +137,7 @@ public class ObjectRelationMigration extends MigrationAbstractServiceImpl<Object
     }
 
     /**
-     * 数据迁移
+     * 数据迁移 数据量超过1w 分页 多线程处理
      *
      * @param context
      * @return sql 文件
@@ -145,15 +146,50 @@ public class ObjectRelationMigration extends MigrationAbstractServiceImpl<Object
     public String migrateForSql(InstanceUrlParam context) {
         StringBuffer sqls = new StringBuffer(" \n -- 数据迁移  dt_relation ------ \n");
         //获取已经采集的数据
-        StringBuffer sqls_temp = new StringBuffer();
+        /* StringBuffer sqls_temp = new StringBuffer();
         for (MiGrationRelCode miGrationRelCode: MiGrationRelCode.values()){
             QueryCriteria queryCriteria = getQueryCriteria(miGrationRelCode.getCode());
             List<ObjectRelation> admData = getAdmData(context,queryCriteria);
             if(CollUtil.isEmpty(admData)){
                 continue;
             }
-            appendValues(admData, sqls_temp);
+        }*/
+
+        String sqls_temp = "";
+        ExecutorService service = getExcecutor();
+        List<Future<StringBuffer>> futureList = new ArrayList<>();
+        for (MiGrationRelCode miGrationRelCode: MiGrationRelCode.values()){
+            QueryCriteria queryCriteria = getQueryCriteria(miGrationRelCode.getCode());
+            queryCriteria.setSize(10000L);
+            CommonResult<List<ObjectRelation>> admDatas = getAdmDataPage(context,queryCriteria);
+            if (admDatas.getResult().equals(DmpResult.SUCCESS) && admDatas.getCount() > queryCriteria.getSize()) {
+                long pages = admDatas.getCount() / queryCriteria.getSize() + 1;
+                for(long i = 2;i < pages; i++){
+                    queryCriteria.setPage(i);
+                    List<ObjectRelation> admData = getAdmData(context,queryCriteria);
+                    appendValuesBefore(service,admData,futureList);
+                }
+            }else{
+                List<ObjectRelation> admData = admDatas.getData();
+                appendValuesBefore(service,admData,futureList);
+            }
         }
+        service.shutdown();
+        for (Future<StringBuffer> future : futureList) {
+            try {
+                StringBuffer stringBuffer = future.get();
+                if(StrUtil.isNotEmpty(stringBuffer)){
+                    sqls_temp += stringBuffer.toString();
+                }
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+                log.error("######################### dt_relation 数据迁移sql失败"+e.getMessage());
+            } catch (ExecutionException e) {
+                e.printStackTrace();
+                log.error("######################### dt_relation 数据迁移sql失败"+e.getMessage());
+            }
+        }
+
         if(sqls_temp.length() == 0){
             return "";
         }
@@ -165,6 +201,26 @@ public class ObjectRelationMigration extends MigrationAbstractServiceImpl<Object
         return sql;
     }
 
+    private void appendValuesBefore(ExecutorService service, List<ObjectRelation> admData, List<Future<StringBuffer>> futureList){
+        if(CollUtil.isEmpty(admData)){
+            return;
+        }
+        StringBuffer sqls_temp = new StringBuffer();
+        Future<StringBuffer> future = service.submit(new Callable<StringBuffer>(){
+            @Override
+            public StringBuffer call() throws Exception {
+                return appendValues(admData, sqls_temp);
+            }
+        });
+        futureList.add(future);
+    }
+
+    /**
+     * 拼接sql字符串
+     * @param admData
+     * @param sqls
+     * @return
+     */
     private StringBuffer appendValues(List<ObjectRelation> admData, StringBuffer sqls){
         admData.forEach(relation -> {
             sqls.append("(");
@@ -344,6 +400,11 @@ public class ObjectRelationMigration extends MigrationAbstractServiceImpl<Object
         return relations;
     }
 
+    public CommonResult<List<ObjectRelation>> getAdmDataPage(InstanceUrlParam context, QueryCriteria queryCriteria) {
+        CommonResult<List<ObjectRelation>> relations = DigitalRelationFacade.queryPrototype(context.getGroupCode(),context.getProjectId(),context.getUserId(),context.getUserId(),queryCriteria);
+        return relations;
+    }
+
     /**
      * 根据map转为需要处理的对象集合
      * 使用 MigrationConstant.spliter 作为拼接符