09062234:使用二维数组的方法给任务模块提供查询的接口

muyu-source-common/src/main/java/com/muyu/source/domain/model/DataValueModel.java

@@ -26,5 +26,7 @@ public class DataValueModel {
      * 拼写的sql语句
      */
     private String sql;
+    private Long one;
+    private Integer two;
 
 }

muyu-source-server/src/main/java/com/muyu/source/base/BaseQuery.java

@@ -1,6 +1,9 @@
 package com.muyu.source.base;
 
+import com.muyu.source.mysql.MysqlQuery;
+import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
@@ -11,9 +14,11 @@ import lombok.experimental.SuperBuilder;
  * @ Date：2024-08-30-21:33
  * @ Version：1.0
  */
+
 @Data
 @SuperBuilder
+@AllArgsConstructor
 @NoArgsConstructor
-public class BaseQuery {
+public class BaseQuery{
-
+    private Long dataSourceId;
 }

muyu-source-server/src/main/java/com/muyu/source/base/BaseQueryHandler.java

@@ -9,12 +9,13 @@ package com.muyu.source.base;
  */
 public class BaseQueryHandler {
 
-    private static final ThreadLocal<BaseQuery> BASE_QUERY_THREAD_LOCAL=new ThreadLocal<>();
+    private static final ThreadLocal<BaseQuery> BASE_QUERY_THREAD_LOCAL = new ThreadLocal<>();
+
     public static void set(BaseQuery baseQuery){
         BASE_QUERY_THREAD_LOCAL.set(baseQuery);
     }
 
-    public static<T> T get(){
+    public static <T> T  get(){
-        return (T)BASE_QUERY_THREAD_LOCAL.get();
+        return (T) BASE_QUERY_THREAD_LOCAL.get();
     }
 }

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

@@ -6,6 +6,7 @@ import com.muyu.source.domain.model.DataValueModel;
 import com.muyu.source.service.DataValueService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
+import org.apache.commons.lang3.SystemUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
 
@@ -42,13 +43,12 @@ public class DataValueController {
     }
 
     /**
-     * 根据基础表ID和SQL语句查询条数
+     * 根据基础表ID和SQL语句查询数据总数
-     *
      * @param dataValueModel 基础表ID和sql语句
-     * @return DataValue{kltv}
+     * @return Integer
      */
     @PostMapping("/getTableValueTotal")
-    @Operation(summary = "根据基础表ID和SQL语句新增数据", description = "根据基础表ID和SQL语句新增数据")
+    @Operation(summary = "根据基础表ID和SQL语句查询数据总数", description = "根据基础表ID和SQL语句查询数据总数")
     public Result getTableValueTotal(@RequestBody DataValueModel dataValueModel) {
         Integer i = dataValueService.getTableValueTotal(dataValueModel);
         return Result.success(i);
@@ -79,4 +79,29 @@ public class DataValueController {
         List<DataValue> dataValueList = dataValueService.findTableValueByTableName(basicId, tableName);
         return Result.success(dataValueList);
     }
+
+    /**
+     * 给任务模块去调用
+     *
+     * 根据基础表ID和表名查询数据
+     * @param dataValueModel 数据
+     * @return DataValue[][]
+     */
+    @PostMapping("/findTableValueByTableName")
+    public Result<DataValue[][]> findTableValueByTableName(@RequestBody DataValueModel dataValueModel) {
+        // 测试耗时
+        //开始执行的时间
+        long begin = System.currentTimeMillis();
+        // 执行查询的方法
+        DataValue[][] dataValues =dataValueService.findTableValueByTable(dataValueModel);
+        // 结束执行的时间
+        long end =System.currentTimeMillis();
+        //打印出执行该方法的耗时
+        System.out.println("耗时："+(end-begin));
+        // 返回数据
+        return Result.success(dataValues);
+    }
+
+
+
 }

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

@@ -1,7 +1,20 @@
 package com.muyu.source.mysql;
 
+import com.muyu.common.core.utils.SpringUtils;
 import com.muyu.source.base.BaseDataAbsSource;
+import com.muyu.source.core.DataType;
 import com.muyu.source.core.DataValue;
+import com.muyu.source.domain.DataSource;
+import com.muyu.source.pool.MysqlPool;
+import com.muyu.source.service.DataSourceService;
+import lombok.Data;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.stereotype.Component;
+
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * @author Lenovo
@@ -11,19 +24,157 @@ import com.muyu.source.core.DataValue;
  * @ Version：1.0
  */
 
+@Log4j2
+@Component
 public class MysqlDataSource extends BaseDataAbsSource {
+    DataSourceService dataSourceService = SpringUtils.getBean(DataSourceService.class);
+
     @Override
     public DataValue getDataValue() {
+
         return null;
     }
 
     @Override
     public DataValue[] getRow() {
-        return new DataValue[0];
+        // 查询数据
+        MysqlQuery query = getQuery();
+        // 获取到sql语句
+        String sql = query.getSql();
+        //获取到数据接入的ID
+        Long dataSourceId = query.getDataSourceId();
+        ConcurrentHashMap<Integer, DataValue> map = new ConcurrentHashMap<>();
+        // 从服务中获取EtlDataScore对象，根据基本ID
+        DataSource etlDataScore = dataSourceService.getById(dataSourceId);
+        // 创建MySQL连接池对象，并初始化
+        MysqlPool mysqlPool = new MysqlPool(etlDataScore);
+        mysqlPool.init();
+        // 从连接池获取数据库连接
+        Connection conn = mysqlPool.getConn();
+
+        // 初始化一个列表，用于存储数据值对象
+        DataValue[] dataValues=null;
+
+        try {
+            PreparedStatement preparedStatement = conn.prepareStatement(sql);
+            ResultSet resultSet = preparedStatement.executeQuery();
+            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+            int columnCount = resultSetMetaData.getColumnCount();
+            while (resultSet.next()){
+                for (int i = 0; i < columnCount; i++) {
+                    if(resultSet.isFirst()){
+                        String columnTypeName = resultSetMetaData.getColumnTypeName(i);
+                        DatabaseMetaData metaData = conn.getMetaData();
+                        ResultSet columns = metaData.getColumns(null, null, resultSetMetaData.getTableName(i), resultSetMetaData.getColumnName(i));
+                        String remarks =null;
+                        while (columns.next()) {
+                            remarks = columns.getString("REMARKS");
+                        }
+                        log.info("字段备注"+remarks);
+                        DataValue build = DataValue.builder()
+                                .key(resultSetMetaData.getColumnName(i))
+                                .label(remarks)
+                                .value(resultSet.getObject(i, DataType.convertType(columnTypeName)))
+                                .type(DataType.findBySqlType(columnTypeName))
+                                .build();
+                        map.put(i,build);
+                        dataValues[i-1]=build;
+                    }
+
+                }
+            }
+        } catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+
+
+        return dataValues;
     }
 
     @Override
     public DataValue[][] getRows() {
-        return new DataValue[0][];
+        // 查询数据
+        MysqlQuery query = getQuery();
+        //获取到sql语句
+        String sql = query.getSql();
+        //获取到数据接入的ID
+        Long dataSourceId = query.getDataSourceId();
+        //创建一个map集合，用于存储数据值对象
+        ConcurrentHashMap<Integer, DataValue> map = new ConcurrentHashMap<>();
+        Integer one = Math.toIntExact(query.getOne());
+        Integer two = query.getTwo();
+        // 从服务中获取EtlDataScore对象，根据基本ID
+        DataSource etlDataScore = dataSourceService.getById(dataSourceId);
+        // 创建MySQL连接池对象，并初始化
+        MysqlPool mysqlPool = new MysqlPool(etlDataScore);
+        // 初始化连接池
+        mysqlPool.init();
+        // 从连接池获取数据库连接
+        Connection conn = mysqlPool.getConn();
+        //new一个二维数组
+        DataValue[][] dataValues =new DataValue[one][two];
+        try {
+            // 准备SQL查询语句
+            PreparedStatement preparedStatement = conn.prepareStatement(sql);
+            // 获取结果集
+            ResultSet resultSet = preparedStatement.executeQuery();
+            // 获取结果集的元数据
+            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+            // 获取列的数量
+            int columnCount = resultSetMetaData.getColumnCount();
+            //定义一个变量统计行数
+            int count=0;
+            // 遍历结果集
+            while (resultSet.next()){
+                //遍历每一列
+                for (int i = 0; i < columnCount; i++) {
+                    //判断他是否是第一行的数据
+                    if(resultSet.isFirst()){
+                        //获取当前的类型名称
+                        String columnTypeName = resultSetMetaData.getColumnTypeName(i);
+                        // 获取数据库的元数据对象
+                        DatabaseMetaData metaData = conn.getMetaData();
+                        // 查询数据库元数据，获取当前列的备注信息
+                        ResultSet columns = metaData.getColumns(null, null, resultSetMetaData.getTableName(i), resultSetMetaData.getColumnName(i));
+                        String remarks =null;
+                        // 遍历备注信息的结果集
+                        while (columns.next()) {
+                            // 获取当前字段的备注信息
+                            remarks = columns.getString("REMARKS");
+                        }
+                        //构建数据值对象
+                        DataValue build = DataValue.builder()
+                                .key(resultSetMetaData.getColumnName(i))
+                                .label(remarks)
+                                .value(resultSet.getObject(i, DataType.convertType(columnTypeName)))
+                                .type(DataType.findBySqlType(columnTypeName))
+                                .build();
+                        // 将数据值对象放入map集合中
+                        map.put(i,build);
+                        // 将数据值对象放入二维数组中
+                        dataValues[count][i]=build;
+                    }else {
+                        // 获取当前列的值，并将其转换为与数据值对象中的类型匹配的类型
+                        DataValue build = DataValue.builder()
+                                .key(resultSetMetaData.getColumnName(i))
+                                .label(map.get(i).getLabel())
+                                .value(resultSet.getObject(i, map.get(i).getType().getTargetType()))
+                                .type(map.get(i).getType())
+                                .build();
+                        // 将转换后的数据值对象放入二维数组中
+                        dataValues[count][i]=build;
+                    }
+                    // 行数++
+                    count++;
+                }
+            }
+        } catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+        // 释放数据库连接
+        mysqlPool.replease(conn);
+        // 关闭数据库连接
+        mysqlPool.closeConn();
+        return dataValues;
     }
 }

muyu-source-server/src/main/java/com/muyu/source/mysql/MysqlQuery.java

@@ -1,12 +1,12 @@
 package com.muyu.source.mysql;
 
+import com.muyu.source.base.BaseQuery;
 import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
-import java.util.Map;
-
 /**
  * @author Lenovo
  * @ Tool：IntelliJ IDEA
@@ -14,12 +14,26 @@ import java.util.Map;
  * @ Date：2024-08-30-21:46
  * @ Version：1.0
  */
+
 @Data
 @SuperBuilder
 @AllArgsConstructor
 @NoArgsConstructor
-public class MysqlQuery {
+@EqualsAndHashCode(callSuper = true)
-    private String dataSourceId;
+public class MysqlQuery extends BaseQuery {
+    /**
+     * 任务传过来的sql语句
+     */
     private String sql;
-    private Map<String, Object> params;
+
+    private String params;
+    /**
+     * 行数
+     */
+    private Long one;
+    /**
+     * 列数
+     */
+
+    private Integer two;
 }

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

@@ -23,4 +23,13 @@ public interface DataValueService extends IService<DataValue> {
     Integer addTableValue(DataValueModel dataValueModel);
 
     Integer getTableValueTotal(DataValueModel dataValueModel);
+    /**
+     * 给任务模块去调用
+     *
+     * 根据基础表ID和表名查询数据
+     * @param dataValueModel 数据
+     * @return DataValue[][]
+     */
+    DataValue[][] findTableValueByTable(DataValueModel dataValueModel);
+
 }

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

@@ -1,11 +1,13 @@
 package com.muyu.source.service.Impl;
 
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.muyu.source.base.BaseQueryHandler;
 import com.muyu.source.core.DataType;
 import com.muyu.source.core.DataValue;
 import com.muyu.source.domain.DataSource;
 import com.muyu.source.domain.model.DataValueModel;
 import com.muyu.source.mapper.DataValueMapper;
+import com.muyu.source.mysql.MysqlDataSource;
 import com.muyu.source.mysql.MysqlQuery;
 import com.muyu.source.pool.MysqlPool;
 import com.muyu.source.service.DataSourceService;
@@ -216,37 +218,10 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
             // 关闭数据库连接
             mysqlPool.closeConn();
         }
-
         // 返回包含数据值的列表
         return dataValues;
     }
 
-//    private void shutdownExecutorService(ExecutorService executorService) {
-//        if (!isShuttingDown.compareAndSet(false, true)) {
-//            // 已经在关闭过程中，直接返回
-//            return;
-//        }
-//
-//        // 记录开始关闭线程池的时间点
-//        System.out.println("开始关闭线程池：" + LocalDateTime.now());
-//
-//        executorService.shutdown();
-//        try {
-//            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
-//                executorService.shutdownNow(); // 取消正在执行的任务
-//                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
-//                    System.err.println("线程池未正常关闭！");
-//                }
-//            }
-//        } catch (InterruptedException ex) {
-//            executorService.shutdownNow(); // (Re-)Cancel if current thread also interrupted
-//            Thread.currentThread().interrupt(); // Preserve interrupt status
-//        }
-//
-//        // 记录线程池关闭完成的时间点
-//        System.out.println("线程池关闭完成：" + LocalDateTime.now());
-//    }
-
     private CompletableFuture<Void> processBatchAsync(final List<DataValue> batch) {
         // 记录任务提交的时间点
         System.out.println("任务提交：" + LocalDateTime.now());
@@ -308,57 +283,10 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
         }
     }
 
