Browse Source

数据迁移 - 支持数据中台一键迁移至数字孪生平台

yucheng 3 năm trước cách đây
mục cha
commit
501bc23ee2

+ 76 - 0
dmp-business/dmp-rwd/src/main/java/com/persagy/dmp/rwd/migrate/controller/DigitalMigrateController.java

@@ -0,0 +1,76 @@
+package com.persagy.dmp.rwd.migrate.controller;
+
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.thread.ThreadUtil;
+import cn.hutool.core.util.StrUtil;
+import com.persagy.dmp.common.constant.ResponseCode;
+import com.persagy.dmp.common.exception.BusinessException;
+import com.persagy.dmp.common.model.response.CommonResult;
+import com.persagy.dmp.common.utils.ResultHelper;
+import com.persagy.dmp.rwd.migrate.service.IDigitalMigrateService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 版本定义Controller
+ * @author Charlie Yu
+ * @date 2021-06-23
+ */
+@RestController
+@RequestMapping("/migrate/digital")
+public class DigitalMigrateController {
+
+    @Autowired
+    private IDigitalMigrateService service;
+
+    /**
+     * 启动数据迁移
+     * 迁移范围:自定义数据字典、对象、关系
+     * @param requestMap
+     * {
+     *     "orgUrl":"数据中台组织服务url前缀(不含最后的/)",
+     *     "oldSchema":"数据中台使用的数据库实例",
+     *     "groupCode":"指定集团编码【单个或数组】 - 为空表示不限集团",
+     *     "projectId":"指定项目ID【单个或数组】 - 为空表示不限项目"
+     * }
+     */
+    @PostMapping("/start")
+    public CommonResult<String> query(@RequestBody Map<String, Object> requestMap) {
+        // 处理参数
+        final String orgUrl = MapUtil.getStr(requestMap, "orgUrl");
+        final String oldSchema = MapUtil.getStr(requestMap, "oldSchema");
+        if(StrUtil.isBlank(orgUrl) || StrUtil.isBlank(oldSchema) ) {
+            throw new BusinessException(ResponseCode.A0400.getCode(), ResponseCode.A0400.getDesc());
+        }
+        // 集团编码
+        final List<String> groupCodes = new ArrayList<>();
+        Object groupCode = requestMap.get("groupCode");
+        if(groupCode != null) {
+            if(groupCode instanceof String) {
+                groupCodes.add((String) groupCode);
+            } else if(groupCode instanceof List) {
+                groupCodes.addAll((List<String>) groupCode);
+            }
+        }
+        // 项目ID
+        final List<String> projectIds = new ArrayList<>();
+        Object projectId = requestMap.get("projectId");
+        if(projectId != null) {
+            if(projectId instanceof String) {
+                projectIds.add((String) projectId);
+            } else if(projectId instanceof List) {
+                projectIds.addAll((List<String>) projectId);
+            }
+        }
+        // 启动迁移 执行时间较长,后台线程运行 TODO 需要加分布式锁,并支持查看执行情况
+        ThreadUtil.execute(() -> service.startMigrate(groupCodes, projectIds, orgUrl, oldSchema));
+        return ResultHelper.success();
+    }
+}

+ 56 - 0
dmp-business/dmp-rwd/src/main/java/com/persagy/dmp/rwd/migrate/dao/DigitalMigrateMapper.java

@@ -0,0 +1,56 @@
+package com.persagy.dmp.rwd.migrate.dao;
+
+import org.apache.ibatis.annotations.Param;
+
+/**
+ * 数字孪生 数据迁移dao
+ * @author Charlie Yu
+ * @date 2021-08-04
+ */
+public interface DigitalMigrateMapper {
+
+    /**
+     * 迁移类型定义数据
+     * @param schema
+     * @param groupCode
+     */
+    void migrateDefineType(@Param("schema") String schema, @Param("groupCode") String groupCode);
+
+    /**
+     * 迁移信息点定义数据
+     * @param schema
+     * @param groupCode
+     */
+    void migrateDefineInfo(@Param("schema") String schema, @Param("groupCode") String groupCode);
+
+    /**
+     * 迁移图定义数据
+     * @param schema
+     * @param groupCode
+     */
+    void migrateDefineGraph(@Param("schema") String schema, @Param("groupCode") String groupCode);
+
+    /**
+     * 迁移边定义数据
+     * @param schema
+     * @param groupCode
+     */
+    void migrateDefineRel(@Param("schema") String schema, @Param("groupCode") String groupCode);
+
+    /**
+     * 迁移对象数据
+     * @param schema
+     * @param groupCode
+     * @param projectId
+     */
+    void migrateObject(@Param("schema") String schema, @Param("groupCode") String groupCode, @Param("projectId") String projectId);
+
+    /**
+     * 迁移关系数据
+     * @param schema
+     * @param groupCode
+     * @param projectId
+     */
+    void migrateRelation(@Param("schema") String schema, @Param("groupCode") String groupCode, @Param("projectId") String projectId);
+
+}

