Parcourir la source

优化代码,避免迁移时可能会出现的无事务风险。

yucheng il y a 3 ans
Parent
commit
6fdddac689

+ 135 - 4
dmp-business/dmp-rwd/src/main/java/com/persagy/dmp/rwd/migrate/controller/DigitalMigrateController.java

@@ -1,28 +1,34 @@
 package com.persagy.dmp.rwd.migrate.controller;
 package com.persagy.dmp.rwd.migrate.controller;
 
 
+import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.thread.ThreadUtil;
 import cn.hutool.core.thread.ThreadUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.core.util.StrUtil;
+import cn.hutool.http.HttpUtil;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.persagy.dmp.common.constant.ResponseCode;
 import com.persagy.dmp.common.constant.ResponseCode;
+import com.persagy.dmp.common.context.AppContext;
 import com.persagy.dmp.common.exception.BusinessException;
 import com.persagy.dmp.common.exception.BusinessException;
 import com.persagy.dmp.common.model.response.CommonResult;
 import com.persagy.dmp.common.model.response.CommonResult;
 import com.persagy.dmp.common.utils.ResultHelper;
 import com.persagy.dmp.common.utils.ResultHelper;
+import com.persagy.dmp.mybatis.helper.DynamicDataSourceHelper;
 import com.persagy.dmp.rwd.migrate.service.IDigitalMigrateService;
 import com.persagy.dmp.rwd.migrate.service.IDigitalMigrateService;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.bind.annotation.RestController;
 
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 
 /**
 /**
  * 版本定义Controller
  * 版本定义Controller
  * @author Charlie Yu
  * @author Charlie Yu
  * @date 2021-06-23
  * @date 2021-06-23
  */
  */
+@Slf4j
 @RestController
 @RestController
 @RequestMapping("/migrate/digital")
 @RequestMapping("/migrate/digital")
 public class DigitalMigrateController {
 public class DigitalMigrateController {
@@ -70,7 +76,132 @@ public class DigitalMigrateController {
             }
             }
         }
         }
         // 启动迁移 执行时间较长,后台线程运行 TODO 需要加分布式锁,并支持查看执行情况
         // 启动迁移 执行时间较长,后台线程运行 TODO 需要加分布式锁,并支持查看执行情况
-        ThreadUtil.execute(() -> service.startMigrate(groupCodes, projectIds, orgUrl, oldSchema));
+        ThreadUtil.execute(() -> startMigrate(groupCodes, projectIds, orgUrl, oldSchema));
         return ResultHelper.success();
         return ResultHelper.success();
     }
     }