-//    @Override
-//    public List<DataValue> findTableValueByTableName(Long basicId, String tableName) {
-//        MysqlQuery mySqlQuery = new MysqlQuery();
-//        mySqlQuery.setDataSourceId(String.valueOf(basicId));
-//        DataSource etlDataScore = dataSourceService.getById(basicId);
-//        MysqlPool mysqlPool = new MysqlPool(etlDataScore);
-//        mysqlPool.init();
-//        Connection conn = mysqlPool.getConn();
-//
-//        List<DataValue> list = new ArrayList<>();
-//
-//        try {
-//             //分页查询
-//            PreparedStatement preparedStatement = conn.prepareStatement(SELECTALL+tableName+" limit 200");
-//
-//            ResultSet resultSet = preparedStatement.executeQuery();
-//            ResultSetMetaData metaData = resultSet.getMetaData();
-//            int columnCount = metaData.getColumnCount();
-//
-//
-//            while (resultSet.next()){
-//                for (int i = 1; i <= columnCount; i++) {
-//                    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");
-//                    }
-//                    DataValue build = DataValue.builder()
-//                            .key(metaData.getColumnName(i))
-//                            .label(remarks)
-//                            .value(resultSet.getObject(i, DataType.convertType(columnTypeName)))
-//                            .type(DataType.convertTypeString(columnTypeName))
-//                            .build();
-//                    list.add(build);
-//                }
-//            }
-//        } catch (SQLException e) {
-//            throw new RuntimeException(e);
-//        }
-//        mysqlPool.replease(conn);
-//        mysqlPool.closeConn();
-//        return list;
-//    }
-
     @Override
     public Integer addTableValue(DataValueModel dataValueModel) {
         MysqlQuery mysqlQuery = new MysqlQuery();
-        mysqlQuery.setDataSourceId(String.valueOf(dataValueModel.getBasicId()));
+        mysqlQuery.setDataSourceId(dataValueModel.getBasicId());
         DataSource dataSource = dataSourceService.getById(dataValueModel.getBasicId());
         MysqlPool mysqlPool = new MysqlPool(dataSource);
         mysqlPool.init();
@@ -374,7 +302,7 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
     @Override
     public Integer getTableValueTotal(DataValueModel dataValueModel) {
         MysqlQuery mysqlQuery = new MysqlQuery();
-        mysqlQuery.setDataSourceId(String.valueOf(dataValueModel.getBasicId()));
+        mysqlQuery.setDataSourceId(dataValueModel.getBasicId());
         DataSource dataSource = dataSourceService.getById(dataValueModel.getBasicId());
         MysqlPool mysqlPool = new MysqlPool(dataSource);
         mysqlPool.init();
@@ -390,6 +318,31 @@ public class DataValueServiceImpl extends ServiceImpl<DataValueMapper, DataValue
         return 0;
     }
 
+    /**
+     * 根据基础表ID和SQL语句查询数据
+     * @param dataValueModel 查询条件
+     * @return 数据列表
+     */
+    @Override
+    public DataValue[][] findTableValueByTable(DataValueModel dataValueModel) {
+        MysqlQuery mysqlQuery = new MysqlQuery();
+        // 设置数据接入的ID
+        mysqlQuery.setDataSourceId(dataValueModel.getBasicId());
+        // 设置接到的sql语句
+        mysqlQuery.setSql(dataValueModel.getSql());
+        // 设置二维数组的第一个参数
+        mysqlQuery.setOne(dataValueModel.getOne());
+        // 设置二维数组的第二个参数
+        mysqlQuery.setTwo(dataValueModel.getTwo());
+        BaseQueryHandler.set(mysqlQuery);
+        // 创建数据源对象
+        MysqlDataSource mysqlDataSource = new MysqlDataSource();
+        // 获取数据
+        DataValue[][] rows = mysqlDataSource.getRows();
+        //返回查询的结果
+        return rows;
+    }
+
 
 }