commit

xpc

2024-02-27 d552ef9f4ea7ea8500f40195517387a6ab865f9a

 product-server-datasource/src/main/java/com/product/datasource/connection/ConnectionManager.java

@@ -1,5 +1,9 @@
package com.product.datasource.connection;

import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.NumberUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.google.common.collect.Maps;
import com.product.common.lang.StringUtils;
import com.product.core.config.Global;
import com.product.core.exception.BaseException;
@@ -9,7 +13,10 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author cheng
@@ -18,237 +25,297 @@
 */
public class ConnectionManager {

    static Logger log = LoggerFactory.getLogger(ConnectionManager.class);
   static Logger log = LoggerFactory.getLogger(ConnectionManager.class);

    public static void main(String[] args) {
        DataBaseEntity DataBaseEntity = new DataBaseEntity(DataBaseType.MYSQL.getValue());
        DataBaseEntity.setDbName("product_db_v2.0.0");
        DataBaseEntity.setUserName("root");
        DataBaseEntity.setIp("127.0.0.1");
        DataBaseEntity.setPassWord("root123");
        DataBaseEntity.setPort("3306");
        DataBaseEntity.setCustomParams("&serverTimezone=Asia/Shanghai");
        getConnection(DataBaseEntity);
    }
   private static final Map<String, DruidDataSource> DB_DRUID_DATA_SOURCE_MAP = new ConcurrentHashMap<>();

    /**
     * 获取系统连接
     *
     * @return
     */
    public static Connection getSystemConnection() throws BaseException {
        String url = Global.getSystemConfig("data.source.url", "");
        String userName = Global.getSystemConfig("data.source.user", "");
        String userPassword = Global.getSystemConfig("data.source.password", "");
        //批量提交参数是否开启
        if (url.indexOf("rewriteBatchedStatements=true") == -1) {
            url += "&rewriteBatchedStatements=true";
        }
        DataBaseEntity DataBaseEntity = new DataBaseEntity(DataBaseType.MYSQL.getValue());
        DataBaseEntity.setUserName(userName);
        DataBaseEntity.setPassWord(userPassword);
        return getConnection(url, DataBaseEntity);
    }
   public static void main(String[] args) {
      DataBaseEntity DataBaseEntity = new DataBaseEntity(DataBaseType.MYSQL.getValue());
      DataBaseEntity.setDbName("product_db_v2.0.0");
      DataBaseEntity.setUserName("root");
      DataBaseEntity.setIp("127.0.0.1");
      DataBaseEntity.setPassWord("root123");
      DataBaseEntity.setPort("3306");
      DataBaseEntity.setCustomParams("&serverTimezone=Asia/Shanghai");
      getConnection(DataBaseEntity);
   }

   /**
    * 获取系统连接
    *
    * @return
    */
   public static Connection getSystemConnection() throws BaseException {
      String url = Global.getSystemConfig("data.source.url", "");
      String userName = Global.getSystemConfig("data.source.user", "");
      String userPassword = Global.getSystemConfig("data.source.password", "");
      //批量提交参数是否开启
      if (url.indexOf("rewriteBatchedStatements=true") == -1) {
         url += "&rewriteBatchedStatements=true";
      }
      DataBaseEntity DataBaseEntity = new DataBaseEntity(DataBaseType.MYSQL.getValue());
      DataBaseEntity.setUserName(userName);
      DataBaseEntity.setPassWord(userPassword);
      return getConnection(url, DataBaseEntity);
   }


