From 5157792eb766826b7721438e0b11de05d760d882 Mon Sep 17 00:00:00 2001
From: gtl <2949451835@qq.com>
Date: Mon, 29 Apr 2024 22:40:42 +0800
Subject: [PATCH] =?UTF-8?q?feat(DataSourceServiceImpl):=E6=96=B0=E5=A2=9E?=
 =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BA=93=E8=BF=9E=E6=8E=A5=E6=B1=A0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 README.md                                     |   2 +-
 .../ruoyi-data_asset-common/pom.xml           |   6 +
 .../dataAsset/service/DataSourceService.java  |  10 --
 .../service/impl/DataSourceServiceImpl.java   | 107 ++++++++++++------
 4 files changed, 79 insertions(+), 46 deletions(-)

diff --git a/README.md b/README.md
index 2144055..06746e1 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@
 
 ### 简介
 
-本项目是 **ETL** 项目,主要功能是对接入的 **数据源 **进行数据的 **抽取**(Extract) > **清洗**(Cleaning) > **转换**(Transform) > **装载**(Load) 。
+本项目是 **ETL** 项目,主要功能是对接入的 **数据源** 进行数据的 **抽取**(Extract) > **清洗**(Cleaning) > **转换**(Transform) > **装载**(Load) 。
 
 ### 背景
 
diff --git a/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-common/pom.xml b/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-common/pom.xml
index 373d7d6..48877e3 100644
--- a/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-common/pom.xml
+++ b/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-common/pom.xml
@@ -24,5 +24,11 @@
             <groupId>com.muyu</groupId>
             <artifactId>muyu-common-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>druid</artifactId>
+            <version>1.2.20</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/DataSourceService.java b/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/DataSourceService.java
index 06e16af..5e8cb15 100644
--- a/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/DataSourceService.java
+++ b/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/DataSourceService.java
@@ -1,9 +1,7 @@
 package com.ruoyi.dataAsset.service;
 
-import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.List;
-
 import com.ruoyi.dataAsset.domain.DataSource;
 import com.baomidou.mybatisplus.extension.service.IService;
 
