实时同步任务列表

database/mcb/update.sql

@@ -5,4 +5,5 @@ alter table dmp_develop_task add TARGET_DB_NAME varchar(64) default NULL COMMENT
 alter table dmp_develop_task add TARGET_TABLE_NAME varchar(64) default NULL COMMENT'目标数据表名称';
 alter table dmp_syncing_datasource_type add DB_ATTRS json default null comment '数据库属性';
 alter table dmp_syncing_datasource add NETWORK_CONNECTION_TYPE varchar(64) default null comment '网络连接类型';
-alter table dmp_syncing_datasource add TEST_CONNECT_STATUS char(2) default '01' comment '测试连通状态：01未测试，02连通性正常，03连通性异常';
+alter table dmp_syncing_datasource add TEST_CONNECT_STATUS char(2) default '01' comment '测试连通状态：01未测试，02连通性正常，03连通性异常';
\ No newline at end of file
+alter table dmp_realtime_sync_info add tree_id varchar(64) DEFAULT NULL COMMENT 'treeID';

src/main/java/com/jz/common/utils/realTime/CmdUtils.java

+package com.jz.common.utils.realTime;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+public class CmdUtils {
+	
+	private static final Logger LOGGER = LoggerFactory.getLogger(CmdUtils.class);
+	
+	/**
+	 * 执行shell命令
+	 * @param shellPath shell路径
+	 * @param args shell参数
+	 * @return
+	 */
+	public static boolean callShell(String shellPath,String ...args) {
+		
+		boolean result = true;
+
+		try {
+
+		StringBuffer shellSb = new StringBuffer();
+		shellSb.append(shellPath);
+		if (args != null && args.length >= 1) {
+			for (String arg : args) {
+				shellSb.append(" ");
+				shellSb.append(arg);
+				shellSb.append(" ");
+			}
+		}
+		
+		String shell = shellSb.toString();
+		LOGGER.info("执行shell:"+shell);
+
+			Process process = Runtime.getRuntime().exec(shell);
+			int status = 0 ;
+			try {
+				status = process.waitFor();
+				LOGGER.info("执行shell:"+shell+" 返回status:"+status);
+				if(status != 0){  
+		        	result = false;
+		        }
+			} catch (InterruptedException e) {
+				result = false;
+				e.printStackTrace();
+			}  
+	        
+		} catch (IOException e) {
+			result = false ;
+			e.printStackTrace();
+			
+		}
+		return result;
+	}
+}

src/main/java/com/jz/dmp/modules/controller/DataIntegration/RealTimeSyncController.java