    /**
     * 获取连接
     *
     * @param dbe
     * @return
     * @throws BaseException
     */
    public static Connection getConnection(DataBaseEntity dbe) throws BaseException {
   /**
    * 获取连接
    *
    * @param dbe
    * @return
    * @throws BaseException
    */
   public static Connection getConnection(DataBaseEntity dbe) throws BaseException {

        try {
            if (dbe.getDataBaseType() == null) {
                throw new BaseException(ErrorCode.UNKNOWN_DATABASE_TYPE);
            }
            //装载驱动
            Class.forName(dbe.getDataBaseType().getDriver());
            if (DataBaseType.MYSQL.equals(dbe.getDataBaseType())) {
                // MYSQL
                return getMysqlConnection(dbe);
            } else if (DataBaseType.ORACLE.equals(dbe.getDataBaseType())) {
                // ORACLE
                return getOracleConnection(dbe);
            } else if (DataBaseType.SQLSERVER.equals(dbe.getDataBaseType())) {
                // SQLSERVER
                return getSqlServerConnection(dbe);
            } else if (DataBaseType.INFORMIX.equals(dbe.getDataBaseType())) {
                // INFORMIX
                return getInformixConnection(dbe);
            } else if (DataBaseType.PSQL.equals(dbe.getDataBaseType())) {
                return getPSQLConnection(dbe);
            }
            throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
        } catch (BaseException e) {
            e.printStackTrace();
            throw e;
        } catch (Exception e) {
            e.printStackTrace();
            throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
        }
    }
      try {
         if (dbe.getDataBaseType() == null) {
            throw new BaseException(ErrorCode.UNKNOWN_DATABASE_TYPE);
         }
         //装载驱动
         Class.forName(dbe.getDataBaseType().getDriver());
         if (DataBaseType.MYSQL.equals(dbe.getDataBaseType())) {
            // MYSQL
            return getMysqlConnection(dbe);
         } else if (DataBaseType.ORACLE.equals(dbe.getDataBaseType())) {
            // ORACLE
            return getOracleConnection(dbe);
         } else if (DataBaseType.SQLSERVER.equals(dbe.getDataBaseType())) {
            // SQLSERVER
            return getSqlServerConnection(dbe);
         } else if (DataBaseType.INFORMIX.equals(dbe.getDataBaseType())) {
            // INFORMIX
            return getInformixConnection(dbe);
         } else if (DataBaseType.PSQL.equals(dbe.getDataBaseType())) {
            return getPSQLConnection(dbe);
         }
         throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
      } catch (BaseException e) {
         e.printStackTrace();
         throw e;
      } catch (Exception e) {
         e.printStackTrace();
         throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
      }
   }

    public static Connection getPSQLConnection(DataBaseEntity dbe) {
        String templateUrl = "jdbc:postgresql://%s:%s/%s%s";
        //替换URL模板
        String url = String.format(templateUrl, dbe.getIp(), dbe.getPort(), dbe.getDbName(), dbe.getCustomParams());
        return getConnection(url, dbe);
    }
   public static Connection getPSQLConnection(DataBaseEntity dbe) {
      String templateUrl = "jdbc:postgresql://%s:%s/%s%s";
      //替换URL模板
      String url = String.format(templateUrl, dbe.getIp(), dbe.getPort(), dbe.getDbName(), dbe.getCustomParams());
      return getConnection(url, dbe);
   }

