添加连接数据池和任务sql查询数量

cloud-etl-server/src/main/java/com/muyu/cloud/etl/MuYuEtlApplication.java

@@ -13,6 +13,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
 public class MuYuEtlApplication {
     public static void main (String[] args) {
         SpringApplication.run(MuYuEtlApplication.class, args);
+
+//        Thread t1 = new Thread();
+//        Thread t2 = new Thread();
+//
+//        t1.start();
+//        t2.start();
+
+
     }
 
 }

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

@@ -31,6 +31,7 @@ public class DataValueController {
     }
 
 
+
     //添加数据库
     //根据值添加的表中
     @PostMapping("/addTable")
@@ -39,6 +40,19 @@ public class DataValueController {
          return Result.success(i);
     }
 
+    //根据任务提供的sql 查询数量
+    @PostMapping("/findCount")
+    public Integer findCount(@RequestParam("basicId") Long basicId,@RequestParam("sql") String sql) {
+        return dataValueService.findCount(basicId,sql);
+    }
+
+
+
+
+
+
+
+
 
 
 

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

@@ -12,7 +12,6 @@ import com.muyu.common.system.domain.SysUser;
 import com.muyu.domain.Structure;
 import com.muyu.domain.TableInfo;
 import com.muyu.domain.rep.TableInfoResp;