+package com.jz.dmp.modules.controller.DataIntegration;
+
+import com.jz.common.constant.JsonResult;
+import com.jz.common.constant.ResultCode;
+import com.jz.common.page.PageInfoResponse;
+import com.jz.common.utils.realTime.CmdUtils;
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListDto;
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListReq;
+import com.jz.dmp.modules.model.DmpRealtimeSyncInfo;
+import com.jz.dmp.modules.service.DmpRealtimeSyncInfoService;
+import com.jz.dmp.modules.service.impl.OfflineSynchServiceImpl;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiOperation;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+
+import javax.servlet.http.HttpServletRequest;
+
+/**
+ * 实时同步任务
+ *
+ * @author Bellamy
+ * @since 2021-01-05 10:56:18
+ */
+@RestController
+@RequestMapping("/realTimeSync")
+@Api(tags = "数据集成--实时同步任务")
+public class RealTimeSyncController {
+
+    private static Logger logger = LoggerFactory.getLogger(RealTimeSyncController.class);
+    /**
+     * 服务对象
+     */
+    @Autowired
+    private DmpRealtimeSyncInfoService dmpRealtimeSyncInfoService;
+
+    /**
+     * 实时同步任务列表分页查询
+     *
+     * @return
+     * @author Bellamy
+     * @since 2021-01-05
+     */
+    @ApiOperation(value = "实时同步任务列表分页查询", notes = "实时同步任务列表分页查询")
+    @PostMapping(value = "/realTimeSyncListPage")
+    public PageInfoResponse<RealTimeSyncListDto> getDataSourceListPage(@RequestBody RealTimeSyncListReq req, HttpServletRequest httpRequest) throws Exception {
+        PageInfoResponse<RealTimeSyncListDto> pageInfo = new PageInfoResponse<RealTimeSyncListDto>();
+        if (StringUtils.isEmpty(req.getProjectId())) {
+            pageInfo.setMessage("项目id不能为空！");
+            pageInfo.setCode(ResultCode.PARAMS_ERROR);
+            return pageInfo;
+        }
+        try {
+            pageInfo = dmpRealtimeSyncInfoService.queryRealTimeSyncListPage(req);
+        } catch (Exception e) {
+            pageInfo.setMessage("查询失败");
+            pageInfo.setCode(ResultCode.INTERNAL_SERVER_ERROR);
+            e.printStackTrace();
+        }
+        return pageInfo;
+    }
+
+    /**
+     * 启动实时同步任务
+     *
+     * @return
+     * @author Bellamy
+     * @since 2021-01-05
+     */
+    @ApiOperation(value = "启动实时同步任务", notes = "启动实时同步任务")
+    @GetMapping(value = "/startRealTimeSync")
+    @ApiImplicitParam(name = "realTaskId", value = "任务id")
+    public JsonResult startRealTimeSync(@RequestParam String realTaskId) throws Exception {
+        if (StringUtils.isEmpty(realTaskId)) {
+            return new JsonResult(ResultCode.PARAMS_ERROR, "任务id不能为空！");
+        }
+        DmpRealtimeSyncInfo dmpRealtimeSyncInfo = dmpRealtimeSyncInfoService.queryById(Integer.valueOf(realTaskId));
+        String srcTopicName = dmpRealtimeSyncInfo.getSrcTopicName();
+        System.out.println(srcTopicName);
+        logger.info("############正常执行表数据........"+realTaskId);
+        String shellPath ="/app/bigdata-app/scripts/trigger_straming.sh";
+        CmdUtils.callShell(shellPath, srcTopicName);
+        return new JsonResult();
+    }
+
+}
\ No newline at end of file

src/main/java/com/jz/dmp/modules/controller/DataIntegration/bean/RealTimeSyncListDto.java

+package com.jz.dmp.modules.controller.DataIntegration.bean;
+
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+
+/**
+ * @ClassName: RealTimeSyncListDto
+ * @Description: 实时同步列表返回参数
+ * @Author：Bellamy
+ * @Date 2021/01/05
+ * @Version 1.0
+ */
+@ApiModel(value = "实时同步列表返回参数", description = "实时同步列表返回参数")
+public class RealTimeSyncListDto {
+
+    /*
+     * 实时同步任务ID
+     * */
+    @ApiModelProperty(value = "实时同步任务ID")
+    private String id;
+
+    /*
+     * 业务树节点id
+     * */
+    @ApiModelProperty(value = "业务树节点id")
+    private String treeId;
+
+    /*
+     * 业务树节点名称
+     * */
+    @ApiModelProperty(value = "业务树节点名称")
+    private String treeName;
+
+    /*
+     * 实时同步任务状态
+     * */
+    @ApiModelProperty(value = "实时同步任务状态")
+    private String status;
+
+    /*
+     * 更新时间
+     * */
+    @ApiModelProperty(value = "更新时间")
+    private String updateTime;
+
+    /*
+     * 来源数据源id
+     * */
+    @ApiModelProperty(value = "来源数据源id")
+    private String srcDatasourceId;
+
+    /*
+     * 来源数据源名称
+     * */
+    @ApiModelProperty(value = "来源数据源名称")
+    private String srcDatasourceName;
+
+    /*
+     * 目标数据源名称
+     * */
+    @ApiModelProperty(value = "来源数据源类型")
+    private String srcDatabaseType;
+
+    /*
+     * 目标数据源id
+     * */
+    @ApiModelProperty(value = "目标数据源id")
+    private String targetDatasourceId;
+
+    /*
+     * 目标数据源名称
+     * */
+    @ApiModelProperty(value = "目标数据源名称")
+    private String targetDatasourceName;
+
+    /*
+     * 目标数据源类型
+     * */
+    @ApiModelProperty(value = "目标数据源类型")
+    private String targetDatabaseType;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getTreeId() {
+        return treeId;
+    }
+
+    public void setTreeId(String treeId) {
+        this.treeId = treeId;
+    }
+
+    public String getTreeName() {
+        return treeName;
+    }
+
+    public void setTreeName(String treeName) {
+        this.treeName = treeName;
+    }
+
+    public String getStatus() {
+        return status;
+    }
+
+    public void setStatus(String status) {
+        this.status = status;
+    }
+
+    public String getUpdateTime() {
+        return updateTime;
+    }
+
+    public void setUpdateTime(String updateTime) {
+        this.updateTime = updateTime;
+    }
+
+    public String getSrcDatasourceId() {
+        return srcDatasourceId;
+    }
+
+    public void setSrcDatasourceId(String srcDatasourceId) {
+        this.srcDatasourceId = srcDatasourceId;
+    }
+
+    public String getSrcDatasourceName() {
+        return srcDatasourceName;
+    }
+
+    public void setSrcDatasourceName(String srcDatasourceName) {
+        this.srcDatasourceName = srcDatasourceName;
+    }
+
+    public String getSrcDatabaseType() {
+        return srcDatabaseType;
+    }
+
+    public void setSrcDatabaseType(String srcDatabaseType) {
+        this.srcDatabaseType = srcDatabaseType;
+    }
+
+    public String getTargetDatasourceId() {
+        return targetDatasourceId;
+    }
+
+    public void setTargetDatasourceId(String targetDatasourceId) {
+        this.targetDatasourceId = targetDatasourceId;
+    }
+
+    public String getTargetDatasourceName() {
+        return targetDatasourceName;
+    }
+
+    public void setTargetDatasourceName(String targetDatasourceName) {
+        this.targetDatasourceName = targetDatasourceName;
+    }
+
+    public String getTargetDatabaseType() {
+        return targetDatabaseType;
+    }
+
+    public void setTargetDatabaseType(String targetDatabaseType) {
+        this.targetDatabaseType = targetDatabaseType;
+    }
+}