    public static long getConnectionSystemTime(Connection connection, DataBaseType dbType) throws BaseException {
        String queryTimeSql;
        if (DataBaseType.MYSQL.equals(dbType)) {
            queryTimeSql = "select current_timestamp(3) `timestamp` ";
        } else if (DataBaseType.ORACLE.equals(dbType)) {
            queryTimeSql = "select systimestamp  as timestamp from dual";
        }
   public static long getConnectionSystemTime(Connection connection, DataBaseType dbType) throws BaseException {
      String queryTimeSql;
      if (DataBaseType.MYSQL.equals(dbType)) {
         queryTimeSql = "select current_timestamp(3) `timestamp` ";
      } else if (DataBaseType.ORACLE.equals(dbType)) {
         queryTimeSql = "select systimestamp  as timestamp from dual";
      }
//        else if (DriverTypeEnum.SqlServer.equals(dbType)) {
//            queryTimeSql = "SELECT CONVERT(BIGINT,DATEDIFF(MI,'1970-01-01 00:00:00.000', GETUTCDATE())) * 60000 + DATEPART(S,GETUTCDATE()) * 1000 + DATEPART(MS, GETUTCDATE()) as timestamp";
//        } else if (DriverTypeEnum.Informix.equals(dbType)) {
//            queryTimeSql = "select current as timestamp from sysmaster:sysshmvals";
//        }
        else {
            throw new BaseException(ErrorCode.GET_CONNECTION_TIME_FAIL);
        }
        try (PreparedStatement pst = connection.prepareStatement(queryTimeSql); ResultSet resultSet = pst.executeQuery();) {
            while (resultSet.next()) {
                final Object timestamp = resultSet.getObject("timestamp");
                if (timestamp != null) {
                    if (timestamp instanceof Time) {
                        return ((Time) timestamp).getTime();
                    } else if (timestamp instanceof Timestamp) {
                        return ((Timestamp) timestamp).getTime();
                    } else {
                        return resultSet.getTimestamp("timestamp").getTime();
                    }
                }
                return resultSet.getLong("timestamp");
            }
            throw new BaseException(ErrorCode.GET_CONNECTION_TIME_FAIL);
        } catch (BaseException e) {
            throw e;
        } catch (Exception e) {
            e.printStackTrace();
            throw new BaseException(ErrorCode.GET_CONNECTION_TIME_FAIL);
        }
    }
      else {
         throw new BaseException(ErrorCode.GET_CONNECTION_TIME_FAIL);
      }
      try (PreparedStatement pst = connection.prepareStatement(queryTimeSql); ResultSet resultSet = pst.executeQuery();) {
         while (resultSet.next()) {
            final Object timestamp = resultSet.getObject("timestamp");
            if (timestamp != null) {
               if (timestamp instanceof Time) {
                  return ((Time) timestamp).getTime();
               } else if (timestamp instanceof Timestamp) {
                  return ((Timestamp) timestamp).getTime();
               } else {
                  return resultSet.getTimestamp("timestamp").getTime();
               }
            }
            return resultSet.getLong("timestamp");
         }
         throw new BaseException(ErrorCode.GET_CONNECTION_TIME_FAIL);
      } catch (BaseException e) {
         throw e;
      } catch (Exception e) {
         e.printStackTrace();
         throw new BaseException(ErrorCode.GET_CONNECTION_TIME_FAIL);
      }
   }


    /**
     * 获取MySql连接
     *
     * @param dbe
     * @return
     * @throws SQLException
     */
    private static Connection getMysqlConnection(DataBaseEntity dbe) {
        String templateUrl = "jdbc:mysql://%s:%s/%s%s";
        //替换URL模板
        String url = String.format(templateUrl, dbe.getIp(), dbe.getPort(), dbe.getDbName(), dbe.getCustomParams());
        return getConnection(url, dbe);
    }
   /**
    * 获取MySql连接
    *
    * @param dbe
    * @return
    * @throws SQLException
    */
   private static Connection getMysqlConnection(DataBaseEntity dbe) {
      String templateUrl = "jdbc:mysql://%s:%s/%s%s";
      //替换URL模板
      String url = String.format(templateUrl, dbe.getIp(), dbe.getPort(), dbe.getDbName(), dbe.getCustomParams());
      return getConnection(url, dbe);
   }

    /**
     * 获取Sql Server连接
     *
     * @param dbe
     * @return
     * @throws SQLException
     */
    private static Connection getSqlServerConnection(DataBaseEntity dbe) {
        String dbInstance = dbe.getDbInstance();
        String url;
        if (StringUtils.isEmpty(dbInstance)) {
            url = String.format("jdbc:sqlserver://%s:%s;databaseName=%s;", dbe.getIp(), dbe.getPort(), dbe.getDbName());
        } else {
            // 有实例名
            url = String.format("jdbc:sqlserver://%s:%s;instanceName=%s;databaseName=%s;", dbe.getIp(), dbe.getPort(), dbe.getDbInstance(), dbe.getDbName());
        }
        return getConnection(url, dbe);
    }
   /**
    * 获取Sql Server连接
    *
    * @param dbe
    * @return
    * @throws SQLException
    */
   private static Connection getSqlServerConnection(DataBaseEntity dbe) {
      String dbInstance = dbe.getDbInstance();
      String url;
      if (StringUtils.isEmpty(dbInstance)) {
         url = String.format("jdbc:sqlserver://%s:%s;databaseName=%s;", dbe.getIp(), dbe.getPort(), dbe.getDbName());
      } else {
         // 有实例名
         url = String.format("jdbc:sqlserver://%s:%s;instanceName=%s;databaseName=%s;", dbe.getIp(), dbe.getPort(), dbe.getDbInstance(), dbe.getDbName());
      }
      return getConnection(url, dbe);
   }