-import jakarta.servlet.http.HttpServletRequest;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
@@ -33,10 +32,8 @@ public class TableInfoController {
     //添加授权和修改授权
     @Autowired
     private AccreditService accreditService;
-    @Autowired
-    private HttpServletRequest request;
-
 
+    //查询所有的数据进行
     @GetMapping("/findTableInfoList")
     public Result findByTableName() {
         List<TableInfo> list = tableInfoService.list();
@@ -121,7 +118,6 @@ public class TableInfoController {
         }
 
 
-
         List<TableInfoResp> tableInfoResps = tableInfoHashSet.stream()
                 .filter(tableInfo ->
                         tableInfo.getParentId()== 0).map(tableInfo -> {
@@ -145,5 +141,4 @@ public class TableInfoController {
 
 
 
-
 }

cloud-etl-server/src/main/java/com/muyu/cloud/etl/mysql/BaseConfig.java

@@ -0,0 +1,17 @@
+package com.muyu.cloud.etl.mysql;
+
+public class BaseConfig {
+
+    /**
+     * mysql连接的前缀
+     */
+    public static  final  String MYSQLJDBCPRO="jdbc:mysql://";
+
+    public static void driver(String driverName){
+        try {
+            Class.forName(driverName);
+        } catch (ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/mysql/BasePool.java

@@ -0,0 +1,15 @@
+package com.muyu.cloud.etl.mysql;
+
+/**
+ * 提供一个连接池的准则
+ */
+public interface BasePool<T> {
+    //初始化
+    public void init();
+    //获取连接
+    public T getConn();
+    //归还连接
+    public void replease(T conn);
+    //创建连接
+    public T createConn();
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/mysql/MysqlConnException.java

@@ -0,0 +1,8 @@
+package com.muyu.cloud.etl.mysql;
+
+public class MysqlConnException extends RuntimeException{
+
+    public MysqlConnException(String message) {
+        super(message);
+    }
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/mysql/MysqlPool.java

@@ -0,0 +1,119 @@
+package com.muyu.cloud.etl.mysql;
+
+import com.muyu.cloud.etl.mysql.config.MysqlPoolConfig;
+import lombok.extern.log4j.Log4j2;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * mysql连接池信息
+ */
+@Log4j2
+public class MysqlPool implements BasePool<Connection> {
+
+
+    /**
+     * 等待(空闲队列) 基础队列
+     */
+    private Queue<Connection>  mysqlConneQueue = null;
+    /**
+     * 活动队列
+     */
+    private Queue<Connection> activeMysqlConnQueue = null;
+
+    /**
+     * 记录队列的连接总数
+     */
+    private AtomicInteger count=new AtomicInteger();
+
+    //mysql基础配置
+    public MysqlPoolConfig mysqlPoolConfig;
+    //进行实例化连接池 加载驱动
+    public MysqlPool(MysqlPoolConfig mysqlPoolConfig) {
+        this.mysqlPoolConfig = mysqlPoolConfig;
+        BaseConfig.driver(mysqlPoolConfig.getDriverName());
+        log.info("MySQL连接池实例化完成");
+    }
+
+    //进行初始化连接池
+    @Override
+    public void init() {
+        int maxTotal = this.mysqlPoolConfig.getMaxTotal();
+        int initTotal = this.mysqlPoolConfig.getInitTotal();
+        this.mysqlConneQueue=new LinkedBlockingQueue<Connection>(maxTotal);
+        this.activeMysqlConnQueue=new LinkedBlockingQueue<Connection>(maxTotal);
+        for (int i = 0; i < initTotal; i++) {
+            this.mysqlConneQueue.offer(createConn());
+            count.incrementAndGet();
+        }
+        log.info("MYSQL连接池初始化完成");
+
+    }
+
+    /**
+     * 1. 获取当前的毫秒，用来计算获取连接是否超时
+     * 2. 判断队列当中是否还有空闲连接
+     * 3. 如果有链接我们获取连接进行返回
+     * 4. 如果没有 判断是否达到最大的连接数
+     * 5. 如果达到最大连接数量进行等待（自旋）
+     * 6. 如果没有达到最大连接数量进行创建连接
+     * 7. 放到使用队列当中，染回在进行返回
+     * 8.如果连接超时的话则进行抛出异常
+     * @return
+     */
+    @Override
+    public Connection getConn() {
+        long startTime = System.currentTimeMillis();
+
+        //从空闲队列当中取出放入活动队列
+        Connection conn = this.mysqlConneQueue.poll();
+        if(conn!=null){
+            this.activeMysqlConnQueue.offer(conn);
+            return conn;
+        }
+        //如果当前的连接数量小于最大的连接数量的时候创建新的连接0000.
+        if(count.get()>this.mysqlPoolConfig.getMaxTotal()){
+            Connection mysqlConn = createConn();
+            this.activeMysqlConnQueue.offer(mysqlConn);
+            count.incrementAndGet();
+            return mysqlConn;
+        }
+
+        if((System.currentTimeMillis()-startTime)>this.mysqlPoolConfig.getMaxWaitTime()){
+                throw new MysqlConnException("连接超时");
+        }
+
+        return null;
+    }
+    //归还连接
+    @Override
+    public void replease(Connection conn) {
+        //删除活动队列当中的 连接
+        if(this.mysqlConneQueue.remove(conn)){
+            //把连接放到空闲队列当中
+            this.mysqlConneQueue.offer(conn);
+        }
+
+
+    }
+    //获取mysql连接信息
+    public Connection createConn() {
+        String url = this.mysqlPoolConfig.getUrl();
+        String userName = this.mysqlPoolConfig.getUserName();
+        String password = this.mysqlPoolConfig.getPassword();
+        Connection mysqlConn = null;
+        try {
+            mysqlConn = DriverManager.getConnection(url, userName, password);
+        } catch (SQLException e) {
+            throw new RuntimeException(e);
+        }
+        return mysqlConn;
+    }
+
+
+}

cloud-etl-server/src/main/java/com/muyu/cloud/etl/mysql/config/MysqlPoolConfig.java

@@ -0,0 +1,53 @@
+package com.muyu.cloud.etl.mysql.config;
+
+import com.muyu.cloud.etl.mysql.BaseConfig;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class MysqlPoolConfig {
+    //id
+    public String id;
+    //初始化连接数量
+    private int initTotal;
+
+    //最大连接数
+    private int maxTotal;
+
+    //最长等待时间  毫秒
+    private long maxWaitTime;
+
+    //驱动
+    private String driverName;
+
+    //url  ip 端口 数据库 编码
+    private String ip;
+    private int port;
+    private String databaseName;
+    private String param;
+
+    //用户名
+    public String userName;
+
+    //密码
+    private String password;
+
+    /**
+     *获取数据库连接
+     * @return
+     */
+    public String getUrl(){
+        StringBuilder urlSb = new StringBuilder(BaseConfig.MYSQLJDBCPRO);
+        urlSb.append(this.ip);
+        urlSb.append(":");
+        urlSb.append(this.port);
+        urlSb.append("/");
+        urlSb.append(this.databaseName);
+        urlSb.append("?");
+        urlSb.append(this.param);
+        return urlSb.toString();
+    }
+}

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

@@ -11,4 +11,6 @@ public interface DataValueService {
 
    Integer addTableDataValue(Long basicId, Long tableId, List<List<DataValue>> dataValue);
 
+   Integer findCount(Long basicId, String sql);
+
 }

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

@@ -156,10 +156,113 @@ public class DataValueServiceImpl implements DataValueService {
     }
 
     @Override
-    public Integer addTableDataValue(Long basicId, Long tableId, List<List<DataValue>> dataValue) {
+    public Integer addTableDataValue(Long basicId, Long tableId, List<List<DataValue>> dataValueList) {
+        // 获取数据源信息
         Source source = sourceService.getInfo(basicId);
         TableInfo tableInfo = tableInfoService.getById(tableId);
 
+// 提取连接信息
+        String host = source.getHost();
+        String port = source.getPort();
+        String databaseName = source.getDatabaseName();
+        String databaseType = source.getDatabaseType();
+// 构建JDBC URL
+        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);
+
+            // 获取表名
+            String tableName = tableInfo.getTableName();
+
+            // 遍历所有的数据值列表
+            PreparedStatement pstmt = null; // 声明预编译语句对象
+            for (List<DataValue> dataValues : dataValueList) {
+                // 初始化列名和占位符字符串
+                StringBuilder columns = new StringBuilder("(");
+                StringBuilder placeholders = new StringBuilder("VALUES (");
+                ArrayList<String> columnNames = new ArrayList<>();
+
+                // 遍历单个数据集中的所有键值对
+                for (DataValue dataValue : dataValues) {
+                    String key = dataValue.getKey();
+                    Object value = dataValue.getValue();
+
+                    // 构建列名部分
+                    columns.append(key).append(",");
+                    // 构建占位符部分
+                    placeholders.append("?,");
+                    // 收集列名用于后续处理
+                    columnNames.add(key);
+                }
+
+                // 移除最后一个逗号
+                columns.deleteCharAt(columns.length() - 1);
+                placeholders.deleteCharAt(placeholders.length() - 1);
+                // 关闭括号
+                columns.append(")");
+                placeholders.append(")");
+
+                // 构造完整的SQL语句
+                String sql = "INSERT INTO " + tableName + " " + columns.toString() + " " + placeholders.toString();
+
+                // 如果有旧的预编译语句，先关闭它
+                if (pstmt != null) pstmt.close();
+                // 准备新的预编译语句
+                pstmt = conn.prepareStatement(sql);
+
+                // 设置参数值
+                int index = 1;
+                for (DataValue dataValue : dataValues) {
+                    Object value = dataValue.getValue();
+                    if (value instanceof String) {
+                        pstmt.setString(index, (String) value);
+                    } else if (value instanceof Integer) {
+                        pstmt.setInt(index, (Integer) value);
+                    } else if (value instanceof Double) {
+                        pstmt.setDouble(index, (Double) value);
+                    } else if (value instanceof Date) {
+                        pstmt.setDate(index, new java.sql.Date(((Date) value).getTime()));
+                    } else {
+                        pstmt.setObject(index, value);
+                    }
+                    index++;
+                }
+
+                // 将语句添加到批处理中
+                pstmt.addBatch();
+            }
+
+            // 执行所有批处理语句
+            if (pstmt != null) {
+                pstmt.executeBatch();
+            }
+
+        } catch (SQLException e) {
+            // 抛出运行时异常，将原始的 SQLException 作为原因
+            throw new RuntimeException(e);
+        } finally {
+            // 在finally块中确保连接关闭
+            if (conn != null) {
+                try {
+                    conn.close();
+                } catch (SQLException e) {
+                    // 处理或记录关闭连接时发生的异常
+                }
+            }
+        }
+
+// 返回一个指示操作成功的值
+        return 1;
+    }
+
+    @Override
+    public Integer findCount(Long basicId, String sql) {
+        Source source = sourceService.getInfo(basicId);
         String host = source.getHost();
         String port = source.getPort();
         String databaseName = source.getDatabaseName();
@@ -167,20 +270,22 @@ public class DataValueServiceImpl implements DataValueService {
         String url = "jdbc:" + databaseType + "://" + host + ":" + port + "/" + databaseName + "?" + source.getConnectionParams();
         String user = source.getUsername();
         String password = source.getPassword();
-
-        String tableName = tableInfo.getTableName();
         Connection conn=null;
 
+        Integer string=0;
         try {
             conn = DriverManager.getConnection(url, user, password);
+            PreparedStatement prepareStatement = conn.prepareStatement(sql);
+            ResultSet resultSet = prepareStatement.executeQuery();
+            while (resultSet.next()){
+                string = resultSet.getInt(1);
+            }
+
+
         } catch (SQLException e) {
             throw new RuntimeException(e);
         }
 
-
+        return string;
-
-
-
-        return 0;
     }
 }

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

@@ -36,11 +36,13 @@ public class SourceServiceImpl extends ServiceImpl<SourceMapper, Source> impleme
     @Autowired
     private StructureService structureService;
 
+
     @Override
     public List<Source> selectSourceList(SourceReq sourceReq) {
        return sourceMapper.selectSourceList(sourceReq);
     }
 
+
     @Override
     public Source getInfo(Long id) {
         LambdaQueryWrapper<Source> lambdaQueryWrapper = new LambdaQueryWrapper<>();
@@ -271,8 +273,13 @@ public class SourceServiceImpl extends ServiceImpl<SourceMapper, Source> impleme
 
 
     public void syncData(Connection conn, String databaseName, TableInfo table) throws SQLException {
+
         ExecutorService threadPool = Executors.newCachedThreadPool();
 
+        long rowCount = 0;
+        int pageSize = 1000; // 每页1000条记录
+        int pageNumber = 1;
+
         PreparedStatement ps = conn.prepareStatement(
                 "    SELECT  " +
                         "    COLUMN_NAME , " +
@@ -295,6 +302,7 @@ public class SourceServiceImpl extends ServiceImpl<SourceMapper, Source> impleme
                         "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