src/main/java/com/jz/dmp/modules/controller/DataIntegration/bean/RealTimeSyncListReq.java

+package com.jz.dmp.modules.controller.DataIntegration.bean;
+
+
+import com.jz.common.page.BasePageBean;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+
+/**
+ * @ClassName: RealTimeSyncListReq
+ * @Description: 实时同步列表请求参数
+ * @Author：Bellamy
+ * @Date 2021/01/05
+ * @Version 1.0
+ */
+@ApiModel(value = "实时同步列表请求参数", description = "实时同步列表请求参数")
+public class RealTimeSyncListReq extends BasePageBean {
+    /*
+     * 源数据类型id
+     * */
+    @ApiModelProperty(value = "来源数据类型id")
+    private String sourceDatabaseTypeId;
+
+    /*
+     * 目标数据源名称
+     * */
+    @ApiModelProperty(value = "来源数据源名称")
+    private String sourceDatabaseName;
+
+    /*
+     * 目标数据类型id
+     * */
+    @ApiModelProperty(value = "目标数据类型id")
+    private String targetDatabaseTypeId;
+
+    /*
+     * 目标数据源名称
+     * */
+    @ApiModelProperty(value = "目标数据源名称")
+    private String targetDatabaseName;
+
+    /*
+     * 项目id
+     * */
+    @ApiModelProperty(value = "项目id")
+    @NotNull(message = "项目id不能为空")
+    @NotEmpty(message = "项目ID不能空")
+    private String projectId;
+
+    /*
+     * 目标数据源名称
+     * */
+    @ApiModelProperty(value = "目标数据源名称")
+    private String taskStatus;
+
+    /*
+     * 节点id
+     * */
+    @ApiModelProperty(value = "节点id")
+    private String treeId;
+
+    public String getProjectId() {
+        return projectId;
+    }
+
+    public void setProjectId(String projectId) {
+        this.projectId = projectId;
+    }
+
+    public String getSourceDatabaseTypeId() {
+        return sourceDatabaseTypeId;
+    }
+
+    public void setSourceDatabaseTypeId(String sourceDatabaseTypeId) {
+        this.sourceDatabaseTypeId = sourceDatabaseTypeId;
+    }
+
+    public String getSourceDatabaseName() {
+        return sourceDatabaseName;
+    }
+
+    public void setSourceDatabaseName(String sourceDatabaseName) {
+        this.sourceDatabaseName = sourceDatabaseName;
+    }
+
+    public String getTargetDatabaseTypeId() {
+        return targetDatabaseTypeId;
+    }
+
+    public void setTargetDatabaseTypeId(String targetDatabaseTypeId) {
+        this.targetDatabaseTypeId = targetDatabaseTypeId;
+    }
+
+    public String getTargetDatabaseName() {
+        return targetDatabaseName;
+    }
+
+    public void setTargetDatabaseName(String targetDatabaseName) {
+        this.targetDatabaseName = targetDatabaseName;
+    }
+
+    public String getTaskStatus() {
+        return taskStatus;
+    }
+
+    public void setTaskStatus(String taskStatus) {
+        this.taskStatus = taskStatus;
+    }
+
+    public String getTreeId() {
+        return treeId;
+    }
+
+    public void setTreeId(String treeId) {
+        this.treeId = treeId;
+    }
+}