    /**
     * 获取Oracle连接
     *
     * @param dbe
     * @return
     * @throws SQLException
     */
    private static Connection getOracleConnection(DataBaseEntity dbe) {
        String url;
        if (!StringUtils.isEmpty(dbe.getSid())) {
            url = String.format("jdbc:oracle:thin:@%s:%s:%s", dbe.getIp(), dbe.getPort(), dbe.getSid());
        } else if (!StringUtils.isEmpty(dbe.getServerName())) {
            url = String.format("jdbc:oracle:thin:@//%s:%s/%s", dbe.getIp(), dbe.getPort(), dbe.getServerName());
        } else {
            throw new BaseException(ErrorCode.GET_ORACLE_SID_SERVERNAME_EMPTY);
        }
        return getConnection(url, dbe);
    }
   /**
    * 获取Oracle连接
    *
    * @param dbe
    * @return
    * @throws SQLException
    */
   private static Connection getOracleConnection(DataBaseEntity dbe) {
      String url;
      String[] params = {dbe.getIp(), dbe.getPort(), null};
      if (!StringUtils.isEmpty(dbe.getSid())) {
         url = String.format("jdbc:oracle:thin:@%s:%s:%s", dbe.getIp(), dbe.getPort(), dbe.getSid());
         params[2] = dbe.getSid();
      } else if (!StringUtils.isEmpty(dbe.getServerName())) {
         url = String.format("jdbc:oracle:thin:@//%s:%s/%s", dbe.getIp(), dbe.getPort(), dbe.getServerName());
         params[2] = dbe.getServerName();
      } else {
         throw new BaseException(ErrorCode.GET_ORACLE_SID_SERVERNAME_EMPTY);
      }
      Boolean enabling = Global.getPropertyToBoolean("data.system.oracle.connection-pool.enabling", "false");
      if (enabling) {
         synchronized (url.intern()) {
            DruidDataSource druidDataSource = DB_DRUID_DATA_SOURCE_MAP.get(ArrayUtil.join(params, ","));
            if (druidDataSource == null || druidDataSource.isClosed()) {
               druidDataSource = new DruidDataSource();
               druidDataSource.setUrl(url);
               druidDataSource.setUsername(dbe.getUserName());
               druidDataSource.setPassword(dbe.getPassWord());
               // 初始化时建立物理连接的个数
               druidDataSource.setInitialSize(getProperty("data.system.oracle.connection-pool.initial-size", 20));
               // 最大活动连接数
               druidDataSource.setMaxActive(getProperty("data.system.oracle.connection-pool.max-active", 100));
               // 最小空闲连接数
               druidDataSource.setMinIdle(getProperty("data.system.oracle.connection-pool.min-idle", 30));
               // 校验查询语句
               druidDataSource.setValidationQuery(dbe.getDataBaseType().getValidationQuery());
               // 当连接空闲时是否测试连接有效性
               druidDataSource.setTestWhileIdle(true);
               // 两次空闲连接清除之间的时间间隔 12小时
               druidDataSource.setTimeBetweenEvictionRunsMillis(43200000L);
               druidDataSource.setMaxWait(60000L);
               druidDataSource.setPoolPreparedStatements(true);
               druidDataSource.setMaxPoolPreparedStatementPerConnectionSize(100);
               druidDataSource.setRemoveAbandoned(true);
               //半小时强制归还连接
               druidDataSource.setRemoveAbandonedTimeout(60 * 30);
               DB_DRUID_DATA_SOURCE_MAP.put(ArrayUtil.join(params, ","), druidDataSource);
            }
            try {
               //获获取连接
               Connection connection = druidDataSource.getConnection();
               //测试连接
               if (!connectionValidity(connection, dbe.getDataBaseType())) {
                  throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
               }
               return connection;
            } catch (BaseException e) {
               throw e;
            } catch (Exception e) {
               e.printStackTrace();
               log.error("获取链接失败", e);
               log.error(url);
               throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
            }
         }
      }
      return getConnection(url, dbe);
   }