@@ -30,14 +28,6 @@ public interface DataSourceService extends IService<DataSource> {
      */
     boolean testConnect(Long id) throws SQLException;
 
-    /**
-     * 获取数据源的链接
-     * @param dataSource 数据源
-     * @return 连接
-     * @throws SQLException 异常
-     */
-    Connection getConnection(DataSource dataSource) throws SQLException;
-
     /**
      * 同步数据结构
      * @param dataSourceId 数据源编号
diff --git a/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/impl/DataSourceServiceImpl.java b/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/impl/DataSourceServiceImpl.java
index dd2d2dd..e98efa6 100644
--- a/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/impl/DataSourceServiceImpl.java
+++ b/ruoyi-modules/ruoyi-data_asset/ruoyi-data_asset-server/src/main/java/com/ruoyi/dataAsset/service/impl/DataSourceServiceImpl.java
@@ -4,6 +4,12 @@ import java.sql.*;
 import java.util.*;
 import java.util.Date;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.pool.DruidPooledConnection;
+import com.ruoyi.common.core.text.Convert;
 import com.ruoyi.common.core.utils.ObjUtils;
 import com.ruoyi.common.core.utils.StringUtils;
 import com.ruoyi.common.security.utils.SecurityUtils;
@@ -81,9 +87,9 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
     public boolean testConnect(Long id){
         boolean result=true;
         DataSource dataSource = this.getById(id);
-        Connection connection=null;
+        DruidDataSource druidDatasource=null;
         try {
-            connection = this.getConnection(dataSource);
+            druidDatasource = this.createDataSource(dataSource);
             //如果为不可用修改为可用
             if(dataSource.getStatus().equals("N")){
                 dataSource.setStatus("Y");
@@ -91,8 +97,8 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
                 dataSource.setUpdateBy(SecurityUtils.getUsername());
                 this.updateById(dataSource);
             }
-            connection.close();
-        } catch (SQLException e) {
+            druidDatasource.close();
+        } catch (Exception e) {
             //连接失败,如果为可用修改为不可用
             if(dataSource.getStatus().equals("Y")){
                 dataSource.setStatus("N");
@@ -106,22 +112,19 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
     }
 
     /**
-     * 获取数据源的连接
+     * 获取Druid连接池数据源
      * @param dataSource 数据源
-     * @return 数据源连接
+     * @return Druid数据源
      * @throws SQLException 异常
      */
-    @Override
-    public Connection getConnection(DataSource dataSource) throws SQLException {
+    public DruidDataSource createDataSource(DataSource dataSource) throws SQLException {
+        String driverType=null;
         String url="jdbc:"+dataSource.getDatabaseType()+dataSource.getIp()+":"+dataSource.getPort();
         if(dataSource.getDatabaseType().contains("mysql")){
+            driverType="com.mysql.cj.jdbc.Driver";
             url+="/"+dataSource.getDatabases();
         }else if(dataSource.getDatabaseType().contains("sqlserver")){
-            try {
-                Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver");
-            } catch (ClassNotFoundException e) {
-                throw new RuntimeException(e);
-            }
+            driverType="com.microsoft.sqlserver.jdbc.SQLServerDriver";
             url+=";DatabaseName="+dataSource.getDatabases()+";";
         }
         if(StringUtils.isNotEmpty(dataSource.getConnectionParameter())){
@@ -131,9 +134,39 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
                 url+=dataSource.getConnectionParameter()+";";
             }
         }
-        //连接超时时间5s
-        DriverManager.setLoginTimeout(5);
-        return DriverManager.getConnection(url, dataSource.getUsername(), dataSource.getPassword());
+        //手动创建数据源
+        DruidDataSource druidDataSource = new DruidDataSource();
+        druidDataSource.setName(dataSource.getId().toString());
+        druidDataSource.setDriverClassName(driverType);
+        druidDataSource.setUrl(url);
+        druidDataSource.setUsername(dataSource.getUsername());
+        druidDataSource.setPassword(dataSource.getPassword());
+        //设置连接配置
+        if(StringUtils.isNotEmpty(dataSource.getConnectionConfig())){
+            String[] configs = dataSource.getConnectionConfig().split(",");
+            for (String config : configs) {
+                String[] split = config.split("=");
+                if(config.contains("initNum")){
+                    //初始化连接池大小
+                    druidDataSource.setInitialSize(split.length>1&&StringUtils.isNotEmpty(split[1])?Convert.toInt(split[1]):5);
+                }else if(config.contains("maxNum")){
+                    //最大连接数
+                    druidDataSource.setMaxActive(split.length>1&&StringUtils.isNotEmpty(split[1])?Convert.toInt(split[1]):20);
+                }else if(config.contains("minIdle")){
+                    //最小空闲数
+                    druidDataSource.setMinIdle(split.length>1&&StringUtils.isNotEmpty(split[1])?Convert.toInt(split[1]):5);
+                }else {
+                    //获取连接最大等待时间，单位毫秒
+                    druidDataSource.setMaxWait(split.length>1&&StringUtils.isNotEmpty(split[1])?Convert.toInt(split[1]):6000);
+                }
+            }
+        }
+        String validationQuery = "select 1";
+        //申请连接时执行validationQuery检测连接是否有效，防止取到的连接不可用
+        druidDataSource.setTestOnBorrow(true);
+        druidDataSource.setValidationQuery(validationQuery);
+        druidDataSource.init();
+        return druidDataSource;
     }
 
     /**
@@ -155,36 +188,38 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
             sql="SELECT COUNT(*) tableTotal,SUM(b.rows) recordsTotal FROM sysobjects AS a \n" +
                     "INNER JOIN sysindexes AS b ON a.id = b.id WHERE ( a.type = 'u' ) AND ( b.indid IN ( 0, 1 ) );";
         }
-        Connection connection=null;
         try {
-            connection = this.getConnection(dataSource);
+            DruidDataSource druidDatasource = this.createDataSource(dataSource);
             //总的记录数,总的表数量
-            ResultSet result = connection.prepareStatement(sql).executeQuery();
+            ResultSet result = druidDatasource.getConnection().prepareStatement(sql).executeQuery();
             while (result.next()){
                 dataSource.setTableTotal(result.getInt("tableTotal"));
                 dataSource.setRecordsTotal(result.getInt("recordsTotal"));
             }
-            //异步同步表结构
-            CompletableFuture.runAsync(()->{
-                this.SynchronousTableStructure(dataSource);
-            });
+            ExecutorService executorService = Executors.newCachedThreadPool();
+            //异步同步表结构并等待所有任务结束
+            CompletableFuture.supplyAsync(() -> {
+                this.SynchronousTableStructure(druidDatasource, dataSource);
+                return "end";
+            }, executorService).get();
             //修改数据
             this.updateById(dataSource);
             result.close();
-            connection.close();
-        } catch (SQLException e) {
+            druidDatasource.close();
+        } catch (SQLException | InterruptedException | ExecutionException e) {
             throw new RuntimeException(e);
         }
     }
 
     /**
      * 同步表结构
+     * @param druidDatasource 数据源
      * @param dataSource 数据源
      */
-    public void SynchronousTableStructure(DataSource dataSource){
-        Connection connection=null;
+    public void SynchronousTableStructure(DruidDataSource druidDatasource,DataSource dataSource){
         try {
-            connection=getConnection(dataSource);
+            List<CompletableFuture<String>> runAsyncList=new ArrayList<>();
+            DruidPooledConnection connection = druidDatasource.getConnection();
             ResultSet tables = connection.getMetaData().getTables(connection.getCatalog(), "dbo", "%", new String[] { "TABLE" });
             while (tables.next()) {
                 String tableName = tables.getString("TABLE_NAME");
@@ -209,10 +244,13 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
                 //添加
                 tableInfoService.save(tableInfo);
                 //异步存储表数据
-                CompletableFuture.runAsync(()->{
-                    this.SynchronousColumnInfo(dataSource,tableName,tableInfo.getId());
-                });
+                runAsyncList.add(CompletableFuture.supplyAsync(() -> {
+                    this.SynchronousColumnInfo(druidDatasource, tableName, tableInfo.getId(), dataSource);
+                    return "end";
+                }));
             }
+            //等待所有任务结束
+            CompletableFuture.allOf(runAsyncList.toArray(CompletableFuture[]::new)).join();
             tables.close();
             connection.close();
         } catch (SQLException e) {
@@ -222,18 +260,17 @@ public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSou
 
     /**
      * 同步字段信息
-     * @param dataSource 数据源
+     * @param druidDatasource 数据源
      * @param tableName 表名
      * @param tableId 表编号
      */
-    public void SynchronousColumnInfo(DataSource dataSource,String tableName,Long tableId) {
+    public void SynchronousColumnInfo(DruidDataSource druidDatasource,String tableName,Long tableId,DataSource dataSource) {
         //声明字段信息空集合
         List<ColumnInfo> list=new ArrayList<>();
         //声明资产模型数据空集合
         List<AssetModelData> moduleDataList=new ArrayList<>();
-        Connection connection=null;
         try {
-            connection=getConnection(dataSource);
+            DruidPooledConnection connection = druidDatasource.getConnection();
             ResultSet columns = connection.getMetaData().getColumns(connection.getCatalog(), null, tableName, "%");
             ResultSet primaryKeys = connection.getMetaData().getPrimaryKeys(connection.getCatalog(), null, tableName);
             String primaryKeyName=null;