src/main/java/com/jz/dmp/modules/dao/DmpRealtimeSyncInfoDao.java

+package com.jz.dmp.modules.dao;
+
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListDto;
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListReq;
+import com.jz.dmp.modules.model.DmpRealtimeSyncInfo;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+/**
+ * 实时同步任务(DmpRealtimeSyncInfo)表数据库访问层
+ *
+ * @author Bellamy
+ * @since 2021-01-05 14:18:03
+ */
+public interface DmpRealtimeSyncInfoDao {
+
+    /**
+     * 通过ID查询单条数据
+     *
+     * @param id 主键
+     * @return 实例对象
+     */
+    DmpRealtimeSyncInfo queryById(Integer id);
+
+    /**
+     * 查询指定行数据
+     *
+     * @param offset 查询起始位置
+     * @param limit  查询条数
+     * @return 对象列表
+     */
+    List<DmpRealtimeSyncInfo> queryAllByLimit(@Param("offset") int offset, @Param("limit") int limit);
+
+
+    /**
+     * 通过实体作为筛选条件查询
+     *
+     * @param dmpRealtimeSyncInfo 实例对象
+     * @return 对象列表
+     */
+    List<DmpRealtimeSyncInfo> queryAll(DmpRealtimeSyncInfo dmpRealtimeSyncInfo);
+
+    /**
+     * 新增数据
+     *
+     * @param dmpRealtimeSyncInfo 实例对象
+     * @return 影响行数
+     */
+    int insert(DmpRealtimeSyncInfo dmpRealtimeSyncInfo);
+
+    /**
+     * 批量新增数据（MyBatis原生foreach方法）
+     *
+     * @param entities List<DmpRealtimeSyncInfo> 实例对象列表
+     * @return 影响行数
+     */
+    int insertBatch(@Param("entities") List<DmpRealtimeSyncInfo> entities);
+
+    /**
+     * 批量新增或按主键更新数据（MyBatis原生foreach方法）
+     *
+     * @param entities List<DmpRealtimeSyncInfo> 实例对象列表
+     * @return 影响行数
+     */
+    int insertOrUpdateBatch(@Param("entities") List<DmpRealtimeSyncInfo> entities);
+
+    /**
+     * 修改数据
+     *
+     * @param dmpRealtimeSyncInfo 实例对象
+     * @return 影响行数
+     */
+    int update(DmpRealtimeSyncInfo dmpRealtimeSyncInfo);
+
+    /**
+     * 通过主键删除数据
+     *
+     * @param id 主键
+     * @return 影响行数
+     */
+    int deleteById(Integer id);
+
+    /**
+     * 实时同步任务列表分页查询
+     * @return
+     */
+    List<RealTimeSyncListDto> queryRealTimeSyncListPage(RealTimeSyncListReq req) throws Exception;
+}
\ No newline at end of file

src/main/java/com/jz/dmp/modules/model/DmpRealtimeSyncInfo.java