    /**
     * 获取Informix 连接
     *
     * @param dbe
     * @return
     * @throws SQLException
     */
    private static Connection getInformixConnection(DataBaseEntity dbe) throws BaseException {
        String url = String.format("jdbc:informix-sqli://%s:%s/%s:INFORMIXSERVER=%s;%s", dbe.getIp(), dbe.getPort(), dbe.getDbName(), dbe.getDbInstance(), dbe.getCustomParams());
        return getConnection(url, dbe);
    }
   private static int getProperty(String key, int defaultValue) {
      String systemConfig = Global.getSystemConfig(key, defaultValue + "");
      if (!NumberUtil.isNumber(systemConfig)) {
         return defaultValue;
      }
      return NumberUtil.parseInt(systemConfig);
   }

    /**
     * 获取连接
     *
     * @param url jdbc连接url
     * @param dbe 数据库基本数据
     * @return
     * @throws SQLException
     */
    private static Connection getConnection(String url, DataBaseEntity dbe) throws BaseException {
        try {
            //获获取连接
            Connection connection = DriverManager.getConnection(url, dbe.getUserName(), dbe.getPassWord());
            //测试连接
            if (!connectionValidity(connection, dbe.getDataBaseType())) {
                throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
            }
   /**
    * 获取Informix 连接
    *
    * @param dbe
    * @return
    * @throws SQLException
    */
   private static Connection getInformixConnection(DataBaseEntity dbe) throws BaseException {
      String url = String.format("jdbc:informix-sqli://%s:%s/%s:INFORMIXSERVER=%s;%s", dbe.getIp(), dbe.getPort(), dbe.getDbName(), dbe.getDbInstance(), dbe.getCustomParams());
      return getConnection(url, dbe);
   }

            return connection;
        } catch (BaseException e) {
            throw e;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("获取链接失败", e);
            log.error(url);
            throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
        }
    }
   /**
    * 获取连接
    *
    * @param url jdbc连接url
    * @param dbe 数据库基本数据
    * @return
    * @throws SQLException
    */
   private static Connection getConnection(String url, DataBaseEntity dbe) throws BaseException {
      try {
         //获获取连接
         Connection connection = DriverManager.getConnection(url, dbe.getUserName(), dbe.getPassWord());
         //测试连接
         if (!connectionValidity(connection, dbe.getDataBaseType())) {
            throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
         }

    /**
     * 验证连接有效性
     *
     * @param connection
     * @param DriverTypeEnum
     * @return
     */
    public static boolean connectionValidity(Connection connection, DataBaseType DriverTypeEnum) {
        try {
            String validationQuery = DriverTypeEnum.getValidationQuery();
            try (PreparedStatement pst = connection.prepareStatement(validationQuery)) {
                boolean execute = pst.execute();
                return execute;
            }
        } catch (Exception e) {
            log.error("验证连接有效性失败", e);
            return false;
        }
         return connection;
      } catch (BaseException e) {
         throw e;
      } catch (Exception e) {
         e.printStackTrace();
         log.error("获取链接失败", e);
         log.error(url);
         throw new BaseException(ErrorCode.GET_CONNECTION_FAIL);
      }
   }

    }
   /**
    * 验证连接有效性
    *
    * @param connection
    * @param DriverTypeEnum
    * @return
    */
   public static boolean connectionValidity(Connection connection, DataBaseType DriverTypeEnum) {
      try {
         String validationQuery = DriverTypeEnum.getValidationQuery();
         try (PreparedStatement pst = connection.prepareStatement(validationQuery)) {
            boolean execute = pst.execute();
            return execute;
         }
      } catch (Exception e) {
         log.error("验证连接有效性失败", e);
         return false;
      }

   }


}