2112A-ONE
/
muyu-source
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

0910:修改整体代码

master
冷调 2024-09-10 15:06:01 +08:00
parent 82b00d2b13
commit 25569131f5
48 changed files with 1714 additions and 563 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
muyu-source-client/src/main/java/com/muyu/config/SourceClientConfig.java
View File

@ -1,79 +0,0 @@
package com.muyu.config;
import com.muyu.common.core.domain.Result;
import com.muyu.connection.DataSourceConfig;
import com.muyu.source.domain.Children;
import com.muyu.source.domain.DataSource;
import com.muyu.source.domain.model.DataModel;
import com.muyu.source.remote.RemoteChildrenService;
import com.muyu.source.remote.RemoteDataSourceService;
import jakarta.annotation.Resource;
import org.springframework.context.annotation.ComponentScan;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-22-14:33
* @ Version：1.0
* @ Description：数据接入客户端配置类
*/
@ComponentScan
public class SourceClientConfig {
@Resource
private RemoteDataSourceService remoteDataSourceService;
@Resource
private RemoteChildrenService remoteChildrenService;
public List<List<DataModel>> getDataModel(Long id) {
List<List<DataModel>> list = new ArrayList<>();
//根据ID查询出表结构的数据
Result<Children> childrenResult = remoteChildrenService.getChildren(id);
Children children = childrenResult.getData();
//根据ID查询出数据源的数据
Result<DataSource> dataSourceResult = remoteDataSourceService.getDataSource(children.getAssetId());
DataSource dataSource = dataSourceResult.getData();
String sql = "";
if ("MySql".equals(dataSource.getDataType())) {
sql = "SELECT * FROM " + children.getName();
}
try {
//调用连接池获取连接
Connection connection = DataSourceConfig.getDataSource(dataSource.getId());
PreparedStatement preparedStatement = connection.prepareStatement(sql);
ResultSet resultSet = preparedStatement.executeQuery();
ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
while (resultSet.next()) {
List<DataModel> dataModels = new ArrayList<>();
for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
//获取列名 字段名 key
String columnName = resultSetMetaData.getColumnName(i);
//获取字段的值 value
Object value = resultSet.getObject(columnName);
//获取列的类型 字段的类型 type
String columnTypeName = resultSetMetaData.getColumnTypeName(i);
//获取列的标题 label
String columnLabel = resultSetMetaData.getColumnLabel(i);
// java类型
String columnClassName = resultSetMetaData.getColumnClassName(i);
DataModel dataModel = DataModel.builder()
.key(columnName)
.label(columnLabel)
.type(columnTypeName)
.value(value)
.build();
dataModels.add(dataModel);
}
list.add(dataModels);
}
//归还连接
DataSourceConfig.returnConnection(connection);
} catch (SQLException e) {
throw new RuntimeException(e);
}
return list;
}
}

45
muyu-source-client/src/main/java/com/muyu/config/SourceClientRunner.java
View File

@ -1,45 +0,0 @@
package com.muyu.config;
import com.muyu.common.core.domain.Result;
import com.muyu.connection.DataSourceConfig;
import com.muyu.source.domain.DataSource;
import com.muyu.source.domain.DataType;
import com.muyu.source.remote.RemoteChildrenService;
import com.muyu.source.remote.RemoteDataSourceService;
import com.muyu.source.remote.RemoteDataTypeService;
import jakarta.annotation.Resource;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Configuration;
import java.util.List;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-23-9:45
* @ Version：1.0
* @ Description：初始化加载
*/
@Log4j2
@Configuration
public class SourceClientRunner implements ApplicationRunner {
@Resource
private RemoteDataSourceService remoteDataSourceService;
@Resource
private RemoteDataTypeService remoteDataTypeService;
@Override
public void run(ApplicationArguments args) throws Exception {
Result<List<DataSource>> sourceResult = remoteDataSourceService.getDataSourceList();
log.info(sourceResult);
List<DataSource> sourceList = sourceResult.getData();
sourceList.stream().forEach(source -> {
Result<DataType> dataTypeResult = remoteDataTypeService.getDataType(source.getDataType());
DataType dataType = dataTypeResult.getData();
DataSourceConfig.init(source, dataType);
});
}
}

72
muyu-source-client/src/main/java/com/muyu/connection/DataSourceConfig.java
View File