+package com.jz.dmp.modules.model;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * 实时同步任务(DmpRealtimeSyncInfo)实体类
+ *
+ * @author makejava
+ * @since 2021-01-05 14:18:00
+ */
+@ApiModel(value = "实时同步任务", description = "实时同步任务")
+public class DmpRealtimeSyncInfo implements Serializable {
+    private static final long serialVersionUID = 622628445093766366L;
+    /**
+     * 实时同步任务ID
+     */
+    @ApiModelProperty(value = "实时同步任务ID")
+    private Integer id;
+    /**
+     * 源数据库id
+     */
+    @ApiModelProperty(value = "源数据库id")
+    private Integer srcDatasourceId;
+    /**
+     * 目标数据库id
+     */
+    @ApiModelProperty(value = "目标数据库id")
+    private Integer targetDatasourceId;
+    /**
+     * 源数据库表名称
+     */
+    @ApiModelProperty(value = "源数据库表名称")
+    private String srcTableName;
+    /**
+     * 目标数据库表名称
+     */
+    @ApiModelProperty(value = "目标数据库表名称")
+    private String targetTableName;
+
+    @ApiModelProperty(value = "${column.comment}")
+    private Integer type;
+
+    @ApiModelProperty(value = "${column.comment}")
+    private String connectorJobId;
+
+    @ApiModelProperty(value = "${column.comment}")
+    private Object connectorJsonData;
+
+    @ApiModelProperty(value = "${column.comment}")
+    private String srcTopicName;
+    /**
+     * 项目id
+     */
+    @ApiModelProperty(value = "项目id")
+    private Object projectId;
+
+    @ApiModelProperty(value = "${column.comment}")
+    private Integer parentId;
+    /**
+     * 脱敏字段
+     */
+    @ApiModelProperty(value = "脱敏字段")
+    private String desensitizationField;
+    /**
+     * 加密算法
+     */
+    @ApiModelProperty(value = "加密算法")
+    private String arithmetic;
+    /**
+     * 主键名称
+     */
+    @ApiModelProperty(value = "主键名称")
+    private String pkName;
+    /**
+     * 源类型名称
+     */
+    @ApiModelProperty(value = "源类型名称")
+    private String sourceTypeName;
+    /**
+     * 目标类型名称
+     */
+    @ApiModelProperty(value = "目标类型名称")
+    private String targetTypeName;
+    /**
+     * 源数据库类型
+     */
+    @ApiModelProperty(value = "源数据库类型")
+    private String srcDatabaseType;
+    /**
+     * 源数据库名称
+     */
+    @ApiModelProperty(value = "源数据库名称")
+    private String srcDatabaseName;
+    /**
+     * 目标数据源URL连接信息
+     */
+    @ApiModelProperty(value = "目标数据源URL连接信息")
+    private String connectorUrl;
+    /**
+     * 目标数据库类型
+     */
+    @ApiModelProperty(value = "目标数据库类型")
+    private String targetDatabaseType;
+    /**
+     * 目标数据库名称
+     */
+    @ApiModelProperty(value = "目标数据库名称")
+    private String targetDatabaseName;
+    /**
+     * 源数据库
+     */
+    @ApiModelProperty(value = "源数据库")
+    private String srcDatasourceName;
+    /**
+     * 目标数据源库
+     */
+    @ApiModelProperty(value = "目标数据源库")
+    private String targetDatasourceName;
+    /**
+     * 存储类型
+     */
+    @ApiModelProperty(value = "存储类型")
+    private String storeType;
+    /**
+     * 任务状态
+     */
+    @ApiModelProperty(value = "任务状态")
+    private String status;
+    /**
+     * 创建时间
+     */
+    @ApiModelProperty(value = "创建时间")
+    private Date createTime;
+    /**
+     * 更新时间
+     */
+    @ApiModelProperty(value = "更新时间")
+    private Date updateTime;
+    /**
+     * 创建人
+     */
+    @ApiModelProperty(value = "创建人")
+    private String crePerson;
+    /**
+     * 更新人
+     */
+    @ApiModelProperty(value = "更新人")
+    private String uptPerson;
+
+
+    public Integer getId() {
+        return id;
+    }
+
+    public void setId(Integer id) {
+        this.id = id;
+    }
+
+    public Integer getSrcDatasourceId() {
+        return srcDatasourceId;
+    }
+
+    public void setSrcDatasourceId(Integer srcDatasourceId) {
+        this.srcDatasourceId = srcDatasourceId;
+    }
+
+    public Integer getTargetDatasourceId() {
+        return targetDatasourceId;
+    }
+
+    public void setTargetDatasourceId(Integer targetDatasourceId) {
+        this.targetDatasourceId = targetDatasourceId;
+    }
+
+    public String getSrcTableName() {
+        return srcTableName;
+    }
+
+    public void setSrcTableName(String srcTableName) {
+        this.srcTableName = srcTableName;
+    }
+
+    public String getTargetTableName() {
+        return targetTableName;
+    }
+
+    public void setTargetTableName(String targetTableName) {
+        this.targetTableName = targetTableName;
+    }
+
+    public Integer getType() {
+        return type;
+    }
+
+    public void setType(Integer type) {
+        this.type = type;
+    }
+
+    public String getConnectorJobId() {
+        return connectorJobId;
+    }
+
+    public void setConnectorJobId(String connectorJobId) {
+        this.connectorJobId = connectorJobId;
+    }
+
+    public Object getConnectorJsonData() {
+        return connectorJsonData;
+    }
+
+    public void setConnectorJsonData(Object connectorJsonData) {
+        this.connectorJsonData = connectorJsonData;
+    }
+
+    public String getSrcTopicName() {
+        return srcTopicName;
+    }
+
+    public void setSrcTopicName(String srcTopicName) {
+        this.srcTopicName = srcTopicName;
+    }
+
+    public Object getProjectId() {
+        return projectId;
+    }
+
+    public void setProjectId(Object projectId) {
+        this.projectId = projectId;
+    }
+
+    public Integer getParentId() {
+        return parentId;
+    }
+
+    public void setParentId(Integer parentId) {
+        this.parentId = parentId;
+    }
+
+    public String getDesensitizationField() {
+        return desensitizationField;
+    }
+
+    public void setDesensitizationField(String desensitizationField) {
+        this.desensitizationField = desensitizationField;
+    }
+
+    public String getArithmetic() {
+        return arithmetic;
+    }
+
+    public void setArithmetic(String arithmetic) {
+        this.arithmetic = arithmetic;
+    }
+
+    public String getPkName() {
+        return pkName;
+    }
+
+    public void setPkName(String pkName) {
+        this.pkName = pkName;
+    }
+
+    public String getSourceTypeName() {
+        return sourceTypeName;
+    }
+
+    public void setSourceTypeName(String sourceTypeName) {
+        this.sourceTypeName = sourceTypeName;
+    }
+
+    public String getTargetTypeName() {
+        return targetTypeName;
+    }
+
+    public void setTargetTypeName(String targetTypeName) {
+        this.targetTypeName = targetTypeName;
+    }
+
+    public String getSrcDatabaseType() {
+        return srcDatabaseType;
+    }
+
+    public void setSrcDatabaseType(String srcDatabaseType) {
+        this.srcDatabaseType = srcDatabaseType;
+    }
+
+    public String getSrcDatabaseName() {
+        return srcDatabaseName;
+    }
+
+    public void setSrcDatabaseName(String srcDatabaseName) {
+        this.srcDatabaseName = srcDatabaseName;
+    }
+
+    public String getConnectorUrl() {
+        return connectorUrl;
+    }
+
+    public void setConnectorUrl(String connectorUrl) {
+        this.connectorUrl = connectorUrl;
+    }
+
+    public String getTargetDatabaseType() {
+        return targetDatabaseType;
+    }
+
+    public void setTargetDatabaseType(String targetDatabaseType) {
+        this.targetDatabaseType = targetDatabaseType;
+    }
+
+    public String getTargetDatabaseName() {
+        return targetDatabaseName;
+    }
+
+    public void setTargetDatabaseName(String targetDatabaseName) {
+        this.targetDatabaseName = targetDatabaseName;
+    }
+
+    public String getSrcDatasourceName() {
+        return srcDatasourceName;
+    }
+
+    public void setSrcDatasourceName(String srcDatasourceName) {
+        this.srcDatasourceName = srcDatasourceName;
+    }
+
+    public String getTargetDatasourceName() {
+        return targetDatasourceName;
+    }
+
+    public void setTargetDatasourceName(String targetDatasourceName) {
+        this.targetDatasourceName = targetDatasourceName;
+    }
+
+    public String getStoreType() {
+        return storeType;
+    }
+
+    public void setStoreType(String storeType) {
+        this.storeType = storeType;
+    }
+
+    public String getStatus() {
+        return status;
+    }
+
+    public void setStatus(String status) {
+        this.status = status;
+    }
+
+    public Date getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(Date createTime) {
+        this.createTime = createTime;
+    }
+
+    public Date getUpdateTime() {
+        return updateTime;
+    }
+
+    public void setUpdateTime(Date updateTime) {
+        this.updateTime = updateTime;
+    }
+
+    public String getCrePerson() {
+        return crePerson;
+    }
+
+    public void setCrePerson(String crePerson) {
+        this.crePerson = crePerson;
+    }
+
+    public String getUptPerson() {
+        return uptPerson;
+    }
+
+    public void setUptPerson(String uptPerson) {
+        this.uptPerson = uptPerson;
+    }
+
+}
\ No newline at end of file

