增加数据源测试效果

cloud-etl-common/src/main/java/com/muyu/domain/Structure.java

@@ -0,0 +1,79 @@
+package com.muyu.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.muyu.common.core.annotation.Excel;
+import com.muyu.common.core.web.domain.BaseEntity;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+/**
+ * 结构对象 structure
+ *
+ * @author Saisai
+ * @date 2024-04-22
+ */
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@SuperBuilder
+public class Structure extends BaseEntity
+{
+    private static final long serialVersionUID = 1L;
+
+    /** 主键 */
+    @TableId(value = "id", type = IdType.AUTO)
+    private Long id;
+
+    /** 表id */
+    @Excel(name = "表id")
+    private Long tableId;
+
+    /** 字段名称 */
+    @Excel(name = "字段名称")
+    private String columnName;
+
+    /** 字段注释 */
+    @Excel(name = "字段注释")
+    private String columnRemark;
+
+    /** 是否主键 'Y'是主键 'N'不是主键 */
+    @Excel(name = "是否主键 'Y'是主键 'N'不是主键")
+    private String isPrimary;
+
+    /** 数据类型 */
+    @Excel(name = "数据类型")
+    private String columnType;
+
+    /** 映射类型 */
+    @Excel(name = "映射类型")
+    private String javaType;
+
+    /** 字段长度 */
+    @Excel(name = "字段长度")
+    private String columnLength;
+
+    /** 小数位数 */
+    @Excel(name = "小数位数")
+    private String columnDecimals;
+
+    /** 是否为空 'Y'是 'N'不是 */
+    @Excel(name = "是否为空 'Y'是 'N'不是")
+    private String isNull;
+
+    /** 默认值 */
+    @Excel(name = "默认值")
+    private String defaultValue;
+
+    /** 是否字典 'Y'是 'N'不是 */
+    @Excel(name = "是否字典 'Y'是 'N'不是")
+    private String isDictionary;
+
+    /** 映射字典 */
+    @Excel(name = "映射字典")
+    private String dictionaryTable;
+
+
+}

cloud-etl-common/src/main/java/com/muyu/domain/TableInfo.java

@@ -2,6 +2,7 @@ package com.muyu.domain;
 
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
 import com.muyu.common.core.annotation.Excel;
 import com.muyu.common.core.web.domain.TreeEntity;
 import lombok.AllArgsConstructor;
@@ -13,6 +14,7 @@ import lombok.experimental.SuperBuilder;
 @NoArgsConstructor
 @AllArgsConstructor
 @SuperBuilder