@ -1,72 +0,0 @@
package com.muyu.connection;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.muyu.source.domain.DataSource;
import com.muyu.source.domain.DataType;
import com.alibaba.druid.pool.DruidDataSource;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Component;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.HashMap;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-22-14:36
* @ Version：1.0
* @ Description：数据源连接配置类
*/
@Log4j2
@Component
public class DataSourceConfig {
//数据源连接池
private static HashMap<Long, DruidDataSource> dataSourceMap = new HashMap<>();
//数据源连接池
public static void init(DataSource source, DataType dataType) {
//创建连接池
DruidDataSource druidDataSource = new DruidDataSource();
//用户名
druidDataSource.setUsername(source.getUserName());
//密码
druidDataSource.setPassword(source.getPassword());
//数据库连接
druidDataSource.setUrl(dataType.getPrefix() + source.getIp() + ":" + source.getPort() + "/" + source.getDatabaseName() + "?" + source.getConnectionParam());
//驱动
druidDataSource.setDriverClassName(dataType.getDriverManager());
//最小连接数
druidDataSource.setMinIdle(Math.toIntExact(source.getMaxWaitSize()));
//最大连接数
druidDataSource.setMaxActive(Math.toIntExact(source.getMaxNum()));
//初始连接数
druidDataSource.setInitialSize(Math.toIntExact(source.getInitTotal()));
try {
druidDataSource.init();
} catch (SQLException e) {
throw new RuntimeException(e);
}
dataSourceMap.put(source.getId(), druidDataSource);
}
//获取连接
public static Connection getDataSource(Long id) {
DruidDataSource druidDataSource = dataSourceMap.get(id);
try {
DruidPooledConnection connection = druidDataSource.getConnection();
return connection;
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
//归还连接
public static void returnConnection(Connection connection) {
try {
connection.close();
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
}

36
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/DataSourcePool.java 100644
View File

@ -0,0 +1,36 @@
package com.muyu.etl.data.access.data.access.client;
import com.muyu.etl.data.access.data.access.client.config.DatabaseConnectionPool;
import com.zaxxer.hikari.HikariDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.sql.Connection;
import java.sql.SQLException;
/**
* @Author: DongZeLiang
* @date: 2024/9/9
* @Description: 数据源池
* @Version: 1.0
*/
@Component
public class DataSourcePool {
@Autowired
private DatabaseConnectionPool databaseConnectionPool;
public HikariDataSource getDataSource(Long dataSourceId) {
return databaseConnectionPool.get(dataSourceId);
}
public Connection getConnection(HikariDataSource hikariDataSource) {
return databaseConnectionPool.getConn(hikariDataSource);
}
public void returnConnection(Connection connection) {
databaseConnectionPool.returnConnection(connection);
}
}

25
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/basic/BaseDataAbsSource.java 100644
View File

@ -0,0 +1,25 @@
package com.muyu.etl.data.access.data.access.client.basic;
import com.muyu.etl.data.access.data.access.client.DataSourcePool;
import org.springframework.beans.factory.annotation.Autowired;
/**
* @Author: DongZeLiang
* @date: 2024/8/28
* @Description: 数据接入抽象类
* @Version: 1.0
*/
public abstract class BaseDataAbsSource implements DataAccessBasic {
@Autowired
private DataSourcePool dataSourcePool;
public void setQuery(QueryBasic baseQuery){
QueryBasicHandler.set(baseQuery);
}
public <T> T getQuery(){
return QueryBasicHandler.get();
}
}

31
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/basic/DataAccessBasic.java 100644
View File

@ -0,0 +1,31 @@
package com.muyu.etl.data.access.data.access.client.basic;
import com.muyu.source.core.DataValue;
/**
* @Author: DongZeLiang
* @date: 2024/9/9
* @Description: 数据接入标准
* @Version: 1.0
*/
public interface DataAccessBasic extends TaskBasic{
/**
* 数据值
* @return 返回查询的数据值
*/
public DataValue getDataStructure();
/**
* 数据记录
* @return 数据记录
*/
public DataValue[] getRow();
/**
* 数据集
* @return 数据集
*/
public DataValue[][] getRows();
}

19
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/basic/DataSourceConfig.java 100644
View File

@ -0,0 +1,19 @@
package com.muyu.etl.data.access.data.access.client.basic;
import com.muyu.source.domain.resp.DataSourceResp;
import java.util.List;
/**
* @Author: DongZeLiang
* @date: 2024/9/9
* @Description: 数据源配置接口
* @Version: 1.0
*/
public interface DataSourceConfig {
public List<DataSourceResp> getIsUseDataSourceList();
}

21
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/basic/QueryBasic.java 100644
View File

@ -0,0 +1,21 @@
package com.muyu.etl.data.access.data.access.client.basic;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
/**
* @Author: DongZeLiang
* @date: 2024/8/28
* @Description: 基础查询
* @Version: 1.0
*/
@Data
@SuperBuilder
@NoArgsConstructor
@AllArgsConstructor
public class QueryBasic {
private Long dataSourceId;
}

25
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/basic/QueryBasicHandler.java 100644
View File

@ -0,0 +1,25 @@
package com.muyu.etl.data.access.data.access.client.basic;
/**
* @Author: DongZeLiang
* @date: 2024/8/28
* @Description: 基础查询
* @Version: 1.0
*/
public class QueryBasicHandler {
private static final ThreadLocal<QueryBasic> BASE_QUERY_THREAD_LOCAL = new ThreadLocal<>();
public static void set(QueryBasic queryBasic){
BASE_QUERY_THREAD_LOCAL.set(queryBasic);
}
public static <T> T get(){
return (T) BASE_QUERY_THREAD_LOCAL.get();
}
public static void remove(){
BASE_QUERY_THREAD_LOCAL.remove();
}
}

19
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/basic/TaskBasic.java 100644
View File

@ -0,0 +1,19 @@
package com.muyu.etl.data.access.data.access.client.basic;
import com.muyu.etl.data.access.data.access.client.scope.TaskScopeBasic;
import com.muyu.etl.data.access.data.access.client.scope.TaskScopeConfig;
;
/**
* @Author: DongZeLiang
* @date: 2024/9/9
* @Description: 任務基準
* @Version: 1.0
*/
public interface TaskBasic {
public default TaskScopeConfig getScopeConfig(){
return TaskScopeBasic.get();
}
}

26
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/basic/impl/DataSourceConfigImpl.java 100644
View File

@ -0,0 +1,26 @@
package com.muyu.etl.data.access.data.access.client.basic.impl;
import com.muyu.etl.data.access.data.access.client.basic.DataSourceConfig;
import com.muyu.source.domain.resp.DataSourceResp;
import com.muyu.source.remote.remote.RemoteDataSourceService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
/**
* @Author: DongZeLiang
* @date: 2024/9/9
* @Description:
* @Version: 1.0
*/
@Service
public class DataSourceConfigImpl implements DataSourceConfig {
@Autowired
private RemoteDataSourceService remoteDataSourceService;
@Override
public List<DataSourceResp> getIsUseDataSourceList () {
return remoteDataSourceService.getIsUseDataSourceList().getData();
}
}

25
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/config/DataBaseInitPool.java 100644
View File

@ -0,0 +1,25 @@
package com.muyu.etl.data.access.data.access.client.config;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @Author WangXin
* @Data 2024/9/8
* @Description 初始化线程池
* @Version 1.0.0
*/
@Component
public class DataBaseInitPool implements ApplicationRunner {
@Resource
private DatabaseConnectionPool databaseConnectionPool;
@Override
public void run(ApplicationArguments args) throws Exception {
databaseConnectionPool.init();
}
}

103
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/config/DatabaseConnectionPool.java 100644
View File

@ -0,0 +1,103 @@
package com.muyu.etl.data.access.data.access.client.config;
import com.muyu.etl.data.access.data.access.client.basic.DataSourceConfig;
import com.muyu.source.domain.resp.DataSourceResp;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
/**
* @Author WangXin
* @Data 2024/9/6
* @Description 数据连接池
* @Version 1.0.0
*/
@Log4j2
@Component
public class DatabaseConnectionPool {
private final String DATA_SOURCE_MAP = "DATA_SOURCE_MAP";
@Autowired
private DataSourceConfig dataSourceConfig;
private HashMap<Long, HikariDataSource> map = new HashMap<>();
/**
* 初始化连接池
*
* @param dataSourceResp 连接池数据
*/
public void initialize (DataSourceResp dataSourceResp) {
HikariConfig config = new HikariConfig();
config.setDriverClassName(dataSourceResp.getDriverName());
config.setJdbcUrl("jdbc:mysql://" + dataSourceResp.getIp() + ":" + dataSourceResp.getPort() + "/" + dataSourceResp.getDatabaseName() + "?" + dataSourceResp.getConnectionParam());
config.setUsername(dataSourceResp.getUserName());
config.setPassword(dataSourceResp.getPassword());
config.setMaximumPoolSize(dataSourceResp.getMaxWaitSize());
config.addDataSourceProperty("cachePrepStmts", "true");
map.putIfAbsent(dataSourceResp.getId(), new HikariDataSource(config));
}
/**
* 初始化数据库
*/
public void init () {
log.info("初始化连接池开始……………………");
long startTime = System.currentTimeMillis();
List<DataSourceResp> dataSourceRespList = dataSourceConfig.getIsUseDataSourceList();
log.info("查询所有可用连接池配置：{}个，连接池名称：{}", dataSourceRespList.size(), dataSourceRespList.stream().map(DataSourceResp::getDatabaseName).toList());
for (DataSourceResp dataSourceResp : dataSourceRespList) {
log.info("开始初始化：[{}]连接", dataSourceResp.getDatabaseName());
initialize(dataSourceResp);
log.info("完成初始化：[{}]连接", dataSourceResp.getDatabaseName());
}
log.info("初始化连接池结束共计耗时:{}MS……………………", System.currentTimeMillis() - startTime);
}
/**
* 获取数据连接对象
*
* @param id 数据库id
*
* @return HikariDataSource数据库连接所需对象
*/
public HikariDataSource get (Long id) {
if (map.get(id) != null) {
return map.get(id);
}
throw new RuntimeException("请重新初始化数据源!!!");
}
/**
* 归还连接
*
* @param connection 连接对象
*/
public void returnConnection (Connection connection) {
try {
if (connection != null) {
connection.close();
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
public Connection getConn (HikariDataSource hikariDataSource) {
Connection conn = null;
try {
conn = hikariDataSource.getConnection();
} catch (SQLException e) {
throw new RuntimeException(e);
}
return conn;
}
}

185
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/mysql/MySqlDataSource.java 100644
View File

@ -0,0 +1,185 @@
package com.muyu.etl.data.access.data.access.client.mysql;
import com.muyu.common.core.domain.Result;
import com.muyu.etl.data.access.data.access.client.DataSourcePool;
import com.muyu.etl.data.access.data.access.client.basic.BaseDataAbsSource;
import com.muyu.source.core.DataType;
import com.muyu.source.core.DataValue;
import com.zaxxer.hikari.HikariDataSource;
import lombok.extern.log4j.Log4j2;
import org.springframework.stereotype.Service;
import java.sql.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* @Author: DongZeLiang
* @date: 2024/8/28
* @Description: mysql数据源
* @Version: 1.0
*/
@Log4j2
@Service("mysql-data-source")
public class MySqlDataSource extends BaseDataAbsSource {
private final DataSourcePool dataSourcePool;
public MySqlDataSource (DataSourcePool dataSourcePool) {
super();
this.dataSourcePool = dataSourcePool;
}
/**
* 获取查询条数
* @return
*/
public Result getCount(){
MySqlQuery query = getQuery();
HikariDataSource hikariDataSource = dataSourcePool.getDataSource(query.getDataSourceId());
try(Connection conn = hikariDataSource.getConnection();
Statement statement = conn.createStatement();) {
ResultSet resultSet = statement.executeQuery(query.getSql());
if (resultSet.next()) {
return Result.success(resultSet.getInt(1));
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
return Result.success(0);
}
/**
* 添加数据库
* @return
*/
public Result addTargetDatabase(){
MySqlQuery query = getQuery();
HikariDataSource hikariDataSource = dataSourcePool.getDataSource(query.getDataSourceId());
try(Connection conn = hikariDataSource.getConnection();
PreparedStatement preparedStatement = conn.prepareStatement(query.getSql());) {
return Result.success(preparedStatement.executeUpdate());
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
/**
* 获取数据值
* @return
*/
@Override
public DataValue getDataStructure () {
MySqlQuery query = getQuery();
Connection connection = null;
String sql = query.getSql();
Map<String, Object> queryParams = query.getParams();
log.info(sql);
log.info(queryParams);
try {
PreparedStatement preparedStatement = connection.prepareStatement(sql);
ResultSet resultSet = preparedStatement.getResultSet();
if(resultSet.next()){
DataValue.builder()
.key(resultSet.getCursorName())
.label("")
.value(resultSet.getObject(resultSet.getCursorName(), String.class))
.type(DataType.STRING);
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
return null;
}
/**
* 获取数据记录
* @return
*/
@Override
public DataValue[] getRow () {
return new DataValue[0];
}
/**
* 获取数据集
* @return
*/
@Override
public DataValue[][] getRows () {
ConcurrentHashMap<Integer, DataValue> map = new ConcurrentHashMap<>();
MySqlQuery query = getQuery();
//初始化一个列表,用于存储数据值对象
DataValue[][] rows = null;
HikariDataSource dataSource = dataSourcePool.getDataSource(query.getDataSourceId());
Connection conn = dataSourcePool.getConnection(dataSource);
try {
Statement statement = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_READ_ONLY);
//准备SQL查询
ResultSet resultSet = statement.executeQuery(query.getSql());
int row = 0;
// 将游标移动到最后行
if (resultSet.last()) {
// 获取行数
row = resultSet.getRow();
log.info("需要处理数据--->{}条",row);
// 如果需要重置游标以便后续处理
resultSet.beforeFirst();
} else {
System.out.println("没有数据");
}
//获取元数据
ResultSetMetaData metaData = resultSet.getMetaData();
//获取列的数量
int columnCount = metaData.getColumnCount();
rows = new DataValue[row][columnCount];
//遍历每一行数据
while (resultSet.next()) {
// 提交任务给线程池
for (int i = 1; i <= columnCount; i++) {
if (resultSet.isFirst()){
String columnTypeName = metaData.getColumnTypeName(i);
DatabaseMetaData metaDataTwo = conn.getMetaData();
ResultSet columns = metaDataTwo.getColumns(null, null, metaData.getTableName(i), metaData.getColumnName(i));
String remarks = null;
while (columns.next()) {
remarks = columns.getString("REMARKS");
}
DataValue dataStructure = DataValue.builder()
.key(metaData.getColumnName(i))
.label(remarks)
.value(resultSet.getObject(i, DataType.getTargetClass(columnTypeName)))
.type(DataType.getDataType(columnTypeName))
.build();
int currentRow = resultSet.getRow() - 1; // 因为行索引从1开始
rows[currentRow][i - 1] = dataStructure;
map.put(i, dataStructure);
}else {
DataValue dataStructure = DataValue.builder()
.key(metaData.getColumnName(i))
.label(map.get(i).getLabel())
.value(resultSet.getObject(i, map.get(i).getType().getTargetType()))
.type(map.get(i).getType())
.build();
int currentRow = resultSet.getRow() - 1; // 因为行索引从1开始
rows[currentRow][i - 1] = dataStructure;
}
}
}
}catch (Exception e) {
throw new RuntimeException(e);
}finally {
dataSource.evictConnection(conn);
}
log.info("查询结果数量为：rows == > {}", rows.length);
return rows;
}
}

28
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/mysql/MySqlQuery.java 100644
View File

@ -0,0 +1,28 @@
package com.muyu.etl.data.access.data.access.client.mysql;
import com.muyu.etl.data.access.data.access.client.basic.QueryBasic;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import java.util.Map;
/**
* @Author: DongZeLiang
* @date: 2024/8/28
* @Description: mysql查询
* @Version: 1.0
*/
@EqualsAndHashCode(callSuper = true)
@Data
@SuperBuilder
@NoArgsConstructor
@AllArgsConstructor
public class MySqlQuery extends QueryBasic {
private String sql;
private Map<String, Object> params;
}

27
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/scope/TaskScopeBasic.java 100644
View File

@ -0,0 +1,27 @@
package com.muyu.etl.data.access.data.access.client.scope;
import org.apache.poi.ss.formula.functions.T;
/**
* @Author: DongZeLiang
* @date: 2024/9/9
* @Description: 作用域标准
* @Version: 1.0
*/
public class TaskScopeBasic {
private static final ThreadLocal<TaskScopeConfig> localScope = new ThreadLocal<>();
public static void set(final TaskScopeConfig handler) {
localScope.set(handler);
}
public static TaskScopeConfig get() {
return localScope.get();
}
public static void remove(){
localScope.remove();
}
}

42
muyu-source-client/src/main/java/com/muyu/etl/data/access/data/access/client/scope/TaskScopeConfig.java 100644
View File

@ -0,0 +1,42 @@
package com.muyu.etl.data.access.data.access.client.scope;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.concurrent.LinkedBlockingDeque;
/**
* @Author: DongZeLiang
* @date: 2024/9/9
* @Description: 作用域存儲配置對象
* @Version: 1.0
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class TaskScopeConfig {
/**
* 任務ID
*/
private String taskId;
private LinkedBlockingDeque taskNodeQueue = new LinkedBlockingDeque();
public void addTaskNode(Object obj){
this.taskNodeQueue.add(obj);
}
public boolean hashTaskNodeNext(){
return !taskNodeQueue.isEmpty();
}
private <T> T nextTaskNode(){
return (T) taskNodeQueue.poll();
}
}

5
muyu-source-common/pom.xml
View File

@ -28,6 +28,11 @@
<artifactId>mybatis-plus-join-boot-starter</artifactId>
<version>1.4.11</version>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>4.0.3</version>
</dependency>
</dependencies>
</project>

77
muyu-source-common/src/main/java/com/muyu/source/access/data/redis/DataType.java 100644
View File

@ -0,0 +1,77 @@
//package com.muyu.source.access.data.redis;
//
//import java.math.BigDecimal;
//import java.util.Date;
//
///**
// * @author Lenovo
// * @ Tool：IntelliJ IDEA
// * @ Author：CHX
// * @ Date：2024-08-30-10:03
// * @ Version：1.0
// * @ Description：数据类型枚举
// */
//
//public enum DataType {
// VARCHAR("varchar",String.class,"String"),
// BIGINT("bigint", Long.class,"Long"),
// INT("int", Integer.class,"Integer"),
// DECIMAL("decimal", BigDecimal.class,"BigDecimal"),
// DATETIME("datetime", Date.class,"Date"),
// TEXT("text", String.class,"String"),
// DOUBLE("double", Double.class,"Double");
//
// // 数据库源类型
// private final String sourceType;
// // 映射到的Java类类型
// private final Class<?> targetType;
// // Java类型的字符串表示
// private final String javaType;
//
// public String getSourceType() {
// return sourceType;
// }
//
// public Class<?> getTargetType() {
// return targetType;
// }
//
// public String getJavaType() {
// return javaType;
// }
//
//
//
// public static Class convertType(String type){
// for (DataType dataType : DataType.values()) {
// if (dataType.sourceType.equalsIgnoreCase(type)){
// return dataType.targetType;
// }
// }
// return String.class;
// }
//
// public static DataType findBySqlType(String sqlType){
// for (DataType dataType : DataType.values()) {
// if (dataType.getSourceType().equalsIgnoreCase(sqlType)){
// return dataType;
// }
// }
// return VARCHAR;
// }
//
// public static String convertTypeString(String type){
// for (DataType dataType : DataType.values()) {
// if (dataType.sourceType.equalsIgnoreCase(type)){
// return dataType.javaType;
// }
// }
// return "String";
// }
//
// DataType(String sourceType, Class<?> targetType, String javaType) {
// this.sourceType = sourceType;
// this.targetType = targetType;
// this.javaType = javaType;
// }
//}

182
muyu-source-common/src/main/java/com/muyu/source/core/DataType.java
View File

@ -1,78 +1,162 @@
package com.muyu.source.core;
import com.muyu.common.core.utils.StringUtils;
import lombok.extern.log4j.Log4j2;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.Date;
import java.util.Map;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-30-10:03
* @ Version：1.0
* @ Description：数据类型枚举
* @Data 2024/8/28
* @Description 数据类型枚举
* @Version 1.0.0
*/
@Log4j2
public enum DataType {
VARCHAR("varchar",String.class,"String"),
BIGINT("bigint", Long.class,"Long"),
INT("int", Integer.class,"Integer"),
DECIMAL("decimal", BigDecimal.class,"BigDecimal"),
DATETIME("datetime", Date.class,"Date"),
TEXT("text", String.class,"String"),
DOUBLE("double", Double.class,"Double");
// 数据库源类型
private final String sourceType;
// 映射到的Java类类型
private final Class<?> targetType;
// Java类型的字符串表示
private final String javaType;
STRING(
new String[]{"char", "varchar", "text", "mediumtext", "longtext", "longblob"},
String.class,
"String"
),
INTEGER(
new String[]{"int", "tinyint"},
Integer.class,
"Integer"
),
DATE(
new String[]{"date", "datetime","timestamp"},
Date.class,
"Date"
),
BIG_DECIMAL(
new String[]{"decimal", "double", "float"},
BigDecimal.class,
"BigDecimal"
),
LONG(
new String[]{"bigint"},
Long.class,
"Long"
);
public String getSourceType() {
private String[] sourceType;
private Class<?> targetType;
private String javaType;
public String[] getSourceType() {
return sourceType;
}
public void setSourceType(String[] sourceType) {
this.sourceType = sourceType;
}
public Class<?> getTargetType() {
return targetType;
}
public void setTargetType(Class<?> targetType) {
this.targetType = targetType;
}
public String getJavaType() {
return javaType;
}
public static Class convertType(String type){
for (DataType dataType : DataType.values()) {
if (dataType.sourceType.equalsIgnoreCase(type)){
return dataType.targetType;
}
}
return String.class;
public void setJavaType(String javaType) {
this.javaType = javaType;
}
public static DataType findBySqlType(String sqlType){
for (DataType dataType : DataType.values()) {
if (dataType.getSourceType().equalsIgnoreCase(sqlType)){
return dataType;
}
}
return VARCHAR;
}
public static String convertTypeString(String type){
for (DataType dataType : DataType.values()) {
if (dataType.sourceType.equalsIgnoreCase(type)){
return dataType.javaType;
}
}
return "String";
}
DataType(String sourceType, Class<?> targetType, String javaType) {
DataType(String[] sourceType, Class<?> targetType, String javaType) {
this.sourceType = sourceType;
this.targetType = targetType;
this.javaType = javaType;
}
/**
* 获取枚举的Class类型
* @param sqlType sql数据类型
* @return class类
*/
public static Class<?> getTargetClass(String sqlType) {
Class<?> targetClass = null;
for (DataType dataType : values()) {
if (sqlType != null && Arrays.asList(dataType.getSourceType()).contains(sqlType.toLowerCase())) {
targetClass = dataType.getTargetType();
break;
}
}
if (targetClass == null) {
log.info("SQL字段类型异常,sqlType ---> {}", sqlType);
throw new RuntimeException(String.format("SQL字段类型异常,sqlType ---> {}", sqlType));
}
return targetClass;
}
/**
* 获取枚举的Class类型
* @param javaType sql数据类型
* @return class类
*/
public static Class<?> getTargetClassFindByJavaType(String javaType) {
Class<?> targetClass = null;
for (DataType dataType : values()) {
if (javaType != null && Arrays.asList(dataType.getJavaType()).contains(javaType.toLowerCase())) {
targetClass = dataType.getTargetType();
break;
}
}
if (targetClass == null) {
log.info("SQL字段类型异常,sqlType ---> {}", javaType);
throw new RuntimeException(String.format("SQL字段类型异常,sqlType ---> {}", javaType));
}
return targetClass;
}
/**
* 获取枚举的java类型
* @param sqlType sql数据类型
* @return java类型
*/
public static String getJavaType(String sqlType) {
String javaType = null;
for (DataType dataType : values()) {
if (sqlType != null && Arrays.asList(dataType.getSourceType()).contains(sqlType.toLowerCase())) {
javaType = dataType.getJavaType();
break;
}
}
if (StringUtils.isEmpty(javaType)) {
log.info("SQL字段类型异常,sqlType ---> {}", sqlType);
throw new RuntimeException(String.format("SQL字段类型异常,sqlType ---> {}", sqlType));
}
return javaType;
}
/**
* 获取枚举类型
* @param sqlType sql数据类型
* @return 枚举类型
*/
public static DataType getDataType(String sqlType) {
DataType dataType = null;
for (DataType value : values()) {
if (sqlType != null && Arrays.asList(value.getSourceType()).contains(sqlType.toLowerCase())) {
dataType = value;
}
}
if (dataType == null) {
log.info("SQL字段类型异常,sqlType ---> {}", sqlType);
throw new RuntimeException(String.format("SQL字段类型异常,sqlType ---> {}", sqlType));
}
return dataType;
}
}

171
muyu-source-common/src/main/java/com/muyu/source/domain/resp/DataSourceResp.java 100644
View File

@ -0,0 +1,171 @@
package com.muyu.source.domain.resp;
import com.muyu.common.core.annotation.Excel;
import com.muyu.source.domain.DataSource;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/8/19 19:12
* @注释
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@SuperBuilder
@Tag(name = "数据源输出信息",description = "查询数据源信息需要输出的字段")
public class DataSourceResp {
/**
* 主键
*/
@Excel(name = "主键")
private Long id;
/**
* 接入源名称
*/
@Schema(type = "String",defaultValue = "接入源名称",description = "接入源名称")
private String name;
/**
* 数据来源系统名称
*/
@Schema(type = "String",defaultValue = "数据来源系统",description = "数据来源系统")
private String systemName;
/**
* 数据库类型
*/
@Schema(type = "String",defaultValue = "mysql等",description = "要选择的数据库类型")
private String dataType;
/**
* ip地址
*/
@Schema(type = "String",defaultValue = "ip地址",description = "ip地址,要连接的mysql的ip")
private String ip;
/**
* 端口号
*/
@Schema(type = "String",defaultValue = "3306",description = "端口号")
private String port;
/**
* 数据库名称
*/
@Schema(type = "String",defaultValue = "t_user",description = "数据库名称")
private String databaseName;
/**
* 用户名
*/
@Schema(type = "String",defaultValue = "root",description = "用户名")
private String userName;
/**
* 密码
*/
@Schema(type = "String",defaultValue = "root",description = "密码")
private String password;
/**
* 数据连接参数
*/
@Schema(type = "String",defaultValue = "useUnicode=true&characterEncoding=utf8",description = "数据连接参数")
private String connectionParam;
/**
* 是否初始化
*/
@Schema(type = "String",defaultValue = "Y",description = "是否初始化 状态 'Y'/'N'")
private String isInit;
/**
* 初始连接数量
*/
@Schema(type = "Long",defaultValue = "5",description = "初始化连接数量")
private Integer initTotal;
/**
* 最大连接数量
*/
@Schema(type = "Long",defaultValue = "10",description = "最大连接数量")
private Integer maxNum;
/**
* 最大等待时间
*/
@Schema(type = "Long",defaultValue = "10",description = "最大等待时间")
private Integer maxWaitTime;
/**
* 最大等待次数
*/
@Schema(type = "Long",defaultValue = "3",description = "最大等待次数")
private Integer maxWaitSize;
/**
* 驱动 com.mysql.cj.jdbc.Driver
* */
@Excel(name = "驱动",defaultValue = "com.mysql.cj.jdbc.Driver")
private String driverName;
public static DataSourceResp dataSourceRespBuild(DataSource dataSource){
return DataSourceResp.builder()
.dataType(dataSource.getDataType())
.name(dataSource.getName())
.ip(dataSource.getIp())
.port(dataSource.getPort())
.databaseName(dataSource.getDatabaseName())
.userName(dataSource.getUserName())
.password(dataSource.getPassword())
.connectionParam(dataSource.getConnectionParam())
.isInit(dataSource.getIsInit())
.initTotal(dataSource.getInitTotal())
.maxNum(dataSource.getMaxNum())
.maxWaitSize(dataSource.getMaxWaitSize())
.maxWaitTime(dataSource.getMaxWaitTime())
.driverName("com.mysql.cj.jdbc.Driver")
.build();
}
}

25
muyu-source-remote/src/main/java/com/muyu/source/remote/RemoteChildrenService.java
View File

@ -1,25 +0,0 @@
package com.muyu.source.remote;
import com.muyu.common.core.constant.ServiceNameConstants;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.Children;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-21-10:50
* @ Version：1.0
* @ Description：数据源远程调用
*/
@FeignClient(contextId = "RemoteChildrenService", value = ServiceNameConstants.SOURCE_SERVICE
, fallbackFactory = RemoteChildrenService.class, path = "/children")
public interface RemoteChildrenService {
@GetMapping("/getChildren/{id}")
Result<Children> getChildren( @PathVariable Long id);
}

31
muyu-source-remote/src/main/java/com/muyu/source/remote/RemoteDataSourceService.java
View File

@ -1,31 +0,0 @@
package com.muyu.source.remote;
import com.muyu.common.core.constant.ServiceNameConstants;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.DataSource;
import com.muyu.source.remote.factory.RemoteDataSourceFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import java.util.List;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-21-10:50
* @ Version：1.0
* @ Description：数据源远程调用
*/
@FeignClient(contextId = "RemoteDataSourceService", value = ServiceNameConstants.SOURCE_SERVICE
, fallbackFactory = RemoteDataSourceFactory.class, path = "/dataSource")
public interface RemoteDataSourceService {
@GetMapping("/getDataSourceById/{id}")
Result<DataSource> getDataSource(@PathVariable Long id);
@GetMapping("/getDataSourceList")
Result<List<DataSource>> getDataSourceList();
}

25
muyu-source-remote/src/main/java/com/muyu/source/remote/RemoteDataTypeService.java
View File

@ -1,25 +0,0 @@
package com.muyu.source.remote;
import com.muyu.common.core.constant.ServiceNameConstants;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.DataType;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-21-10:50
* @ Version：1.0
* @ Description：数据源远程调用
*/
@FeignClient(contextId = "RemoteDataTypeService", value = ServiceNameConstants.SOURCE_SERVICE
, fallbackFactory = RemoteDataTypeService.class, path = "/dataType")
public interface RemoteDataTypeService {
@GetMapping("/getDataType/{type}")
Result<DataType> getDataType( @PathVariable String type);
}

26
muyu-source-remote/src/main/java/com/muyu/source/remote/factory/RemoteChildrenFactory.java
View File

@ -1,26 +0,0 @@
package com.muyu.source.remote.factory;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.Children;
import com.muyu.source.remote.RemoteChildrenService;
import org.springframework.cloud.openfeign.FallbackFactory;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-21-10:54
* @ Version：1.0
* @ Description：
*/
public class RemoteChildrenFactory implements FallbackFactory<RemoteChildrenService> {
@Override
public RemoteChildrenService create(Throwable cause) {
return new RemoteChildrenService() {
@Override
public Result<Children> getChildren(Long id) {
return Result.error(cause.getMessage());
}
};
}
}

33
muyu-source-remote/src/main/java/com/muyu/source/remote/factory/RemoteDataSourceFactory.java
View File

@ -1,33 +0,0 @@
package com.muyu.source.remote.factory;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.DataSource;
import com.muyu.source.remote.RemoteDataSourceService;
import org.springframework.cloud.openfeign.FallbackFactory;
import java.util.List;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-21-10:54
* @ Version：1.0
* @ Description：
*/
public class RemoteDataSourceFactory implements FallbackFactory<RemoteDataSourceService> {
@Override
public RemoteDataSourceService create(Throwable cause) {
return new RemoteDataSourceService() {
@Override
public Result<DataSource> getDataSource(Long id) {
return Result.error(cause.getMessage());
}
@Override
public Result<List<DataSource>> getDataSourceList() {
return Result.error(cause.getMessage());
}
};
}
}

26
muyu-source-remote/src/main/java/com/muyu/source/remote/factory/RemoteDataTypeFactory.java
View File

@ -1,26 +0,0 @@
package com.muyu.source.remote.factory;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.DataType;
import com.muyu.source.remote.RemoteDataTypeService;
import org.springframework.cloud.openfeign.FallbackFactory;
/**
* @author Lenovo
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-08-21-10:54
* @ Version：1.0
* @ Description：
*/
public class RemoteDataTypeFactory implements FallbackFactory<RemoteDataTypeService> {
@Override
public RemoteDataTypeService create(Throwable cause) {
return new RemoteDataTypeService() {
@Override
public Result<DataType> getDataType(String type) {
return Result.error(cause.getMessage());
}
};
}
}

29
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/RemoteDataAnalysisService.java 100644
View File

@ -0,0 +1,29 @@
package com.muyu.source.remote.remote;
import com.muyu.common.core.constant.ServiceNameConstants;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.TableData;
import com.muyu.source.remote.remote.factory.RemoteDataAnalysisBackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/2 14:35
* @注释
*/
@FeignClient(contextId = "remoteDataAnalysisService",
value = ServiceNameConstants.SOURCE_SERVICE,
path = "structure",
fallbackFactory = RemoteDataAnalysisBackFactory.class)
public interface RemoteDataAnalysisService {
@PostMapping("/findDataAnalysisByTableId/{id}")
Result<List<TableData>> findDataAnalysisByTableId(@PathVariable("id") Long id);
}

27
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/RemoteDataSourceService.java 100644
View File

@ -0,0 +1,27 @@
package com.muyu.source.remote.remote;
import com.muyu.common.core.constant.ServiceNameConstants;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.resp.DataSourceResp;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/2 14:35
* @注释
*/
@FeignClient(contextId = "remoteDataAnalysisService",
value = ServiceNameConstants.SOURCE_SERVICE,
path = "source")
// TODO 熔断
public interface RemoteDataSourceService {
@GetMapping("/isUseDatasource")
public Result<List<DataSourceResp>> getIsUseDataSourceList();
}

48
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/RemoteEtlDataService.java 100644
View File

@ -0,0 +1,48 @@
package com.muyu.source.remote.remote;
import com.muyu.common.core.constant.ServiceNameConstants;
import com.muyu.common.core.domain.Result;
import com.muyu.source.core.DataValue;
import com.muyu.source.domain.model.DataValueModel;
import com.muyu.source.remote.remote.factory.RemoteEtlDataBackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/1 9:27
* @注释
*/
@FeignClient(contextId = "remoteEtlDataService",
value = ServiceNameConstants.SOURCE_SERVICE,
path = "value",
fallbackFactory = RemoteEtlDataBackFactory.class)
public interface RemoteEtlDataService {
/**
* 远调ktly格式数据
*/
@PostMapping("/findTableValueBySql")
Result<List<List<DataValue>>> selEtlData(@RequestBody DataValueModel etlDataSqlReq);
/**
* 根据基础表ID和SQL语句查询数据
*
* @param etlDataSqlReq 基础表ID和sql语句
* @return DataValue{kltv}
*/
@PostMapping("/addTableValueBy")
Result addTableValue(@RequestBody DataValueModel etlDataSqlReq);
/**
* 根据基础表ID和SQL语句查询数据总数
* @param etlDataSqlReq 基础表ID和sql语句
* @return Integer
*/
@PostMapping("/getTableTotal")
Result getTableValueTotal(@RequestBody DataValueModel etlDataSqlReq);
}

32
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/RemoteTableInfoService.java 100644
View File

@ -0,0 +1,32 @@
package com.muyu.source.remote.remote;
import com.muyu.common.core.constant.ServiceNameConstants;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.TableInfo;
import com.muyu.source.domain.rep.TableInfoResp;
import com.muyu.source.remote.remote.factory.RemoteTableInfoBackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/1 10:56
* @注释
*/
@FeignClient(
contextId = "remoteTableInfoService",
path = "tableInfo",
value = ServiceNameConstants.SOURCE_SERVICE,
fallbackFactory = RemoteTableInfoBackFactory.class)
public interface RemoteTableInfoService {
@GetMapping("/findByTableName")
Result<List<TableInfoResp>> findByTableName();
@GetMapping("/getTableInfoDetails")
Result<List<TableInfo>> getTableInfoDetails(Integer[] ids);
}

33
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/factory/RemoteDataAnalysisBackFactory.java 100644
View File

@ -0,0 +1,33 @@
package com.muyu.source.remote.remote.factory;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.TableData;
import com.muyu.source.remote.remote.RemoteDataAnalysisService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/1 10:04
* @注释
*/
@Component
public class RemoteDataAnalysisBackFactory implements FallbackFactory<RemoteDataAnalysisService> {
private static final Logger log = LoggerFactory.getLogger(RemoteDataAnalysisBackFactory.class);
@Override
public RemoteDataAnalysisService create(Throwable cause) {
log.error("文件服务调用失败:{}", cause.getMessage());
return new RemoteDataAnalysisService() {
@Override
public Result<List<TableData>> findDataAnalysisByTableId(Long id) {
return Result.error("查询数据失败:"+cause.getMessage());
}
};
}
}

22
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/factory/RemoteDataSourceBackFactory.java 100644
View File

@ -0,0 +1,22 @@
package com.muyu.source.remote.remote.factory;
import com.muyu.common.core.domain.Result;
import com.muyu.source.remote.remote.RemoteDataSourceService;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component;
/**
* @Author xie ya ru
* @Data 2024/9/9
* @Description
* @Version 1.0.0
*/
@Component
public class RemoteDataSourceBackFactory implements FallbackFactory<RemoteDataSourceService> {
@Override
public RemoteDataSourceService create(Throwable cause) {
return () -> Result.error("查询数据失败:"+cause.getMessage());
}
}

47
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/factory/RemoteEtlDataBackFactory.java 100644
View File

@ -0,0 +1,47 @@
package com.muyu.source.remote.remote.factory;
import com.muyu.common.core.domain.Result;
import com.muyu.source.core.DataValue;
import com.muyu.source.domain.model.DataValueModel;
import com.muyu.source.remote.remote.RemoteEtlDataService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/1 10:04
* @注释
*/
@Component
public class RemoteEtlDataBackFactory implements FallbackFactory<RemoteEtlDataService> {
private static final Logger log = LoggerFactory.getLogger(RemoteEtlDataBackFactory.class);
@Override
public RemoteEtlDataService create(Throwable cause) {
log.error("文件服务调用失败:{}", cause.getMessage());
return new RemoteEtlDataService() {
@Override
public Result<List<List<DataValue>>> selEtlData(DataValueModel etlDataSqlReq) {
return Result.error("查询数据失败:"+cause.getMessage());
}
@Override
public Result addTableValue(DataValueModel etlDataSqlReq) {
return Result.error("查询数据失败:"+cause.getMessage());
}
@Override
public Result getTableValueTotal(DataValueModel etlDataSqlReq) {
return Result.error("查询数据失败:"+cause.getMessage());
}
};
}
}

40
muyu-source-remote/src/main/java/com/muyu/source/remote/remote/factory/RemoteTableInfoBackFactory.java 100644
View File

@ -0,0 +1,40 @@
package com.muyu.source.remote.remote.factory;
import com.muyu.common.core.domain.Result;
import com.muyu.source.domain.TableInfo;
import com.muyu.source.domain.rep.TableInfoResp;
import com.muyu.source.remote.remote.RemoteTableInfoService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/1 10:58
* @注释
*/
@Component
public class RemoteTableInfoBackFactory implements FallbackFactory<RemoteTableInfoService> {
private static final Logger log = LoggerFactory.getLogger(RemoteTableInfoBackFactory.class);
@Override
public RemoteTableInfoService create(Throwable cause) {
return new RemoteTableInfoService() {
@Override
public Result<List<TableInfoResp>> findByTableName() {
return Result.error("获取数据库和表失败:"+cause.getMessage());
}
@Override
public Result<List<TableInfo>> getTableInfoDetails(Integer[] ids) {
return Result.error("获取数据库和表失败:"+cause.getMessage());
}
};
}
}

7
muyu-source-remote/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
View File

@ -1,3 +1,4 @@
com.muyu.source.remote.factory.RemoteDataSourceFactory
com.muyu.source.remote.factory.RemoteDataTypeFactory
com.muyu.source.remote.factory.RemoteChildrenFactory
com.muyu.source.remote.remote.factory.RemoteDataAnalysisBackFactory
com.muyu.source.remote.remote.factory.RemoteDataSourceBackFactory
com.muyu.source.remote.remote.factory.RemoteEtlDataBackFactory
com.muyu.source.remote.remote.factory.RemoteTableInfoBackFactory

6
muyu-source-server/pom.xml
View File

@ -88,6 +88,12 @@
<groupId>com.muyu</groupId>
<artifactId>muyu-source-common</artifactId>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>muyu-source-client</artifactId>
<version>3.6.5</version>
<scope>compile</scope>
</dependency>
</dependencies>
<build>

8
muyu-source-server/src/main/java/com/muyu/source/controller/DataSourceController.java
View File

@ -186,7 +186,13 @@ public class DataSourceController extends BaseController {
Integer i = dataSourceService.syncAssetStructure(dataSource);
return Result.success(i);
}
@GetMapping("/isUseDatasource")
public Result<List<DataSource>> isUseDatasource(){
List<DataSource> dataSourceList = dataSourceService.list(new LambdaUpdateWrapper<>() {{
eq(DataSource::getIsInit, "Y");
}});
return Result.success(dataSourceList);
}
}

93
muyu-source-server/src/main/java/com/muyu/source/controller/DataValueController.java
View File

@ -50,6 +50,7 @@ public class DataValueController {
return Result.success(dataValueList);
}
/**
* 根据基础表ID和SQL语句查询数据总数
* @param dataValueModel 基础表ID和sql语句
@ -74,19 +75,19 @@ public class DataValueController {
Integer i = dataValueService.addTableValue(dataValueModel);
return Result.success(i);
}
/**
* 根据基础表ID和表名查询数据
*
* @param basicId 基础表ID
* @param tableName 表名
* @return DataValue{kltv}
*/
@PostMapping("/findTableValueByTableName")
@Operation(summary = "根据基础表ID和SQL语句查询数据", description = "根据基础表ID和SQL语句查询数据")
public Result findTableValueByTableName(@RequestParam("basicId") Long basicId, @RequestParam("tableName") String tableName) {
List<DataValue> dataValueList = dataValueService.findTableValueByTableName(basicId, tableName);
return Result.success(dataValueList);
}
// /**
// * 根据基础表ID和表名查询数据
// *
// * @param basicId 基础表ID
// * @param tableName 表名
// * @return DataValue{kltv}
// */
// @PostMapping("/findTableValueByTableName")
// @Operation(summary = "根据基础表ID和SQL语句查询数据", description = "根据基础表ID和SQL语句查询数据")
// public Result findTableValueByTableName(@RequestParam("basicId") Long basicId, @RequestParam("tableName") String tableName) {
// List<DataValue> dataValueList = dataValueService.findTableValueByTableName(basicId, tableName);
// return Result.success(dataValueList);
// }
/**
* 给任务模块去调用
@ -120,4 +121,70 @@ public class DataValueController {
System.out.println("耗时："+(end-begin));
return Result.success();
}
/**
* 根据基础表ID和SQL语句查询数据
* 给任务模块调用
*
* @param dataValueModel 基础表ID和sql语句
* @return DataValue{kltv}
*/
@PostMapping("/findTableValueBySql")
@Operation(summary = "根据基础表ID和SQL语句查询数据", description = "根据基础表ID和SQL语句查询数据")
public Result findTableValueBySql(@RequestBody DataValueModel dataValueModel) {
// 测试耗时
//开始执行的时间
long begin = System.currentTimeMillis();
List<List<DataValue>> dataValueList = dataValueService.findTableValueBySql(dataValueModel);
// 结束执行的时间
long end =System.currentTimeMillis();
//打印出执行该方法的耗时
System.out.println("耗时："+(end-begin));
return Result.success(dataValueList);
}
/**
* 获取表中的数据值前台调用
* @param basicId 响应参数
* @param tableName 响应参数
* @return 返回结果
*/
@PostMapping("/findTableValueList")
@Operation(summary = "获取表中的数据值前台调用",
description = "获取表中的数据值前台调用")
public Result<List<DataValue>> findTableValueByTable(@RequestParam("basicId") Long basicId,@RequestParam("tableName") String tableName){
List<DataValue> list = dataValueService.findTableValueByTable(basicId,tableName);
return Result.success(list);
}
/**
* 根据基础表ID和SQL语句查询数据总数
* @param etlDataSqlReq 基础表ID和sql语句
* @return Integer
*/
@PostMapping("/getTableTotal")
@Operation(summary = "根据基础表ID和SQL语句查询数据总数", description = "根据基础表ID和SQL语句查询数据总数")
public Result getTableTotal(@RequestBody DataValueModel etlDataSqlReq) {
Integer i = dataValueService.getTableTotal(etlDataSqlReq);
return Result.success(i);
}
/**
* 根据基础表ID和SQL语句查询数据
*
* @param etlDataSqlReq 基础表ID和sql语句
* @return DataValue{kltv}
*/
@PostMapping("/addTableValueBy")
@Operation(summary = "根据基础表ID和SQL语句新增数据", description = "根据基础表ID和SQL语句新增数据")
public Result addTableValueBy(@RequestBody DataValueModel etlDataSqlReq) {
//开始执行的时间
long begin = System.currentTimeMillis();
Integer i = dataValueService.addTableValueBy(etlDataSqlReq);
// 结束执行的时间
long end =System.currentTimeMillis();
//打印出执行该方法的耗时
System.out.println("耗时："+(end-begin));
return Result.success(i);
}
}

39
muyu-source-server/src/main/java/com/muyu/source/controller/StructureController.java 100644
View File

@ -0,0 +1,39 @@
package com.muyu.source.controller;
import com.muyu.common.core.domain.Result;
import com.muyu.common.core.web.controller.BaseController;
import com.muyu.source.domain.Structure;
import com.muyu.source.service.StructureService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.annotation.Resource;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
/**
* @ Tool：IntelliJ IDEA
* @ Author：CHX
* @ Date：2024-09-10-14:36
* @ Version：1.0
* @ Description：资产结构
* @author Lenovo
*/
@RestController
@RequestMapping("/structure")
@Tag(name = "数据资产结构",description = "根据对应的表名称查询数据资产结构")
public class StructureController extends BaseController {
@Resource
private StructureService structureService;
@Operation(summary = "数据资产结构",description = "根据表的ID查询表中数据资产结构")
@PostMapping("/findDataAnalysisByTableId/{id}")
public Result<List<Structure>> findDataAnalysisByTableId(@PathVariable("id") Long id){
List<Structure> list = structureService.findStructureByTableId(id);
return Result.success(list,"操作成功");
}
}

8
muyu-source-server/src/main/java/com/muyu/source/controller/TableInfoController.java
View File

@ -234,7 +234,13 @@ public class TableInfoController {
List<TableInfo> list =tableInfoService.findTableNameById(id);
return Result.success(list);
}
@Operation(summary = "根据主键ID查询数据",description = "根据主键ID查询数据")
@GetMapping("/getTableInfoDetails/{ids}")
Result<List<TableInfo>> getTableInfoDetails(@PathVariable("ids") Integer[] ids){
LambdaQueryWrapper<TableInfo> wrapper = new LambdaQueryWrapper<>();
wrapper.in(TableInfo::getId, (Object[]) ids);
return Result.success(tableInfoService.list(wrapper));
}
/**
* 给规则引擎调用
* 根据parentId查询数据接入

20
muyu-source-server/src/main/java/com/muyu/source/mysql/MysqlDataSource.java
View File

@ -74,14 +74,14 @@ public class MysqlDataSource extends BaseDataAbsSource {
remarks = columns.getString("REMARKS");
}
log.info("字段备注"+remarks);
DataValue build = DataValue.builder()
DataValue dataStructure = DataValue.builder()
.key(resultSetMetaData.getColumnName(i))
.label(remarks)
.value(resultSet.getObject(i, DataType.convertType(columnTypeName)))
.type(DataType.findBySqlType(columnTypeName))
.value(resultSet.getObject(i, DataType.getTargetClass(columnTypeName)))
.type(DataType.getDataType(columnTypeName))
.build();
map.put(i,build);
dataValues[i-1]=build;
map.put(i,dataStructure);
dataValues[i-1]=dataStructure;
}
}
@ -146,16 +146,16 @@ public class MysqlDataSource extends BaseDataAbsSource {
remarks = columns.getString("REMARKS");
}
//构建数据值对象
DataValue build = DataValue.builder()
DataValue dataStructure = DataValue.builder()
.key(resultSetMetaData.getColumnName(i))
.label(remarks)
.value(resultSet.getObject(i, DataType.convertType(columnTypeName)))
.type(DataType.findBySqlType(columnTypeName))
.value(resultSet.getObject(i, DataType.getTargetClass(columnTypeName)))
.type(DataType.getDataType(columnTypeName))
.build();
// 将数据值对象放入map集合中
map.put(i,build);
map.put(i,dataStructure);
// 将数据值对象放入二维数组中
dataValues[count][i]=build;
dataValues[count][i]=dataStructure;
}else {
// 获取当前列的值，并将其转换为与数据值对象中的类型匹配的类型
DataValue build = DataValue.builder()

4
muyu-source-server/src/main/java/com/muyu/source/service/DataSourceService.java
View File

@ -43,5 +43,9 @@ public interface DataSourceService extends IService<DataSource> {
Integer syncAssetStructure(DataSource dataSource);
// void synchronous(DataSource dataSource);
}

10
muyu-source-server/src/main/java/com/muyu/source/service/DataValueService.java
View File

@ -19,7 +19,7 @@ import java.util.List;
public interface DataValueService extends IService<DataValue> {
List<List<DataValue>> findTableValue(DataValueModel dataValueModel);
List<DataValue> findTableValueByTableName(Long basicId, String tableName);
// List<DataValue> findTableValueByTableName(Long basicId, String tableName);
Integer addTableValue(DataValueModel dataValueModel);
@ -34,4 +34,12 @@ public interface DataValueService extends IService<DataValue> {
DataValue[][] findTableValueByTable(DataValueModel dataValueModel);
int addTableValueByType(DataValueModels dataValueModel);
List<List<DataValue>> findTableValueBySql(DataValueModel dataValueModel);
List<DataValue> findTableValueByTable(Long basicId, String tableName);
Integer getTableTotal(DataValueModel etlDataSqlReq);
Integer addTableValueBy(DataValueModel etlDataSqlReq);
}

1
muyu-source-server/src/main/java/com/muyu/source/service/Impl/DataSourceServiceImpl.java
View File

@ -241,6 +241,7 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
return 1;
}
private void syncData(Connection conn, String databaseName, TableInfo table) {
ExecutorService threadPool = Executors.newCachedThreadPool();
PreparedStatement ps = null;

376
muyu-source-server/src/main/java/com/muyu/source/service/Impl/DataValueServiceImpl.java
View File

@ -1,6 +1,7 @@
package com.muyu.source.service.Impl;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.muyu.etl.data.access.data.access.client.config.DatabaseConnectionPool;
import com.muyu.source.base.BaseQueryHandler;
import com.muyu.source.core.DataType;
import com.muyu.source.core.DataValue;
@ -13,6 +14,8 @@ import com.muyu.source.mysql.MysqlQuery;
import com.muyu.source.pool.MysqlPool;
import com.muyu.source.service.DataSourceService;
import com.muyu.source.service.DataValueService;
import com.zaxxer.hikari.HikariDataSource;
import jakarta.annotation.Resource;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@ -59,6 +62,8 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
@Autowired
private DataSourceService dataSourceService;
@Resource
private DatabaseConnectionPool databaseConnectionPool;
@Override
public List<List<DataValue>> findTableValue(DataValueModel dataValueModel) {
@ -108,10 +113,10 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
}
// 构建数据值对象，包含列名、备注、值、类型等信息
DataValue build = DataValue.builder()
.key(metaData.getColumnName(i))// 当前列的名称
.label(remarks)// 当前列的备注信息
.value(resultSet.getObject(i, DataType.convertType(columnTypeName)))// 当前列的值，类型转换
.type(DataType.findBySqlType(columnTypeName))// 当前列的类型，转换为字符串表示
.key(metaData.getColumnName(i))
.label(remarks)
.value(resultSet.getObject(i, DataType.getTargetClass(columnTypeName)))
.type(DataType.getDataType(columnTypeName))
.build();
dataValues.add(build);
map.put(i, build);
@ -139,124 +144,283 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
// 返回包含数据值的列表
return list;
}
@Override
public List<List<DataValue>> findTableValueBySql(DataValueModel dataValueModel) {
//根据数据源ID获取数据源信息
ConcurrentHashMap<Integer, DataValue> map = new ConcurrentHashMap<>();
//初始化一个列表,用于存储数据值对象
List<List<DataValue>> list = new ArrayList<>();
HikariDataSource hikariDataSource = databaseConnectionPool.get(dataValueModel.getBasicId());
Connection conn = null;
try {
conn = hikariDataSource.getConnection();
//准备SQL查询
Statement statement = conn.createStatement();
ResultSet resultSet = statement.executeQuery(dataValueModel.getSql());
//获取元数据
ResultSetMetaData metaData = resultSet.getMetaData();
//获取列的数量
int columnCount = metaData.getColumnCount();
//遍历每一行数据
while (resultSet.next()) {
List<DataValue> dataStructures = new ArrayList<>();
// 提交任务给线程池
for (int i = 1; i <= columnCount; i++) {
if (resultSet.isFirst()) {
String columnTypeName = metaData.getColumnTypeName(i);
DatabaseMetaData metaDataTwo = conn.getMetaData();
ResultSet columns = metaDataTwo.getColumns(null, null, metaData.getTableName(i), metaData.getColumnName(i));
String remarks = null;
while (columns.next()) {
remarks = columns.getString("REMARKS");
}
DataValue build = DataValue.builder()
.key(metaData.getColumnName(i))
.label(remarks)
.value(resultSet.getObject(i, DataType.getTargetClass(columnTypeName)))
.type(DataType.getDataType(columnTypeName))
.build();
dataStructures.add(build);
map.put(i, build);
} else {
DataValue build = DataValue.builder()
.key(metaData.getColumnName(i))
.label(map.get(i).getLabel())
.value(resultSet.getObject(i, map.get(i).getType().getTargetType()))
.type(map.get(i).getType())
.build();
dataStructures.add(build);
}
}
list.add(dataStructures);
}
}catch (Exception e) {
throw new RuntimeException(e);
}finally {
databaseConnectionPool.returnConnection(conn);
}
return list;
}
@Override
public List<DataValue> findTableValueByTableName(Long basicId, String tableName) {
List<DataValue> dataValues = new ArrayList<>();
DataSource dataSources = dataSourceService.getById(basicId);
MysqlPool mysqlPool = new MysqlPool(dataSources);
mysqlPool.init();
Connection conn = mysqlPool.getConn();
public List<DataValue> findTableValueByTable(Long basicId, String tableName) {
ConcurrentHashMap<Integer, DataValue> map = new ConcurrentHashMap<>();
ArrayList<DataValue> dataValues = new ArrayList<>();
HikariDataSource hikariDataSource = databaseConnectionPool.get(basicId);
Connection conn = null;
try {
conn = hikariDataSource.getConnection();
// 准备SQL查询语句
String sql = SELECTALL + tableName;
PreparedStatement preparedStatement = conn.prepareStatement(sql);
PreparedStatement preparedStatement = conn.prepareStatement(SELECTALL + tableName + " LIMIT " + PAGE_SIZE);
// 执行查询，获取结果集
ResultSet resultSet = preparedStatement.executeQuery();
// 获取结果集的元数据，用于获取列的数量和类型等信息
ResultSetMetaData metaData = preparedStatement.getMetaData();
ResultSetMetaData metaData = resultSet.getMetaData();
// 获取列的数量
int columnCount = metaData.getColumnCount();
// 使用Map存储列的元数据，包括备注信息
Map<String, ColumnMeta> columnMetadata = getColumnMetadata(conn, metaData, columnCount);
// 初始化分页参数
int offset = 0;
boolean hasMorePages = true;
while (hasMorePages) {
// 构造带分页的SQL查询
String pageSql = sql + " LIMIT 1 ";
PreparedStatement pageStatement = conn.prepareStatement(pageSql);
ResultSet pageResultSet = pageStatement.executeQuery();
List<DataValue> pageDataValues = new ArrayList<>();
while (pageResultSet.next()) {
List<DataValue> rowValues = new ArrayList<>();
// 遍历结果集中的每一行数据
while (resultSet.next()) {
// 遍历每一列
for (int i = 1; i <= columnCount; i++) {
String columnName = metaData.getColumnName(i);
ColumnMeta columnData = columnMetadata.get(columnName);
Object value = pageResultSet.getObject(i, columnData.getType().getTargetType());
if (resultSet.isFirst()) {
// 获取当前列的类型名称
String columnTypeName = metaData.getColumnTypeName(i);
// 获取数据库的元数据对象
DatabaseMetaData metaDataColumns = conn.getMetaData();
// 查询数据库元数据，获取当前列的备注信息
ResultSet columns = metaDataColumns.getColumns(null, null, metaData.getTableName(i), metaData.getColumnName(i));
String remarks = null;
// 遍历备注信息的结果集
while (columns.next()) {
// 获取当前字段的备注信息
remarks = columns.getString("REMARKS");
// 记录日志，显示字段的备注信息
log.info("字段备注:" + remarks);
}
// 构建数据值对象，包含列名、备注、值、类型等信息
DataValue dataValue = DataValue.builder()
.key(columnName)
.label(columnData.getRemarks())
.value(value)
.type(columnData.getType())
DataValue build = DataValue.builder()
.key(metaData.getColumnName(i))// 当前列的名称
.label(remarks)// 当前列的备注信息
.value(resultSet.getObject(i, DataType.getTargetClass(columnTypeName)))// 当前列的值，类型转换
.type(DataType.getDataType(columnTypeName))// 当前列的类型，转换为字符串表示
.build();
rowValues.add(dataValue);
dataValues.add(build);
map.put(i, build);
} else {
DataValue build = DataValue.builder()
.key(metaData.getColumnName(i))// 当前列的名称
.label(map.get(i).getLabel())// 当前列的备注信息
.value(resultSet.getObject(i, map.get(i).getType().getTargetType()))// 当前列的值，类型转换
.type(map.get(i).getType())// 当前列的类型，转换为字符串表示
.build();
dataValues.add(build);
}
pageDataValues.addAll(rowValues);
}
// 每处理完一定数量的数据，提交一次
if (!pageDataValues.isEmpty()) {
CompletableFuture<Void> future = processBatchAsync(pageDataValues);
futures.add(future);
dataValues.addAll(pageDataValues);
}
// 检查是否有更多页面
hasMorePages = pageResultSet.getFetchSize() >= PAGE_SIZE;
offset += PAGE_SIZE;
}
// 确保所有异步任务完成后再关闭线程池
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
} catch (SQLException e) {
// 如果发生SQL异常，抛出运行时异常
throw new RuntimeException(e);
} finally {
// 在所有异步任务完成后关闭线程池
// shutdownExecutorService(executorService);
// 释放数据库连接
mysqlPool.replease(conn);
// 关闭数据库连接
mysqlPool.closeConn();
}finally {
databaseConnectionPool.returnConnection(conn);
}
// 返回包含数据值的列表
return dataValues;
}
private CompletableFuture<Void> processBatchAsync(final List<DataValue> batch) {
// 记录任务提交的时间点
System.out.println("任务提交：" + LocalDateTime.now());
return CompletableFuture.runAsync(() -> {
@Override
public Integer getTableTotal(DataValueModel etlDataSqlReq) {
HikariDataSource hikariDataSource = databaseConnectionPool.get(etlDataSqlReq.getBasicId());
Connection conn = databaseConnectionPool.getConn(hikariDataSource);
try {
// 具体的批量处理逻辑，例如写入文件、更新数据库等
this.saveBatch(batch);
} catch (Exception e) {
// 异步处理中的异常处理
e.printStackTrace();
Statement statement = conn.createStatement();
ResultSet resultSet = statement.executeQuery(etlDataSqlReq.getSql());
if (resultSet.next()) {
return resultSet.getInt(1);
}
// 记录任务完成的时间点
System.out.println("任务完成：" + LocalDateTime.now());
}, executorService);
} catch (SQLException e) {
throw new RuntimeException(e);
}finally {
databaseConnectionPool.returnConnection(conn);
}
return 0;
}
private Map<String, ColumnMeta> getColumnMetadata(Connection conn, ResultSetMetaData metaData, int columnCount) throws SQLException {
Map<String, ColumnMeta> metadataMap = new HashMap<>();
for (int i = 1; i <= columnCount; i++) {
String columnName = metaData.getColumnName(i);
String columnTypeName = metaData.getColumnTypeName(i);
// 查询数据库元数据，获取当前列的备注信息
ResultSet columns = conn.getMetaData().getColumns(null, null, metaData.getTableName(i), columnName);
String remarks = null;
while (columns.next()) {
remarks = columns.getString("REMARKS");
@Override
public Integer addTableValueBy(DataValueModel etlDataSqlReq) {
HikariDataSource hikariDataSource = databaseConnectionPool.get(etlDataSqlReq.getBasicId());
Connection conn = databaseConnectionPool.getConn(hikariDataSource);
try {
PreparedStatement preparedStatement = conn.prepareStatement(etlDataSqlReq.getSql());
return preparedStatement.executeUpdate();
} catch (SQLException e) {
throw new RuntimeException(e);
}finally {
databaseConnectionPool.returnConnection(conn);
}
}
// 构建列元数据对象
ColumnMeta columnData = new ColumnMeta(columnName, remarks, DataType.findBySqlType(columnTypeName));
metadataMap.put(columnName, columnData);
}
return metadataMap;
}
// @Override
// public List<DataValue> findTableValueByTableName(Long basicId, String tableName) {
// List<DataValue> dataValues = new ArrayList<>();
// DataSource dataSources = dataSourceService.getById(basicId);
// MysqlPool mysqlPool = new MysqlPool(dataSources);
// mysqlPool.init();
// Connection conn = mysqlPool.getConn();
//
// try {
// // 准备SQL查询语句
// String sql = SELECTALL + tableName;
// PreparedStatement preparedStatement = conn.prepareStatement(sql);
// // 获取结果集的元数据，用于获取列的数量和类型等信息
// ResultSetMetaData metaData = preparedStatement.getMetaData();
// // 获取列的数量
// int columnCount = metaData.getColumnCount();
//
// // 使用Map存储列的元数据，包括备注信息
// Map<String, ColumnMeta> columnMetadata = getColumnMetadata(conn, metaData, columnCount);
//
// // 初始化分页参数
// int offset = 0;
// boolean hasMorePages = true;
//
// while (hasMorePages) {
// // 构造带分页的SQL查询
// String pageSql = sql + " LIMIT " + PAGE_SIZE + " OFFSET " + offset;
// PreparedStatement pageStatement = conn.prepareStatement(pageSql);
// ResultSet pageResultSet = pageStatement.executeQuery();
//
// List<DataValue> pageDataValues = new ArrayList<>();
// while (pageResultSet.next()) {
// List<DataValue> rowValues = new ArrayList<>();
// for (int i = 1; i <= columnCount; i++) {
// String columnName = metaData.getColumnName(i);
// ColumnMeta columnData = columnMetadata.get(columnName);
// Object value = pageResultSet.getObject(i, columnData.getType().getTargetType());
//
// // 构建数据值对象，包含列名、备注、值、类型等信息
// DataValue dataValue = DataValue.builder()
// .key(columnName)
// .label(columnData.getRemarks())
// .value(value)
// .type(columnData.getType())
// .build();
// rowValues.add(dataValue);
// }
// pageDataValues.addAll(rowValues);
// }
//
// // 每处理完一定数量的数据，提交一次
// if (!pageDataValues.isEmpty()) {
// CompletableFuture<Void> future = processBatchAsync(pageDataValues);
// futures.add(future);
// dataValues.addAll(pageDataValues);
// }
//
// // 检查是否有更多页面
// hasMorePages = pageResultSet.getFetchSize() >= PAGE_SIZE;
// offset += PAGE_SIZE;
// }
//
// // 确保所有异步任务完成后再关闭线程池
// CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
//
// } catch (SQLException e) {
// // 如果发生SQL异常，抛出运行时异常
// throw new RuntimeException(e);
// } finally {
// // 在所有异步任务完成后关闭线程池
//// shutdownExecutorService(executorService);
//
// // 释放数据库连接
// mysqlPool.replease(conn);
// // 关闭数据库连接
// mysqlPool.closeConn();
// }
// // 返回包含数据值的列表
// return dataValues;
// }
//
// private CompletableFuture<Void> processBatchAsync(final List<DataValue> batch) {
// // 记录任务提交的时间点
// System.out.println("任务提交：" + LocalDateTime.now());
//
// return CompletableFuture.runAsync(() -> {
// try {
// // 具体的批量处理逻辑，例如写入文件、更新数据库等
// this.saveBatch(batch);
// } catch (Exception e) {
// // 异步处理中的异常处理
// e.printStackTrace();
// }
// // 记录任务完成的时间点
// System.out.println("任务完成：" + LocalDateTime.now());
// }, executorService);
// }
//
// private Map<String, ColumnMeta> getColumnMetadata(Connection conn, ResultSetMetaData metaData, int columnCount) throws SQLException {
// Map<String, ColumnMeta> metadataMap = new HashMap<>();
// for (int i = 1; i <= columnCount; i++) {
// String columnName = metaData.getColumnName(i);
// String columnTypeName = metaData.getColumnTypeName(i);
//
// // 查询数据库元数据，获取当前列的备注信息
// ResultSet columns = conn.getMetaData().getColumns(null, null, metaData.getTableName(i), columnName);
// String remarks = null;
// while (columns.next()) {
// remarks = columns.getString("REMARKS");
// }
//
// // 构建列元数据对象
// ColumnMeta columnData = new ColumnMeta(columnName, remarks, DataType.findBySqlType(columnTypeName));
// metadataMap.put(columnName, columnData);
// }
// return metadataMap;
// }
private static class ColumnMeta {
private final String columnName;
@ -409,32 +573,6 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
}
return atomicInteger.get();
}
/**
* 格式化数据
* @param type 数据类型
* @param value 值
* @return 格式化后的值
*/
private String formatValue(DataType type, Object value) {
if (type == DataType.VARCHAR || type == DataType.TEXT) {
return "'" + value.toString().replace("'", "''") + "'";
} else if (type == DataType.BIGINT) {
return value.toString();
} else if (type == DataType.INT) {
return value.toString();
} else if (type == DataType.DECIMAL) {
return value.toString();
} else if (type == DataType.DATETIME) {
return "'" + new java.sql.Date(((java.util.Date) value).getTime()) + "'";
} else if (type == DataType.DOUBLE) {
return value.toString();
} else {
// 其他类型的处理
return "'" + value.toString() + "'";
}
}
/**
* 将数据拆分成多个小批量
* @param dataValues 数据

8
muyu-source-server/src/main/java/com/muyu/source/service/Impl/StructureServiceImpl.java
View File

@ -39,5 +39,13 @@ public class StructureServiceImpl extends ServiceImpl<StructureMapper, Structure
}});
return list;
}
@Override
public List<Structure> findStructureByTableId(Long id) {
LambdaQueryWrapper<Structure> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(Structure::getTableId,id);
List<Structure> list = structureMapper.selectList(queryWrapper);
return list;
}
}

2
muyu-source-server/src/main/java/com/muyu/source/service/StructureService.java
View File

@ -12,4 +12,6 @@ public interface StructureService extends IService<Structure> {
List<Structure> findStructurelistS(Integer id);
List<Long> findTableIdByStructureId(Long parentId);
List<Structure> findStructureByTableId(Long id);
}