src/main/java/com/jz/dmp/modules/service/DmpRealtimeSyncInfoService.java

+package com.jz.dmp.modules.service;
+
+import com.jz.common.page.PageInfoResponse;
+import com.jz.dmp.modules.controller.DataIntegration.bean.DataSourceListDto;
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListDto;
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListReq;
+import com.jz.dmp.modules.model.DmpRealtimeSyncInfo;
+
+import java.util.List;
+
+/**
+ * 实时同步任务(DmpRealtimeSyncInfo)表服务接口
+ *
+ * @author Bellamy
+ * @since 2021-01-05 14:18:03
+ */
+public interface DmpRealtimeSyncInfoService {
+
+    /**
+     * 通过ID查询单条数据
+     *
+     * @param id 主键
+     * @return 实例对象
+     */
+    DmpRealtimeSyncInfo queryById(Integer id);
+
+    /**
+     * 查询多条数据
+     *
+     * @param offset 查询起始位置
+     * @param limit  查询条数
+     * @return 对象列表
+     */
+    List<DmpRealtimeSyncInfo> queryAllByLimit(int offset, int limit);
+
+    /**
+     * 新增数据
+     *
+     * @param dmpRealtimeSyncInfo 实例对象
+     * @return 实例对象
+     */
+    DmpRealtimeSyncInfo insert(DmpRealtimeSyncInfo dmpRealtimeSyncInfo);
+
+    /**
+     * 修改数据
+     *
+     * @param dmpRealtimeSyncInfo 实例对象
+     * @return 实例对象
+     */
+    DmpRealtimeSyncInfo update(DmpRealtimeSyncInfo dmpRealtimeSyncInfo);
+
+    /**
+     * 通过主键删除数据
+     *
+     * @param id 主键
+     * @return 是否成功
+     */
+    boolean deleteById(Integer id);
+
+    /**
+     * 实时同步任务列表分页查询
+     *
+     * @return
+     */
+    PageInfoResponse<RealTimeSyncListDto> queryRealTimeSyncListPage(RealTimeSyncListReq req) throws Exception;
+}
\ No newline at end of file