+
+    /**
+     * 启动数据迁移
+     * @param groupCodes 指定集团编码 - 为空表示不限集团
+     * @param projectIds 指定项目ID - 为空表示不限项目
+     * @param orgUrl 数据中台组织服务url前缀(不含最后的/)
+     * @param oldSchema 数据中台使用的数据库实例
+     */
+    private void startMigrate(List<String> groupCodes, List<String> projectIds, String orgUrl, String oldSchema) {
+        Set<String> groupCodeSet = queryGroupCodes(orgUrl, groupCodes);
+        if(CollUtil.isEmpty(groupCodeSet)) {
+            log.error("没有查询到集团信息!");
+            return;
+        }
+        log.info("#########数据迁移已启动##########");
+        // 按集团循环处理
+        for(String groupCode:groupCodeSet) {
+            log.info(StrUtil.format("开始处理{}集团", groupCode));
+            // 获取待处理的项目
+            Set<String> projectIdSet = queryProjectIds(orgUrl, groupCode, projectIds);
+            if(CollUtil.isEmpty(projectIdSet)) {
+                log.info(StrUtil.format("结束处理{}集团:没有找到需要迁移的项目", groupCode));
+                continue;
+            }
+            // 切换集团数据源
+            AppContext.getContext().setGroupCode(groupCode);
+            DynamicDataSourceHelper.loadDataSource();
+            // 集团数据迁移
+            service.doMigrate(oldSchema, groupCode, projectIdSet);
+            log.info(StrUtil.format("结束处理{}集团:迁移完成!!!", groupCode));
+        }
+        log.info("#########数据迁移已结束##########");
+    }
+
+    /**
+     * 查询项目主键
+     * @param orgUrl 组织服务url
+     * @param groupCode 集团编码,非空
+     * @param ids 指定的项目ID,为空表示所有
+     * @return
+     */
+    private Set<String> queryProjectIds(String orgUrl, String groupCode, List<String> ids) {
+        // 调用数据中台接口查询项目信息
+        String queryProjectUrl = StrUtil.format(orgUrl + "/org/project/query?groupCode={}&userId=systemId", groupCode);
+        String response = HttpUtil.post(queryProjectUrl, "{}");
+        JSONArray projects = fetchResult(response);
+        if(CollUtil.isEmpty(projects)) {
+            return null;
+        }
+        // 收集项目ID
+        Set<String> projectIds = new HashSet<>();
+        for(int i = 0;i < projects.size();i++) {
+            JSONObject project = projects.getJSONObject(i);
+            String id = project.getString("id");
+            boolean valid = project.getBooleanValue("status");
+            // 已删除或没有编码的不处理
+            if(StrUtil.isBlank(id) || !valid) {
+                continue;
+            }
+            projectIds.add(id);
+        }
+        // 处理返回结果
+        if(CollUtil.isEmpty(projectIds)) {
+            return null;
+        } else if(CollUtil.isNotEmpty(ids)) {
+            // 有指定的集团,则取交集
+            return CollUtil.intersectionDistinct(projectIds, ids);
+        } else {
+            // 没有指定集团,则返回所有有效集团
+            return projectIds;
+        }
+    }
+
+    /**
+     * 查询集团编码
+     * @param orgUrl 组织服务url
+     * @param codes 指定集团编码 - 为空表示所有集团
+     * @return
+     */
+    private Set<String> queryGroupCodes(String orgUrl, List<String> codes) {
+        // 调用数据中台接口查询集团信息
+        String queryGroupUrl = orgUrl + "/org/group/query";
+        String groupResponse = HttpUtil.post(queryGroupUrl, "{}");
+        JSONArray groups = fetchResult(groupResponse);
+        if(CollUtil.isEmpty(groups)) {
+            return null;
+        }
+        // 收集集团编码
+        Set<String> groupCodes = new HashSet<>();
+        for(int i = 0;i < groups.size();i++) {
+            JSONObject group = groups.getJSONObject(i);
+            String code = group.getString("code");
+            boolean valid = group.getBooleanValue("status");
+            // 已删除或没有编码的不处理
+            if(StrUtil.isBlank(code) || !valid) {
+                continue;
+            }
+            groupCodes.add(code);
+        }
+        // 处理返回结果
+        if(CollUtil.isEmpty(groupCodes)) {
+            return null;
+        } else if(CollUtil.isNotEmpty(codes)) {
+            // 有指定的集团,则取交集
+            return CollUtil.intersectionDistinct(groupCodes, codes);
+        } else {
+            // 没有指定集团,则返回所有有效集团
+            return groupCodes;
+        }
+    }
+
+    /**
+     * 提取结果
+     * @param response
+     * @return
+     */
+    private JSONArray fetchResult(String response) {
+        JSONObject responseObject = JSONObject.parseObject(response);
+        // 如果结果不成功,抛出失败结果
+        if(!StrUtil.equals("success", responseObject.getString("result"))) {
+            throw new BusinessException(responseObject.getString("message"));
+        }
+        JSONArray datas = responseObject.getJSONArray("data");
+        return CollUtil.isEmpty(datas) ? null : datas;
+    }
 }
 }

+ 6 - 7
dmp-business/dmp-rwd/src/main/java/com/persagy/dmp/rwd/migrate/service/IDigitalMigrateService.java

@@ -1,6 +1,6 @@
 package com.persagy.dmp.rwd.migrate.service;
 package com.persagy.dmp.rwd.migrate.service;
 
 