+ 20 - 0
dmp-business/dmp-rwd/src/main/java/com/persagy/dmp/rwd/migrate/service/IDigitalMigrateService.java

@@ -0,0 +1,20 @@
+package com.persagy.dmp.rwd.migrate.service;
+
+import java.util.List;
+
+/**
+ * 数字孪生 数据迁移接口
+ * @author Charlie Yu
+ * @date 2021-08-04
+ */
+public interface IDigitalMigrateService {
+
+    /**
+     * 启动数据迁移
+     * @param groupCodes 指定集团编码 - 为空表示不限集团
+     * @param projectIds 指定项目ID - 为空表示不限项目
+     * @param orgUrl 数据中台组织服务url前缀(不含最后的/)
+     * @param oldSchema 数据中台使用的数据库实例
+     */
+    void startMigrate(List<String> groupCodes, List<String> projectIds, String orgUrl, String oldSchema);
+}

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

@@ -0,0 +1,184 @@
+package com.persagy.dmp.rwd.migrate.service.impl;
+
+import cn.hutool.core.collection.CollUtil;
+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.service.IDigitalMigrateService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.annotation.Resource;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * 数字孪生 数据迁移
+ * @author Charlie Yu
+ * @date 2021-08-04
+ */
+@Service
+@Transactional(rollbackFor = Exception.class)
+@Slf4j
+public class DigitalMigrateServiceImpl implements IDigitalMigrateService {
+
+    @Resource
+    private DigitalMigrateMapper dao;
+
+    @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;
+        }
+        // 按集团循环处理
+        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;
+            }
+            // 集团数据迁移
+            doMigrate(oldSchema, groupCode);
+            // 项目数据迁移
+            for(String projectId:projectIdSet) {
+                doMigrate(oldSchema, groupCode, projectId);
+            }
+        }
+    }
+
+    /**
+     * 执行集团级数据迁移
+     * @param schema 数据中台库实例
+     * @param groupCode 集团编码
+     */
+    private void doMigrate(String schema, String groupCode) {
+        // 切换集团数据源
+        AppContext.getContext().setGroupCode(groupCode);
+        DynamicDataSourceHelper.loadDataSource();
+        // 变小写
+        groupCode = groupCode.toLowerCase();
+        // 迁移
+        dao.migrateDefineType(schema, groupCode);
+        dao.migrateDefineInfo(schema, groupCode);
+        dao.migrateDefineGraph(schema, groupCode);
+        dao.migrateDefineRel(schema, groupCode);
+    }
+
+    /**
+     * 执行项目级迁移
+     * @param schema 数据中台库实例
+     * @param groupCode 集团编码
+     * @param projectId 项目ID
+     */
+    private void doMigrate(String schema, String groupCode, String projectId) {
+        // 已经切换到集团数据源了,不需要再切换了
+        groupCode = groupCode.toLowerCase();
+        projectId = projectId.toLowerCase();
+        // 迁移
+        dao.migrateObject(schema, groupCode, projectId);
+        dao.migrateRelation(schema, groupCode, projectId);
+    }
+
+    /**
+     * 查询项目主键
+     * @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;
+    }
+}

+ 0 - 5
dmp-business/dmp-rwd/src/main/resources/mapper/DictDataMapper.xml

@@ -1,5 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
-<mapper namespace="DictDataMapper">
-
-</mapper>

+ 72 - 0
dmp-business/dmp-rwd/src/main/resources/mapper/DigitalMigrateMapper.xml

@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.persagy.dmp.rwd.migrate.dao.DigitalMigrateMapper">
+
+    <insert id="migrateDefineType">
+        replace into `dt_define_type`(`id`,`code`,`name`,`alias_code`,
+          `alias_name`,`obj_type`,`group_code`,`project_id`,`major_code`,
+          `system_code`,`equipment_code`,`parent_code`,`version`,`creator`,
+          `creation_time`,`modifier`,`modified_time`,`valid`)
+        SELECT `id`,`code`,`name`,`alias_code`,
+          `alias_name`,`obj_type`,`group_code`,`project_id`,`major_code`,
+          `system_code`,`equipment_code`,`parent_code`,null,'0',
+          date_format(now(), '%Y%m%d%h%i%s'),null,null,1
+        FROM `${schema}`.`rwd_def_class_${groupCode}`
+	</insert>
+    <insert id="migrateDefineInfo">
+        replace into `dt_define_info`(`id`,`code`,`name`,`alias_code`,
+          `alias_name`,`class_code`,`group_code`,`project_id`,`category`,
+          `first_tag`,`second_tag`,`priority`,`input_mode`,`unit`,`data_type`,
+          `multiple_flag`,`region_flag`,`formatter`,`data_source`,`remark`,
+          `sub_flag`,`weak_point`,`show_order`,`version`,
+          `creator`,`creation_time`,`modifier`,`modified_time`,`valid`)
+        SELECT `id`,`code`,`name`,`alias_code`,
+          `alias_name`,`class_code`,`group_code`,`project_id`,`category`,
+          `first_tag`,`second_tag`,`priority`,`input_mode`,`unit`,`data_type`,
+          `is_multiple`,`is_region`,`formater`,`data_source`,`note`,
+          `sub_flag`,`weak_point`,null,null,
+          '0',date_format(now(), '%Y%m%d%h%i%s'),null,null,1
+        FROM `${schema}`.`rwd_def_funcid_${groupCode}`
+    </insert>
+    <insert id="migrateDefineGraph">
+        replace into `dt_define_graph`(`id`,`code`,`name`,`first_category`,
+          `second_category`,`remark`,`object_types`,
+          `creator`,`creation_time`,`modifier`,`modified_time`,`valid`)
+        SELECT `id`,`code`,`name`,`first_category`,
+          `second_category`,`category_remark`,`object_types`,
+          '0',date_format(now(), '%Y%m%d%h%i%s'),null,null,1
+        FROM `${schema}`.`rwd_def_graph_${groupCode}`
+    </insert>
+    <insert id="migrateDefineRel">
+        replace into `dt_define_rel`(`id`,`code`,`name`,`graph_code`,
+          `directional`,`attribute`,`target_objs`,`remark`,
+          `creator`,`creation_time`,`modifier`,`modified_time`,`valid`)
+        SELECT `id`,`code`,`name`,`graph_code`,
+          `directional`,`attribute`,`target_objs`,`remark`,
+          '0',date_format(now(), '%Y%m%d%h%i%s'),null,null,1
+        FROM `${schema}`.`rwd_def_rel_${groupCode}`
+    </insert>
+    <insert id="migrateObject">
+        replace into `dt_object`(`id`,`name`,`local_id`,
+          `local_name`,`group_code`,`project_id`,`obj_type`,`class_code`,
+          `grouping`,`infos`,`virtual_codes`,`create_app`,`update_app`,
+          `creator`,`creation_time`,`modifier`,`modified_time`,`valid`)
+        SELECT `id`,`name`,`local_id`,
+          `local_name`,`group_code`,`project_id`,`obj_type`,`class_code`,
+          `grouping`,`infos`,`virtual_info_codes`,`create_app`,`update_app`,
+          '0',date_format(now(), '%Y%m%d%h%i%s'),null,null,1
+        FROM `${schema}`.`rwd_object_${groupCode}_${projectId}`
+    </insert>
+    <insert id="migrateRelation">
+        replace into `dt_relation`(`id`,`graph_id`,`graph_code`,
+          `rel_code`,`rel_value`,`group_code`,`project_id`,`obj_from`,`obj_to`,
+          `create_app`,`update_app`,
+          `creator`,`creation_time`,`modifier`,`modified_time`,`valid`)
+        SELECT `id`,`graph_id`,`graph_code`,
+          `rel_code`,`rel_value`,`group_code`,`project_id`,`obj_from`,`obj_to`,
+          `create_app`,`update_app`,
+          '0',date_format(now(), '%Y%m%d%h%i%s'),null,null,1
+        FROM `${schema}`.`rwd_relation_${groupCode}_${projectId}`
+    </insert>
+
+</mapper>