src/main/java/com/jz/dmp/modules/service/impl/DmpRealtimeSyncInfoServiceImpl.java

+package com.jz.dmp.modules.service.impl;
+
+import com.github.pagehelper.PageHelper;
+import com.github.pagehelper.PageInfo;
+import com.jz.common.constant.ResultCode;
+import com.jz.common.page.PageInfoResponse;
+import com.jz.dmp.modules.controller.DataIntegration.bean.DataSourceListDto;
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListDto;
+import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListReq;
+import com.jz.dmp.modules.controller.DataIntegration.bean.TaskListPageDto;
+import com.jz.dmp.modules.dao.DmpRealtimeSyncInfoDao;
+import com.jz.dmp.modules.model.DmpRealtimeSyncInfo;
+import com.jz.dmp.modules.service.DmpRealtimeSyncInfoService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+
+/**
+ * 实时同步任务(DmpRealtimeSyncInfo)表服务实现类
+ *
+ * @author Bellamy
+ * @since 2021-01-05 14:18:03
+ */
+@Service("dmpRealtimeSyncInfoService")
+public class DmpRealtimeSyncInfoServiceImpl implements DmpRealtimeSyncInfoService {
+
+    @Autowired
+    private DmpRealtimeSyncInfoDao dmpRealtimeSyncInfoDao;
+
+    /**
+     * 通过ID查询单条数据
+     *
+     * @param id 主键
+     * @return 实例对象
+     */
+    @Override
+    public DmpRealtimeSyncInfo queryById(Integer id) {
+        return this.dmpRealtimeSyncInfoDao.queryById(id);
+    }
+
+    /**
+     * 查询多条数据
+     *
+     * @param offset 查询起始位置
+     * @param limit  查询条数
+     * @return 对象列表
+     */
+    @Override
+    public List<DmpRealtimeSyncInfo> queryAllByLimit(int offset, int limit) {
+        return this.dmpRealtimeSyncInfoDao.queryAllByLimit(offset, limit);
+    }
+
+    /**
+     * 新增数据
+     *
+     * @param dmpRealtimeSyncInfo 实例对象
+     * @return 实例对象
+     */
+    @Override
+    public DmpRealtimeSyncInfo insert(DmpRealtimeSyncInfo dmpRealtimeSyncInfo) {
+        this.dmpRealtimeSyncInfoDao.insert(dmpRealtimeSyncInfo);
+        return dmpRealtimeSyncInfo;
+    }
+
+    /**
+     * 修改数据
+     *
+     * @param dmpRealtimeSyncInfo 实例对象
+     * @return 实例对象
+     */
+    @Override
+    public DmpRealtimeSyncInfo update(DmpRealtimeSyncInfo dmpRealtimeSyncInfo) {
+        this.dmpRealtimeSyncInfoDao.update(dmpRealtimeSyncInfo);
+        return this.queryById(dmpRealtimeSyncInfo.getId());
+    }
+
+    /**
+     * 通过主键删除数据
+     *
+     * @param id 主键
+     * @return 是否成功
+     */
+    @Override
+    public boolean deleteById(Integer id) {
+        return this.dmpRealtimeSyncInfoDao.deleteById(id) > 0;
+    }
+
+    /**
+     * 实时同步任务列表分页查询
+     * @return
+     */
+    @Override
+    public PageInfoResponse<RealTimeSyncListDto> queryRealTimeSyncListPage(RealTimeSyncListReq req) throws Exception {
+        PageInfoResponse<RealTimeSyncListDto> pageInfoResponse = new PageInfoResponse<>();
+
+        PageHelper.startPage(req.getPageNum(), req.getPageSize());
+        List<RealTimeSyncListDto> list = dmpRealtimeSyncInfoDao.queryRealTimeSyncListPage(req);
+        PageInfo<RealTimeSyncListDto> pageInfo = new PageInfo<>(list);
+
+        pageInfoResponse.setCode(ResultCode.SUCCESS);
+        pageInfoResponse.setMessage("查询成功");
+        pageInfoResponse.setData(pageInfo);
+        return pageInfoResponse;
+    }
+}
\ No newline at end of file