+@TableName(value ="table_info",autoResultMap = true) //数据库表相关
 public class TableInfo extends TreeEntity{
 
         private static final long serialVersionUID = 1L;

cloud-etl-server/src/main/java/com/muyu/cloud/etl/controller/SourceController.java

@@ -86,6 +86,13 @@ public class SourceController extends BaseController {
     }
 
 
+    //同步资产结构
+    @PostMapping("/structure")
+    public Result structure(@RequestBody Source source) throws ServletException {
+        return toAjax(sourceService.structure(source));
+    }
+
+
 
 
 

cloud-etl-server/src/main/java/com/muyu/cloud/etl/mapper/StructureMapper.java

@@ -0,0 +1,9 @@
+package com.muyu.cloud.etl.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.muyu.domain.Structure;
+import org.apache.ibatis.annotations.Mapper;
+
+@Mapper
+public interface StructureMapper extends BaseMapper<Structure> {
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/mapper/TableInfoMapper.java

@@ -0,0 +1,9 @@
+package com.muyu.cloud.etl.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.muyu.domain.TableInfo;
+import org.apache.ibatis.annotations.Mapper;
+
+@Mapper
+public interface TableInfoMapper extends BaseMapper<TableInfo> {
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/SourceService.java

@@ -18,4 +18,6 @@ public interface SourceService extends IService<Source> {
     int updataSource(Source sourceReq);
 
     int connectionTest(Source source);
+
+    int structure(Source source);
 }

cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/StructureService.java

@@ -0,0 +1,7 @@
+package com.muyu.cloud.etl.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.muyu.domain.Structure;
+
+public interface StructureService extends IService<Structure> {
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/TableInfoService.java

@@ -0,0 +1,9 @@
+package com.muyu.cloud.etl.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.muyu.domain.TableInfo;
+
+public interface TableInfoService extends IService<TableInfo> {
+
+    TableInfo selectTableInfoByName(TableInfo tableInfoInsert);
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/impl/SourceServiceImpl.java

@@ -1,21 +1,34 @@
 package com.muyu.cloud.etl.service.impl;
 
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import com.muyu.cloud.etl.mapper.SourceMapper;
 import com.muyu.cloud.etl.service.SourceService;
+import com.muyu.cloud.etl.service.StructureService;
+import com.muyu.cloud.etl.service.TableInfoService;
+import com.muyu.common.security.utils.SecurityUtils;
 import com.muyu.domain.Source;
+import com.muyu.domain.Structure;
+import com.muyu.domain.TableInfo;
 import com.muyu.domain.req.SourceReq;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.sql.*;
+import java.util.Date;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 @Service
 public class SourceServiceImpl extends ServiceImpl<SourceMapper, Source> implements SourceService {
     @Autowired
     private SourceMapper sourceMapper;
+    @Autowired
+    private TableInfoService tableInfoService;
+    @Autowired
+    private StructureService structureService;
 
     @Override
     public List<Source> selectSourceList(SourceReq sourceReq) {
@@ -44,6 +57,7 @@ public class SourceServiceImpl extends ServiceImpl<SourceMapper, Source> impleme
         return delete;
     }
 
+
     @Override
     public int updataSource(Source sourceReq) {
       Integer i=  sourceMapper.updataSource(sourceReq);
@@ -94,9 +108,171 @@ public class SourceServiceImpl extends ServiceImpl<SourceMapper, Source> impleme
             return 0;
         }
 
+        return  1;
+    }
 
+    @Override
+    public int structure(Source source) {
+
+        String host = source.getHost();
+        String port = source.getPort();
+        String databaseName = source.getDatabaseName();
+        String databaseType = source.getDatabaseType();
+        String url = "jdbc:" + databaseType + "://" + host + ":" + port + "/" + databaseName + "?" + source.getConnectionParams();
+        String user = source.getUsername();
+        String password = source.getPassword();
+        Connection conn = null;
+        try {
+            conn = DriverManager.getConnection(url, user, password);
+            System.out.println("Connected to the MySQL server successfully.");
+
+
+            TableInfo tableInfoInsert = TableInfo.builder()
+                    .basicId(source.getId())
+                    .parentId(0L)
+                    .tableRemark("")
+                    .center("Y")
+                    .type("dataSource")
+                    .tableName(source.getDataSourcesSystemName() + "(" + databaseName + ")")
+                    .createBy(SecurityUtils.getUsername())
+                    .createTime(new Date())
+                    .build();
+
+
+            tableInfoService.saveOrUpdate(tableInfoInsert, new LambdaUpdateWrapper<TableInfo>(TableInfo.class) {{
+                eq(TableInfo::getTableName, tableInfoInsert.getTableName());
+                eq(TableInfo::getBasicId, source.getId());
+            }});
+
+            TableInfo tableInfo = tableInfoService.selectTableInfoByName(tableInfoInsert);
+
+
+            DatabaseMetaData metaData = conn.getMetaData();
+
+            ResultSet rs = metaData.getTables(databaseName,
+                    null, "%", new String[]{"TABLE", "VIEW"});
+
+            while (rs.next()) {
+                String tableName = rs.getString("TABLE_NAME"); // 或者 rs.getString(3)，但使用列名更清晰
+                System.out.println("Table Name: " + tableName);
+                String tableRemark = rs.getString("REMARKS");
+                Connection finalConn = conn;
+                PreparedStatement ps = conn.prepareStatement("select * from " + tableName);
+                ResultSet rset = ps.executeQuery();
+                Long rowCount = 0L;
+                while (rset.next()) {
+                    rowCount++;
+                }
+
+                TableInfo build = TableInfo.builder()
+                        .basicId(source.getId())
+                        .tableName(tableName)
+                        //bug点，tableRemark为空，造成空指针异常
+                        .tableRemark(tableRemark == null ? "" : tableRemark)
+                        .parentId(tableInfo.getId())
+                        .type("dataTable")
+                        .center("Y")
+                        .updateBy(SecurityUtils.getUsername())
+                        .dataNum(rowCount)
+                        .updateTime(new Date())
+                        .build();
+
+                tableInfoService.saveOrUpdate(build, new LambdaUpdateWrapper<>(TableInfo.class) {{
+                    eq(TableInfo::getTableName, build.getTableName());
+                    eq(TableInfo::getBasicId, source.getId());
+                }});
+
+                TableInfo table = tableInfoService.selectTableInfoByName(build);
+
+                //线程池
+                ExecutorService threadPool = Executors.newCachedThreadPool();
+
+                threadPool.submit(() -> {
+                    try {
+                        syncData(finalConn, databaseName, table);
+                    } catch (SQLException e) {
+                        throw new RuntimeException(e);
+                    }
+                });
+            }
+
+        } catch (SQLException e) {
+            return 0;
+        }
 
         return  1;
     }
 
+
+
+    public void syncData(Connection conn, String databaseName, TableInfo table) throws SQLException {
+        ExecutorService threadPool = Executors.newCachedThreadPool();
+
+        PreparedStatement ps = conn.prepareStatement(
+                "    SELECT  " +
+                        "    COLUMN_NAME , " +
+                        "    COLUMN_COMMENT  ," +
+                        "    CASE WHEN COLUMN_KEY = 'PRI' THEN '是' ELSE '否' END ," +
+                        "    CASE \n" +
+                        "    WHEN DATA_TYPE = 'int' THEN 'Integer' " +
+                        "    WHEN DATA_TYPE = 'bigint' THEN 'Long' " +
+                        "    WHEN DATA_TYPE = 'varchar' THEN 'String' " +
+                        "    WHEN DATA_TYPE = 'decimal' THEN 'BigDecimal' " +
+                        "    WHEN DATA_TYPE = 'tinyint' AND COLUMN_TYPE = 'tinyint(1)' THEN 'Boolean'" +
+                        "    ELSE DATA_TYPE   \n" +
+                        "    END , " +
+                        "    DATA_TYPE ,   \n" +
+                        "    COLUMN_TYPE ,  \n" +
+                        "    CHARACTER_MAXIMUM_LENGTH ,  \n" +
+                        "    NUMERIC_SCALE ,  \n" +
+                        "    IS_NULLABLE ,  \n" +
+                        "    COLUMN_DEFAULT   \n" +
+                        "FROM  INFORMATION_SCHEMA.COLUMNS  WHERE  \n" +
+                        "TABLE_SCHEMA = '" + databaseName + "'   \n" +
+                        "AND TABLE_NAME = '" + table.getTableName() + "'");
+//        "SELECT COLUMN_NAME,DATA_TYPE,IS_NULLABLE,COLUMN_KEY,COLUMN_DEFAULT,
+//        COLUMN_COMMENT,CHARACTER_MAXIMUM_LENGTH,NUMERIC_PRECISION,
+//        NUMERIC_SCALE FROM INFORMATION_SCHEMA.COLUMNS
+//        WHERE TABLE_SCHEMA = '数据库名' AND TABLE_NAME = '表名'"
+        ResultSet resultSet = ps.executeQuery();
+        while (resultSet.next()) {
+            String columnName = String.valueOf(resultSet.getString(1));
+            String columnComment = String.valueOf(resultSet.getObject(2));
+            String columnKey = String.valueOf(resultSet.getObject(3));
+            String end = String.valueOf(resultSet.getObject(4));
+            String dataType = String.valueOf(resultSet.getObject(5));
+            String columnType = String.valueOf(resultSet.getObject(6));
+            String characterMaximumLength = String.valueOf(resultSet.getInt(7));
+            String NumericScale = String.valueOf(resultSet.getInt(8));
+            String isNullable = String.valueOf(resultSet.getObject(9));
+            String columnDefault = String.valueOf(resultSet.getObject(10));
+            Structure build = Structure.builder()
+                    .tableId(table.getId())
+                    .columnName(String.valueOf(columnName))
+                    .columnRemark(columnComment)
+                    .isPrimary("是".equals(columnKey) ? "Y" : "N")
+                    .javaType(end)
+                    .columnType(dataType)
+                    .columnType(columnType)
+                    .columnLength(characterMaximumLength)
+                    .columnDecimals(NumericScale)
+                    .isNull("YES".equals(isNullable) ? "Y" : "N")
+                    .defaultValue(columnDefault)
+                    .build();
+            threadPool.submit(() -> {
+                structureService.saveOrUpdate(build, new LambdaUpdateWrapper<Structure>() {{
+                    eq(Structure::getTableId, build.getTableId());
+                    eq(Structure::getColumnName, build.getColumnName());
+                    eq(Structure::getColumnRemark, build.getColumnRemark());
+                }});
+            });
+        }
+        threadPool.shutdown();
+        ps.close();
+    }
+
+
+
+
+
 }

cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/impl/StructureServiceImpl.java

@@ -0,0 +1,12 @@
+package com.muyu.cloud.etl.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.muyu.cloud.etl.mapper.StructureMapper;
+import com.muyu.cloud.etl.service.StructureService;
+import com.muyu.domain.Structure;
+import org.springframework.stereotype.Service;
+
+@Service
+public class StructureServiceImpl extends ServiceImpl<StructureMapper, Structure> implements StructureService {
+
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/service/impl/TableServiceImpl.java

@@ -0,0 +1,26 @@
+package com.muyu.cloud.etl.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.muyu.cloud.etl.mapper.TableInfoMapper;
+import com.muyu.cloud.etl.service.TableInfoService;
+import com.muyu.domain.TableInfo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Service
+public class TableServiceImpl extends ServiceImpl<TableInfoMapper, TableInfo> implements TableInfoService {
+    @Autowired
+    private TableInfoMapper tableInfoMapper;
+
+    @Override
+    public TableInfo selectTableInfoByName(TableInfo tableInfoInsert) {
+        LambdaQueryWrapper<TableInfo> tableInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
+//        tableInfoLambdaQueryWrapper.eq(TableInfo::getId, tableInfoInsert.getId());
+        //
+        tableInfoLambdaQueryWrapper.eq(TableInfo::getBasicId, tableInfoInsert.getBasicId());
+        tableInfoLambdaQueryWrapper.eq(TableInfo::getTableName, tableInfoInsert.getTableName()).or().eq(TableInfo::getParentId, tableInfoInsert.getParentId());
+
+        return  this.tableInfoMapper.selectOne(tableInfoLambdaQueryWrapper);
+    }
+}