-import java.util.List;
+import java.util.Set;
 
 
 /**
 /**
  * 数字孪生 数据迁移接口
  * 数字孪生 数据迁移接口
@@ -10,11 +10,10 @@ import java.util.List;
 public interface IDigitalMigrateService {
 public interface IDigitalMigrateService {
 
 
     /**
     /**
-     * 启动数据迁移
-     * @param groupCodes 指定集团编码 - 为空表示不限集团
-     * @param projectIds 指定项目ID - 为空表示不限项目
-     * @param orgUrl 数据中台组织服务url前缀(不含最后的/)
-     * @param oldSchema 数据中台使用的数据库实例
+     * 执行迁移
+     * @param schema 数据中台实例名
+     * @param groupCode 集团编码 - 非空
+     * @param projectIds 此集团内的项目 - 非空
      */
      */
-    void startMigrate(List<String> groupCodes, List<String> projectIds, String orgUrl, String oldSchema);
+    void doMigrate(String schema, String groupCode, Set<String> projectIds);
 }
 }

+ 2 - 134
dmp-business/dmp-rwd/src/main/java/com/persagy/dmp/rwd/migrate/service/impl/DigitalMigrateServiceImpl.java

@@ -1,13 +1,6 @@
 package com.persagy.dmp.rwd.migrate.service.impl;
 package com.persagy.dmp.rwd.migrate.service.impl;
 
 
-import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.core.util.StrUtil;
-import cn.hutool.http.HttpUtil;
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
-import com.persagy.dmp.common.context.AppContext;
-import com.persagy.dmp.common.exception.BusinessException;
-import com.persagy.dmp.mybatis.helper.DynamicDataSourceHelper;
 import com.persagy.dmp.rwd.migrate.dao.DigitalMigrateMapper;
 import com.persagy.dmp.rwd.migrate.dao.DigitalMigrateMapper;
 import com.persagy.dmp.rwd.migrate.service.IDigitalMigrateService;
 import com.persagy.dmp.rwd.migrate.service.IDigitalMigrateService;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
@@ -15,8 +8,6 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.transaction.annotation.Transactional;
 
 
 import javax.annotation.Resource;
 import javax.annotation.Resource;
-import java.util.HashSet;
-import java.util.List;
 import java.util.Set;
 import java.util.Set;
 
 
 /**
 /**
@@ -26,46 +17,14 @@ import java.util.Set;
  */
  */
 @Service
 @Service
 @Slf4j
 @Slf4j