src/main/resources/realtime.json

+{
+  "projectId":31,
+  "srcDataSourceId":205,
+  "targetDataSourceId":202,
+  "sourceName":"test34",
+  "sourceCustomArg":{
+    "decimal.handling.mode":"double"
+  },
+  "blacklistTables":"dmp_azkaban_exector_server_config,dmp_realtime_sync_handle_count,dmp_realtime_sync_info,dmp_realtime_sync_submit_result,table_operate_log",
+  "notSelectTables":"dmp_develop_function,dmp_develop_resource,dmp_develop_script,dmp_develop_task,dmp_develop_task_history,dmp_locks,dmp_module_operate_log,dmp_navigation_tree,dmp_open_api_es_fields,dmp_open_api_es_tagconfig,dmp_open_api_kafka_connector,dmp_ops_monitor,dmp_ops_monitor_setting,dmp_permission,dmp_project,dmp_project_member,dmp_project_member_role,dmp_project_openapi,dmp_project_orgid,dmp_project_permission,dmp_project_permission_bak,dmp_project_role,dmp_project_role_menu,dmp_project_role_permission,dmp_project_superuser_role,dmp_project_system_info,dmp_realtime_sync_blacklist_table_info,dmp_spark_applications,dmp_syncing_datasource,dmp_syncing_datasource_type,dmp_syncing_job_conf,dmp_syncing_job_reader,dmp_syncing_job_writer,dmp_system_lookup,dmp_system_user,dmp_table,dmp_table_access_auth,dmp_table_category,dmp_table_column,dmp_table_ddl_log,dmp_table_field_mapping,dmp_table_field_schema,dmp_table_manage,dmp_task_dependent,dmp_task_instance,dmp_task_instance_optlog,dmp_task_instance_runlog,dmp_task_schedule,dmp_user_member,dmp_work_flow_publish_details,dmp_work_flow_submit_details,dv_rule_check_result_t,dv_rule_t,dv_task_rule_t,oauth_token,oauth_token_key,salej2",
+  "connectorUrl":"connect1@http://172.18.104.130:9993/connectors",
+  "tables":[
+    {
+      "sourceTableName":"dmp_data_contrast",
+      "targetTableName":"dmp_data_contrast",
+      "toptic":"",
+      "desensitizationField":"target_database_name",
+      "arithmetic":"HmacSHA256",
+      "pkName":"source_database_name,source_table_name,stat_date"
+    }
+  ]
+}
+