+@Transactional(rollbackFor = Exception.class)
 public class DigitalMigrateServiceImpl implements IDigitalMigrateService {
 public class DigitalMigrateServiceImpl implements IDigitalMigrateService {
 
 
     @Resource
     @Resource
     private DigitalMigrateMapper dao;
     private DigitalMigrateMapper dao;
 
 
     @Override
     @Override
-    public void startMigrate(List<String> groupCodes, List<String> projectIds, String orgUrl, String oldSchema) {
-        Set<String> groupCodeSet = queryGroupCodes(orgUrl, groupCodes);
-        if(CollUtil.isEmpty(groupCodeSet)) {
-            log.error("没有查询到集团信息!");
-            return;
-        }
-        log.info("#########数据迁移已启动##########");
-        // 按集团循环处理
-        for(String groupCode:groupCodeSet) {
-            log.info(StrUtil.format("开始处理{}集团", groupCode));
-            // 获取待处理的项目
-            Set<String> projectIdSet = queryProjectIds(orgUrl, groupCode, projectIds);
-            if(CollUtil.isEmpty(projectIdSet)) {
-                log.info(StrUtil.format("结束处理{}集团:没有找到需要迁移的项目", groupCode));
-                continue;
-            }
-            // 切换集团数据源
-            AppContext.getContext().setGroupCode(groupCode);
-            DynamicDataSourceHelper.loadDataSource();
-            // 集团数据迁移
-            doMigrate(oldSchema, groupCode, projectIdSet);
-            log.info(StrUtil.format("结束处理{}集团:迁移完成!!!", groupCode));
-        }
-        log.info("#########数据迁移已结束##########");
-    }
-
-    /**
-     * 执行迁移,在此方法上启动事务
-     * @param schema
-     * @param groupCode
-     * @param projectIds
-     */
-    @Transactional(rollbackFor = Exception.class)
-    protected void doMigrate(String schema, String groupCode, Set<String> projectIds) {
+    public void doMigrate(String schema, String groupCode, Set<String> projectIds) {
         // 变小写
         // 变小写
         groupCode = groupCode.toLowerCase();
         groupCode = groupCode.toLowerCase();
         // 集团数据迁移
         // 集团数据迁移
@@ -132,95 +91,4 @@ public class DigitalMigrateServiceImpl implements IDigitalMigrateService {
         }
         }
     }
     }
 
 
-    /**
-     * 查询项目主键
-     * @param orgUrl 组织服务url
-     * @param groupCode 集团编码,非空
-     * @param ids 指定的项目ID,为空表示所有
-     * @return
-     */
-    private Set<String> queryProjectIds(String orgUrl, String groupCode, List<String> ids) {
-        // 调用数据中台接口查询项目信息
-        String queryProjectUrl = StrUtil.format(orgUrl + "/org/project/query?groupCode={}&userId=systemId", groupCode);
-        String response = HttpUtil.post(queryProjectUrl, "{}");
-        JSONArray projects = fetchResult(response);
-        if(CollUtil.isEmpty(projects)) {
-            return null;
-        }
-        // 收集项目ID
-        Set<String> projectIds = new HashSet<>();
-        for(int i = 0;i < projects.size();i++) {
-            JSONObject project = projects.getJSONObject(i);
-            String id = project.getString("id");
-            boolean valid = project.getBooleanValue("status");
-            // 已删除或没有编码的不处理
-            if(StrUtil.isBlank(id) || !valid) {
-                continue;
-            }
-            projectIds.add(id);
-        }
-        // 处理返回结果
-        if(CollUtil.isEmpty(projectIds)) {
-            return null;
-        } else if(CollUtil.isNotEmpty(ids)) {
-            // 有指定的集团,则取交集
-            return CollUtil.intersectionDistinct(projectIds, ids);
-        } else {
-            // 没有指定集团,则返回所有有效集团
-            return projectIds;
-        }
-    }
-
-    /**
-     * 查询集团编码
-     * @param orgUrl 组织服务url
-     * @param codes 指定集团编码 - 为空表示所有集团
-     * @return
-     */
-    private Set<String> queryGroupCodes(String orgUrl, List<String> codes) {
-        // 调用数据中台接口查询集团信息
-        String queryGroupUrl = orgUrl + "/org/group/query";
-        String groupResponse = HttpUtil.post(queryGroupUrl, "{}");
-        JSONArray groups = fetchResult(groupResponse);
-        if(CollUtil.isEmpty(groups)) {
-            return null;
-        }
-        // 收集集团编码
-        Set<String> groupCodes = new HashSet<>();
-        for(int i = 0;i < groups.size();i++) {
-            JSONObject group = groups.getJSONObject(i);
-            String code = group.getString("code");
-            boolean valid = group.getBooleanValue("status");
-            // 已删除或没有编码的不处理
-            if(StrUtil.isBlank(code) || !valid) {
-                continue;
-            }
-            groupCodes.add(code);
-        }
-        // 处理返回结果
-        if(CollUtil.isEmpty(groupCodes)) {
-            return null;
-        } else if(CollUtil.isNotEmpty(codes)) {
-            // 有指定的集团,则取交集
-            return CollUtil.intersectionDistinct(groupCodes, codes);
-        } else {
-            // 没有指定集团,则返回所有有效集团
-            return groupCodes;
-        }
-    }
-
-    /**
-     * 提取结果
-     * @param response
-     * @return
-     */
-    private JSONArray fetchResult(String response) {
-        JSONObject responseObject = JSONObject.parseObject(response);
-        // 如果结果不成功,抛出失败结果
-        if(!StrUtil.equals("success", responseObject.getString("result"))) {
-            throw new BusinessException(responseObject.getString("message"));
-        }
-        JSONArray datas = responseObject.getJSONArray("data");
-        return CollUtil.isEmpty(datas) ? null : datas;
-    }
 }
 }