• JDBC连接池封装MaxCompute/Hive/Oracle/Mysql

    有时候需要连接第三方的各种数据源，总是要去写不同的代码，于是将MaxCompute, Hive, Oracle, Mysql等JDBC连接封装起来，只需要传入不同的参数即可创建一个不同类型的连接池。

    连接参数基础类封装

    封装了JDBC基础的连接参数，如果不需要这些属性可以继承该类，增加新的属性即可。

    1. @Data
    2. public class BaseJdbcConnParam implements Serializable {
    3.  
    4.     /**
    5.      * driver name
    6.      */
    7.     private String driverName;
    8.  
    9.     /**
    10.      * IP
    11.      */
    12.     private String ip;
    13.  
    14.     /**
    15.      * db server port
    16.      */
    17.     private Integer port;
    18.  
    19.     /**
    20.      * db name
    21.      */
    22.     private String dbName;
    23.  
    24.     /**
    25.      * db connection username
    26.      */
    27.     private String username;
    28.  
    29.     /**
    30.      * db connection password
    31.      */
    32.     private String password;
    33. }

    抽象连接工具类封装

    功能如下：

    • 1、构造函数：根据连接参数不同构建不同的连接对象
    • 2、构建具体的连接，子类实现buildConnection()
    • 3、获取连接，构建好之后直接获取getConnection()
    1. /**
    2.  * @Description 抽象连接工具类父类
    3.  * @Author itdl
    4.  * @Date 2022/08/15 09:54
    5.  */
    6. public abstract class AbstractConnUtil<P extends BaseJdbcConnParam> {
    7.  
    8.     /**
    9.      * connection params
    10.      */
    11.     protected final P connParam;
    12.  
    13.     /**
    14.      * jdbc connection object
    15.      */
    16.     protected final Connection connection;
    17.  
    18.  
    19.     /**
    20.      * 构造函数， 构造工具类对象
    21.      * @param connParam 连接参数
    22.      */
    23.     public AbstractConnUtil(P connParam) {
    24.         this.connParam = connParam;
    25.         this.connection = buildConnection();
    26.     }
    27.  
    28.  
    29.     /**
    30.      * 构建连接对象
    31.      * @return 连接对象
    32.      */
    33.     protected abstract Connection buildConnection();
    34.  
    35.  
    36.     /**
    37.      * 获取连接
    38.      */
    39.     public Connection getConnection() {
    40.         return connection;
    41.     }
    42. }

    连接池管理

    功能如下：

    • 1、根据不同的连接参数，和最大连接数去创建一个对应类型的连接池。
    • 2、获取连接方法，如果连接没有了，等待其他线程释放（最多等待十分钟）
    • 3、释放连接方法，将连接放回连接池，然后唤醒等待的线程
    • 4、关闭连接池所有的连接
    1. /**
    2.  * @Description 连接池管理
    3.  * @Author itdl
    4.  * @Date 2022/08/16 09:42
    5.  */
    6. @Slf4j
    7. public class DbConnPool<T extends BaseJdbcConnParam> {
    8.     /**
    9.      * 用于存放连接
    10.      */
    11.     private final LinkedList<Connection> connPool = new LinkedList<Connection>();
    12.  
    13.     /**
    14.      * 最大连接池数量
    15.      */
    16.     private final Integer maxPoolSize;
    17.  
    18.     private final T connParam;
    19.  
    20.     /**
    21.      * 构造函数
    22.      * @param connParam 连接参数
    23.      * @param maxPoolSize 连接池大小
    24.      */
    25.     public DbConnPool(T connParam, Integer maxPoolSize)  {
    26.         this.maxPoolSize = maxPoolSize;
    27.         this.connParam = connParam;
    28.         // 初始化连接池
    29.         for (int i = 0; i < maxPoolSize; i++) {
    30.             connPool.addLast(this.createConnection());
    31.         }
    32.     }
    33.  
    34.     /**
    35.      * 创建数据库连接
    36.      * @return 连接
    37.      */
    38.     private Connection createConnection() {
    39.         if (connParam instanceof OracleJdbcConnParam){
    40.             final OracleConnUtil util = new OracleConnUtil((OracleJdbcConnParam) connParam);
    41.             return util.getConnection();
    42.         }
    43.  
    44.         if (connParam instanceof HiveJdbcConnParam){
    45.             final HiveConnUtil util = new HiveConnUtil((HiveJdbcConnParam) connParam);
    46.             return util.getConnection();
    47.         }
    48.  
    49.         if (connParam instanceof MysqlJdbcConnParam){
    50.             final MysqlConnUtil util = new MysqlConnUtil((MysqlJdbcConnParam) connParam);
    51.             return util.getConnection();
    52.         }
    53.  
    54.         if (connParam instanceof MaxComputeJdbcConnParam){
    55.             final MaxComputeJdbcUtil util = new MaxComputeJdbcUtil((MaxComputeJdbcConnParam) connParam);
    56.             return util.getConnection();
    57.         }
    58.  
    59.         throw new BizException(ResultCode.CONN_TYPE_NOT_SUPPORT);
    60.     }
    61.  
    62.  
    63.     /**
    64.      * 获取连接
    65.      * @return 连接
    66.      */
    67.     public synchronized Connection getConnection(){
    68.         if (connPool.size() == 0){
    69. //            throw new BizException(ResultCode.CONN_POOL_EMPTY_ERR);
    70.             // 最长等待十分钟
    71.             try {
    72.                 log.info("==========连接池已经空了， 请等待其他线程释放==========");
    73.                 wait(10 * 60 * 1000);
    74.             } catch (InterruptedException e) {
    75.                 log.info("==========连接池已经空了， 等待了10分钟还没有释放，抛出异常==========");
    76.                 e.printStackTrace();
    77.                 throw new BizException(ResultCode.CONN_POOL_EMPTY_ERR);
    78.             }
    79.         }
    80.         // 去除最上面一个连接 如果没有连接了，将会抛出异常
    81.         return connPool.removeFirst();
    82.     }
    83.  
    84.     /**
    85.      * 用完后释放连接
    86.      * @param conn 要释放的连接
    87.      */
    88.     public synchronized void freeConnection(Connection conn){
    89.         // 通知连接已经释放
    90.         notifyAll();
    91.         this.connPool.addLast(conn);
    92.     }
    93.  
    94.     /**
    95.      * 关闭连接池
    96.      */
    97.     public synchronized void close(){
    98.         for (Connection connection : connPool) {
    99.             SqlUtil.close(connection);
    100.         }
    101.     }
    102.  
    103. }

    SQL操作工具类

    根据连接对象Connection和数据库房源，封装不同的sql执行。执行SQL核心功能封装。

    1. /**
    2.  * @Description SQL操作工具类
    3.  * @Author itdl
    4.  * @Date 2022/08/10 17:13
    5.  */
    6. @Slf4j
    7. public class SqlUtil {
    8.     /**查询mysql表注释sql*/
    9.     public static final String SELECT_TABLES_MYSQL = "select table_name, table_comment from information_schema.tables where TABLE_SCHEMA = '%s'";
    10.     /**查询MaxCompute表注释sql*/
    11.     public static final String SELECT_TABLES_MAX_COMPUTE = "select table_name, table_comment from information_schema.tables where TABLE_SCHEMA = '%s'";
    12.     /**查询oracle表注释sql*/
    13.     public static final String SELECT_TABLES_ORACLE = "SELECT t2.TABLE_NAME as table_name, t2.COMMENTS as table_comment FROM user_tables t1 inner join user_tab_comments t2 on t1.TABLE_NAME = t2.TABLE_NAME";
    14.     /**查询hive表注释sql, 先查询表名，根据表名获取建表语句，正则提取表注释*/
    15.     public static final String SELECT_TABLES_HIVE = "show tables";
    16.     public static final String SELECT_TABLES_2_HIVE = "describe extended %s";
    17.  
    18.     /**分页数量统计Mysql*/
    19.     private static final String SELECT_COUNT_MYSQL = "select count(1) from (%s) z";
    20.     /**分页数量统计MaxCompute*/
    21.     private static final String SELECT_COUNT_MAX_COMPUTE = "select count(1) from (%s) z;";
    22.     /**分页数量统计Hive*/
    23.     private static final String SELECT_COUNT_ORACLE = "select count(1) from (%s) z";
    24.     /**分页数量统计Oracle*/
    25.     private static final String SELECT_COUNT_HIVE = "select count(1) from (%s) z";
    26.     /**maxCompute开启全表扫描sql*/
    27.     private static final String FULL_SCAN_MAX_COMPUTE = "set odps.sql.allow.fullscan=true;";
    28.  
    29.     /**分页查询sql-Mysql*/
    30.     private static final String SELECT_PAGE_MYSQL = "select z.* from (%s) z limit %s, %s";
    31.     /**分页查询sql-MaxCompute*/
    32.     private static final String SELECT_PAGE_MAX_COMPUTE = "select z.* from (%s) z limit %s, %s;";
    33.     /**分页查询sql-Hive*/
    34.     private static final String SELECT_PAGE_HIVE = "select * from (select row_number() over () as row_num_01,u.* from (%s) u) mm where mm.row_num_01 between %s and %s";
    35.     /**分页查询sql-Oracle*/
    36.     private static final String SELECT_PAGE_ORACLE = "select * from (SELECT ROWNUM as row_num_01,z.* from (%s) z) h where h.row_num_01 > %s and h.row_num_01 <= %s";
    37.  
    38.     /**数据库连接*/
    39.     private final Connection connection;
    40.  
    41.     /**数据库方言*/
    42.     private final Integer dbDialect;
    43.  
    44.     /**支持的方言列表*/
    45.     private static final List<Integer> supportDbTypes =
    46.             Arrays.asList(DbDialectEnum.ORACLE.getCode(), DbDialectEnum.HIVE.getCode(), DbDialectEnum.MYSQL.getCode(), DbDialectEnum.MAX_COMPUTE.getCode());
    47.  
    48.     public SqlUtil(Connection connection, Integer dbDialect) {
    49.         if (!supportDbTypes.contains(dbDialect)){
    50.             throw new BizException(ResultCode.CONN_TYPE_NOT_SUPPORT);
    51.         }
    52.         this.connection = connection;
    53.         this.dbDialect = dbDialect;
    54.     }
    55.  
    56.  
    57.     /**
    58.      * 根据connection获取所有的表和对应的注释
    59.      */
    60.     public List<TableMetaInfo> getTables(String schemaName){
    61.         List<TableMetaInfo> result = new ArrayList<>();
    62.         String sql = "";
    63.         switch (this.dbDialect){
    64.             case 1:
    65.                 sql = SELECT_TABLES_ORACLE;
    66.                 break;
    67.             case 2:
    68.                 sql = SELECT_TABLES_HIVE;
    69.                 break;
    70.             case 3:
    71.                 if (StringUtils.isBlank(schemaName)){
    72.                     throw new BizException(ResultCode.SELECT_TABLES_SCHEMA_NOT_NULL_ERR);
    73.                 }
    74.                 sql = String.format(SELECT_TABLES_MYSQL, schemaName);
    75.                 break;
    76.             case 4:
    77.                 if (StringUtils.isBlank(schemaName)){
    78.                     throw new BizException(ResultCode.SELECT_TABLES_SCHEMA_NOT_NULL_ERR);
    79.                 }
    80.                 sql = String.format(SELECT_TABLES_MAX_COMPUTE, schemaName);
    81.             default:
    82.                 break;
    83.         }
    84.  
    85.         if (StringUtils.isBlank(sql)){
    86.             throw new BizException(ResultCode.CONN_TYPE_NOT_SUPPORT);
    87.         }
    88.  
    89.         // 执行SQL语句
    90.         final List<LinkedHashMap<String, Object>> resultMaps = querySql(sql);
    91.         if (ObjectUtils.isEmpty(resultMaps)){
    92.             return Lists.newArrayList();
    93.         }
    94.  
    95.         // hive单独处理
    96.         List<TableMetaInfo> result1 = getHiveTableMetaInfos(result, resultMaps);
    97.         if (result1 != null) return result1;
    98.  
    99.         // 转换结果
    100.         return resultMaps.stream().map(
    101.                 m->{
    102.                     final TableMetaInfo info = new TableMetaInfo();
    103.                     Object tableNameObj = m.get("table_name");
    104.                     String tableName = tableNameObj == null ? m.get("TABLE_NAME") == null ? "" : String.valueOf(m.get("TABLE_NAME")) : String.valueOf(tableNameObj);
    105.                     Object tableCommentObj = m.get("table_comment");
    106.                     String tableComment = tableCommentObj == null ? m.get("TABLE_COMMENT") == null ? "" : String.valueOf(m.get("TABLE_COMMENT")) : String.valueOf(tableCommentObj);
    107.                     info.setTableName(tableName);
    108.                     info.setComment(tableComment);
    109.                     return info;
    110.                 }
    111.         ).collect(Collectors.toList());
    112.     }
    113.  
    114.  
    115.     /**
    116.      * 根据schemeName，表名获取字段列表
    117.      * @param tableName 一般是数据库 oracle是用户名
    118.      */
    119.     public List<TableColumnMetaInfo> getColumnsByTableName(String tableName){
    120.         try {
    121.             List<TableColumnMetaInfo> list = new ArrayList<>();
    122.             final DatabaseMetaData metaData = connection.getMetaData();
    123.             final ResultSet columns = metaData.getColumns(null, null, tableName, null);
    124.             while (columns.next()){
    125.                 String columnName = columns.getString("COLUMN_NAME");
    126.                 String remarks = columns.getString("REMARKS");
    127.                 remarks = StringUtils.isBlank(remarks) ? "" : remarks;
    128.                 final TableColumnMetaInfo metaInfo = new TableColumnMetaInfo(tableName, columnName, remarks);
    129.                 list.add(metaInfo);
    130.             }
    131.             return list;
    132.         } catch (SQLException e) {
    133.             e.printStackTrace();
    134.             return Lists.newArrayList();
    135.         }
    136.     }
    137.  
    138.  
    139.     /**
    140.      * 执行sql查询
    141.      * @param querySql 查询sql
    142.      * @return List> 通过LinkedHashMap接受，序列化时可保证顺序一致
    143.      */
    144.     public List<LinkedHashMap<String, Object>> queryData(String querySql, boolean... fullScan){
    145.         Statement statement = null;
    146.         ResultSet resultSet = null;
    147.         try {
    148.             // 创建statement
    149.             statement = this.connection.createStatement();
    150.  
    151.             // 执行全表扫描sql
    152.             for (boolean b : fullScan) {
    153.                 if (b){
    154.                     statement.execute(FULL_SCAN_MAX_COMPUTE);
    155.                     break;
    156.                 }
    157.             }
    158.             // 执行查询语句
    159.             resultSet = statement.executeQuery(querySql);
    160.  
    161.             // 构建结果返回
    162.             return buildListMap(resultSet);
    163.         } catch (SQLException e) {
    164.             e.printStackTrace();
    165.             throw new BizException(ResultCode.SQL_EXEC_ERR);
    166.         } finally {
    167.             // 关闭resultSet, statement
    168.             close(resultSet, statement);
    169.         }
    170.     }
    171.  
    172.  
    173.     /**
    174.      * 执行sql查询
    175.      * @param querySql 查询sql
    176.      * @return List>
    177.      */
    178.     public List<LinkedHashMap<String, Object>> queryData(String querySql, Integer page, Integer size){
    179.         Statement statement = null;
    180.         ResultSet resultSet = null;
    181.         try {
    182.             // 1、替换分号
    183.             querySql = querySql.replaceAll(";", "");
    184.             // 创建statement
    185.             statement = this.connection.createStatement();
    186.             // 2、格式化SQL
    187.             int offset = (page - 1 ) * size;
    188.             String execSql = "";
    189.             switch (this.dbDialect){
    190.                 case 1:
    191.                     // oracle
    192.                     execSql = String.format(SELECT_PAGE_ORACLE, querySql, offset, size);
    193.                     break;
    194.                 case 2:
    195.                     // hive
    196.                     execSql = String.format(SELECT_PAGE_HIVE, querySql, offset, size);
    197.                     break;
    198.                 case 3:
    199.                     // mysql
    200.                     execSql = String.format(SELECT_PAGE_MYSQL, querySql, offset, size);
    201.                     break;
    202.                 case 4:
    203.                     // maxCompute
    204.                     execSql = String.format(SELECT_PAGE_MAX_COMPUTE, querySql, offset, size);
    205.                     break;
    206.                 default:
    207.                     break;
    208.             }
    209.             // maxCompute开启全表扫描
    210.             if (DbDialectEnum.MAX_COMPUTE.getCode().equals(this.dbDialect)){
    211.                 statement.execute(FULL_SCAN_MAX_COMPUTE);
    212.             }
    213.             log.info("=======>>>执行分页sql为：{}", execSql);
    214.             // 执行查询语句
    215.             resultSet = statement.executeQuery(execSql);
    216.             // 构建结果返回
    217.             return buildListMap(resultSet);
    218.         } catch (SQLException e) {
    219.             e.printStackTrace();
    220.             throw new BizException(ResultCode.SQL_EXEC_ERR);
    221.         } finally {
    222.             // 关闭resultSet, statement
    223.             close(resultSet, statement);
    224.         }
    225.     }
    226.  
    227.  
    228.     /**
    229.      * 执行分页查询
    230.      * @param querySql 分页查询sql
    231.      * @param page 页码 从1开始 第n页传n
    232.      * @param size 每页记录数
    233.      * @return 分页查询结果
    234.      */
    235.     public PageResult<LinkedHashMap<String, Object>> pageQueryMap(String querySql, Integer page, Integer size){
    236.         // 1、替换分号
    237.         querySql = querySql.replaceAll(";", "");
    238.         String countSql = "";
    239.         switch (this.dbDialect){
    240.             case 1:
    241.                 // oracle
    242.                 countSql = String.format(SELECT_COUNT_ORACLE, querySql);
    243.                 break;
    244.             case 2:
    245.                 // hive
    246.                 countSql = String.format(SELECT_COUNT_HIVE, querySql);
    247.                 break;
    248.             case 3:
    249.                 // mysql
    250.                 countSql = String.format(SELECT_COUNT_MYSQL, querySql);
    251.                 break;
    252.             case 4:
    253.                 // maxCompute
    254.                 countSql = String.format(SELECT_COUNT_MAX_COMPUTE, querySql);
    255.                 break;
    256.             default:
    257.                 break;
    258.         }
    259.         log.info("=======>>>执行分页统计总数sql为：{}", countSql);
    260.         // 查询总数
    261.         final List<LinkedHashMap<String, Object>> countMap = queryData(countSql, DbDialectEnum.MAX_COMPUTE.getCode().equals(this.dbDialect));
    262.         if (CollectionUtils.isEmpty(countMap)){
    263.             return new PageResult<>(0L, new ArrayList<>());
    264.         }
    265.  
    266.         long count = 0L;
    267.         for (Object value : countMap.get(0).values()) {
    268.             count = Long.parseLong(String.valueOf(value));
    269.         }
    270.  
    271.         if (count == 0){
    272.             return new PageResult<>(0L, new ArrayList<>());
    273.         }
    274.  
    275.         // 执行分页查询 开启全表扫描
    276.         final List<LinkedHashMap<String, Object>> resultList = queryData(querySql, page, size);
    278.         return new PageResult<>(count, resultList);
    279.     }
    280.  
    281.  
    282.     /**
    283.      * 执行分页查询
    284.      * @param querySql 分页查询sql
    285.      * @param page 页码 从1开始 第n页传n
    286.      * @param size 每页记录数
    287.      * @return 分页查询结果
    288.      */
    289.     public <T>PageResult<T> pageQuery(String querySql, Integer page, Integer size, Class<T> clazz){
    290.         final PageResult<LinkedHashMap<String, Object>> result = pageQueryMap(querySql, page, size);
    291.         List<T> rows = new ArrayList<>();
    292.         for (LinkedHashMap<String, Object> row : result.getRows()) {
    293.             final T t = JSONObject.parseObject(JSONObject.toJSONString(row), clazz);
    294.             rows.add(t);
    295.         }
    296.         return new PageResult<>(result.getTotal(), rows);
    297.     }
    298.  
    299.  
    300.  
    301.     /**
    302.      * 获取hive的表注释
    303.      * @param result 结果
    304.      * @param resultMaps show tables结果
    305.      * @return List
    306.      */
    307.     private List<TableMetaInfo> getHiveTableMetaInfos(List<TableMetaInfo> result, List<LinkedHashMap<String, Object>> resultMaps) {
    308.         if (dbDialect.equals(DbDialectEnum.HIVE.getCode())){
    309.             for (LinkedHashMap<String, Object> resultMap : resultMaps) {
    310.                 final String tabName = String.valueOf(resultMap.get("tab_name"));
    311.                 final String descTableCommentSql = String.format(SELECT_TABLES_2_HIVE, tabName);
    312.                 List<LinkedHashMap<String, Object>> resultMapsComments = querySql(descTableCommentSql);
    313. //                col_name -> Detailed Table Information
    314.                 String comments = resultMapsComments.stream()
    315.                         .filter(m -> "Detailed Table Information".equals(m.get("col_name")))
    316.                         .map(m -> String.valueOf(m.get("data_type"))).findFirst()
    317.                         .orElse("");
    318.                 comments = ReUtil.get("parameters:\\{(?!.*?\\().*transient_lastDdlTime.*?comment=(.*?)\\}", comments,1);
    319.                 if (StringUtils.isBlank(comments)) {
    320.                     comments = "";
    321.                 }
    322.                 if (comments.contains(",")){
    323.                     comments = comments.substring(0, comments.lastIndexOf(","));
    324.                 }
    325.                 result.add(new TableMetaInfo(tabName, comments));
    326.                 log.info("===========>>>获取表{}的注释成功:{}", tabName, comments);
    327.                 resultMapsComments.clear();
    328.             }
    329.             return result;
    330.         }
    331.         return null;
    332.     }
    333.  
    334.  
    335.     /**
    336.      * 执行SQL查询
    337.      * @param sql sql语句
    338.      * @return 数据列表，使用LinkedHashMap是为了防止HashMap序列化后导致顺序乱序
    339.      */
    340.     public List<LinkedHashMap<String, Object>> querySql(String sql){
    341.         // 执行sql
    342.         Statement statement = null;
    343.         ResultSet resultSet = null;
    344.         try {
    345.             statement = connection.createStatement();
    346.             resultSet = statement.executeQuery(sql);
    347.             return buildListMap(resultSet);
    348.         } catch (SQLException e) {
    349.             e.printStackTrace();
    350.             throw new BizException(ResultCode.SQL_EXEC_ERR);
    351.         }finally {
    352.             // 关闭
    353.             close(resultSet, statement);
    354.         }
    355.     }
    356.  
    357.  
    358.     /**
    359.      * 关闭对象 传入多个时注意顺序， 需要先关闭哪个就传在参数前面
    360.      * @param objs 对象动态数组
    361.      */
    362.     public static void close(Object ...objs){
    363.         if (objs == null || objs.length == 0){
    364.             return;
    365.         }
    366.  
    367.         for (Object obj : objs) {
    368.             if (obj instanceof Statement){
    369.                 try {
    370.                     ((Statement) obj).close();
    371.                 }catch (Exception e){
    372.                     e.printStackTrace();
    373.                 }
    374.             }
    375.  
    376.             if (obj instanceof ResultSet){
    377.                 try {
    378.                     ((ResultSet) obj).close();
    379.                 }catch (Exception e){
    380.                     e.printStackTrace();
    381.                 }
    382.             }
    383.  
    384.             if (obj instanceof Connection){
    385.                 try {
    386.                     ((Connection) obj).close();
    387.                 }catch (Exception e){
    388.                     e.printStackTrace();
    389.                 }
    390.             }
    391.         }
    392.     }
    393.  
    394.  
    395.     /**
    396.      * @Description 功能描述：将resultSet构造为List
    397.      * @Author itdl
    398.      * @Date 2022/4/18 21:13
    399.      * @Param {@link ResultSet} resultSet
    400.      * @Return {@link List < Map >}
    401.      **/
    402.     private List<LinkedHashMap<String, Object>> buildListMap(ResultSet resultSet) throws SQLException {
    403.         if (resultSet == null) {
    404.             return Lists.newArrayList();
    405.         }
    406.  
    407.         List<LinkedHashMap<String, Object>> resultList = new ArrayList<>();
    408.         // 获取元数据
    409.         ResultSetMetaData metaData = resultSet.getMetaData();
    410.         while (resultSet.next()) {
    411.             // 获取列数
    412.             int columnCount = metaData.getColumnCount();
    413.             LinkedHashMap<String, Object> map = new LinkedHashMap<>();
    414.             for (int i = 0; i < columnCount; i++) {
    415.                 String columnName = metaData.getColumnName(i + 1);
    416.                 // 过滤掉查询的结果包含序号的
    417.                 if("mm.row_num_01".equalsIgnoreCase(columnName)
    418.                         || "row_num_01".equalsIgnoreCase(columnName)){
    419.                     continue;
    420.                 }
    421.  
    422.                 // 去除hive查询结果的mm.别名前缀
    423.                 if (columnName.startsWith("mm.")){
    424.                     columnName = columnName.substring(columnName.indexOf(".") + 1);
    425.                 }
    426.  
    427.                 Object object = resultSet.getObject(columnName);
    428.  
    429.                 // maxCompute里面的空返回的是使用\n
    430.                 if ("\\N".equalsIgnoreCase(String.valueOf(object))) {
    431.                     map.put(columnName, "");
    432.                 } else {
    433.                     map.put(columnName, object);
    434.                 }
    435.             }
    436.  
    437.             resultList.add(map);
    438.         }
    439.         return resultList;
    440.     }
    441. }

    MaxCompute JDBC连接池封装

    MaxCompute 已经有了JDBC连接方式 也就是 odbc-jdbc, 最终能够获取一个Connection. 官方文档：https://help.aliyun.com/document_detail/161246.html

    封装MaxCompute JDBC连接参数

    1. /**
    2.  * @author itdl
    3.  * @description maxCompute使用JDBC的连接参数
    4.  * @date 2022/08/08 10:07
    5.  */
    6. @Data
    7. public class MaxComputeJdbcConnParam extends BaseJdbcConnParam{
    8.     /**阿里云accessId 相当于用户名 */
    9.     private String aliyunAccessId;
    10.     /**阿里云accessKey 相当于密码 */
    11.     private String aliyunAccessKey;
    12.     /** maxcompute_endpoint */
    13.     private String endpoint;
    14.     /**项目名称*/
    15.     private String projectName;
    16. }

    封装MaxCompute JDBC连接实现类

    就是实现父类AbstractConnUtil，实现抽象方法buildConnection

    1. /**
    2.  * @Description maxCompute JDBC连接实现
    3.  * @Author itdl
    4.  * @Date 2022/08/08 14:26
    5.  */
    6. @Slf4j
    7. public class MaxComputeJdbcUtil extends AbstractConnUtil<MaxComputeJdbcConnParam>{
    8.     /**JDBC 驱动名称*/
    9.     private static final String DRIVER_NAME = "com.aliyun.odps.jdbc.OdpsDriver";
    10.     /**
    11.      * 构造函数， 构造工具类对象
    12.      *
    13.      * @param connParam 连接参数
    14.      */
    15.     public MaxComputeJdbcUtil(MaxComputeJdbcConnParam connParam) {
    16.         super(connParam);
    17.     }
    18.  
    19.     @Override
    20.     protected Connection buildConnection() {
    21.         return buildConn();
    22.     }
    23.  
    24.     /**
    25.      * 创建连接
    26.      * @return 数据库连接
    27.      */
    28.     private Connection buildConn() {
    29.         try {
    30.             Class.forName(DRIVER_NAME);
    31.         } catch (ClassNotFoundException e) {
    32.             e.printStackTrace();
    33.             throw new BizException(ResultCode.MAX_COMPUTE_DRIVE_LOAD_ERR);
    34.         }
    35.  
    36.         try {
    37.             Properties dbProperties = new Properties();
    38.             dbProperties.put("user", connParam.getAliyunAccessId());
    39.             dbProperties.put("password", connParam.getAliyunAccessKey());
    40.             dbProperties.put("remarks", "true");
    41.             // JDBCURL连接模板
    42.             String jdbcUrlTemplate = "jdbc:odps:%s?project=%s&useProjectTimeZone=true";
    43.             // 使用驱动管理器连接获取连接
    44.             return DriverManager.getConnection(
    45.                     String.format(jdbcUrlTemplate, connParam.getEndpoint(), connParam.getProjectName()), dbProperties);
    46.         } catch (SQLException e) {
    47.             e.printStackTrace();
    48.             throw new BizException(ResultCode.CONN_USER_PWD_ERR);
    49.         }
    50.     }
    51. }

    连接测试代码一起放在结尾，将会开启多个线程获取连接，然后去获取表名，表注释，字段名，字段注释，传入page, size和普通sql就可以实现分页查询的封装方法

    Hive JDBC连接池封装

    Hive JDBC连接参数

    Hive连接参数封装，除了基础的JDBC所需字段，还需要kerberos相关字段，因为hive开启kerberos认证后，需要使用kertab密钥文件和kbr5.conf配置文件去认证。将会在参数和测试代码中得到重复的体现。

    1. /**
    2.  * @Description Hive JDBC connection params
    3.  * @Author itdl
    4.  * @Date 2022/08/10 16:40
    5.  */
    6. @Data
    7. @EqualsAndHashCode(callSuper = false)
    8. public class HiveJdbcConnParam extends BaseJdbcConnParam {
    9.     /**
    10.      * enable kerberos authentication
    11.      */
    12.     private boolean enableKerberos;
    13.  
    14.     /**
    15.      * principal
    16.      */
    17.     private String principal;
    18.  
    19.     /**
    20.      * kbr5 file path in dick
    21.      */
    22.     private String kbr5FilePath;
    23.  
    24.     /**
    25.      * keytab file path in dick
    26.      */
    27.     private String keytabFilePath;
    28. }

    Hive JDBC获取连接实现

    Hive获取JDBC连接之后，本来可以从Connection的元数据中获取表的注释，但是获取的中文注释居然是乱码，但是我们Hue上查看表注释又是正常，暂时没找到这种方式如何解决，从而退而求其次，通过表名去获取建表语句，从建表语句中通过正则表达式提取表的注释。

    1. /**
    2.  * @Description hive connection util
    3.  * @Author itdl
    4.  * @Date 2022/08/10 16:52
    5.  */
    6. @Slf4j
    7. public class HiveConnUtil extends AbstractConnUtil<HiveJdbcConnParam>{
    8.  
    9.     public HiveConnUtil(HiveJdbcConnParam connParam) {
    10.         super(connParam);
    11.     }
    12.  
    13.     /**
    14.      * 获取连接
    15.      * @return 连接
    16.      */
    17.     public Connection getConnection() {
    18.         return connection;
    19.     }
    20.  
    21.     @Override
    22.     protected Connection buildConnection(){
    23.         try {
    24. //            Class.forName("org.apache.hive.jdbc.HiveDriver");
    25.             Class.forName(connParam.getDriverName());
    26.         } catch (ClassNotFoundException e) {
    27.             e.printStackTrace();
    28.             throw new BizException(ResultCode.HIVE_DRIVE_LOAD_ERR);
    29.         }
    30.         // 开启kerberos后需要私钥
    31.         // 拼接jdbcUrl
    32.         String jdbcUrl = "jdbc:hive2://%s:%s/%s";
    33.         String ip = connParam.getIp();
    34.         String port = connParam.getPort() + "";
    35.         String dbName = connParam.getDbName();
    36.         final String username = connParam.getUsername();
    37.         final String password = connParam.getPassword();
    38.         // is enable kerberos authentication
    39.         final boolean enableKerberos = connParam.isEnableKerberos();
    40.         // 格式化
    41.         Connection connection;
    42.         // 获取连接
    43.         try {
    44.             Properties dbProperties = new Properties();
    45.             dbProperties.put("user", username);
    46.             dbProperties.put("password", password);
    47.             // 加上remark后， 能够获取到标注释 但是会出现中文乱码
    48.             dbProperties.put("remarks", "true");
    49.             if (!enableKerberos) {
    50.                 jdbcUrl = String.format(jdbcUrl, ip, port, dbName);
    51.                 connection = DriverManager.getConnection(jdbcUrl, dbProperties);
    52.             } else {
    53.                 final String principal = connParam.getPrincipal();
    54.                 final String kbr5FilePath = connParam.getKbr5FilePath();
    55.                 final String secretFilePath = connParam.getKeytabFilePath();
    56.  
    57.                 String format = "jdbc:hive2://%s:%s/%s;principal=%s";
    58.                 jdbcUrl = String.format(format, ip, port, dbName, principal);
    59.  
    60.                 // 使用hadoop安全认证
    61.                 System.setProperty("java.security.krb5.conf", kbr5FilePath);
    62.                 System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
    63.                 // 解决windows中执行可能出现找不到HADOOP_HOME或hadoop.home.dir问题
    64.                 // Kerberos认证
    65.                 org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
    66.                 conf.set("hadoop.security.authentication", "Kerberos");
    67.                 conf.set("keytab.file", secretFilePath);
    68.                 conf.set("kerberos.principal", principal);
    69.                 UserGroupInformation.setConfiguration(conf);
    70.                 try {
    71.                     UserGroupInformation.loginUserFromKeytab(username, secretFilePath);
    72.                 } catch (IOException e) {
    73.                     e.printStackTrace();
    74.                     throw new BizException(ResultCode.KERBEROS_AUTH_FAIL_ERR);
    75.                 }
    76.                 try {
    77.                     connection = DriverManager.getConnection(jdbcUrl, dbProperties);
    78.                 } catch (SQLException e) {
    79.                     e.printStackTrace();
    80.                     throw new BizException(ResultCode.KERBEROS_AUTH_SUCCESS_GET_CONN_FAIL_ERR);
    81.                 }
    82.             }
    83.             log.info("=====>>>获取hive连接成功：username:{},jdbcUrl: {}", username, jdbcUrl);
    84.             return connection;
    85.         } catch (SQLException e) {
    86.             e.printStackTrace();
    87.             throw new BizException(ResultCode.HIVE_CONN_USER_PWD_ERR);
    88.         } catch (BizException e){
    89.             throw e;
    90.         }
    91.         catch (Exception e) {
    92.             e.printStackTrace();
    93.             throw new BizException(ResultCode.HIVE_CONN_ERR);
    94.         }
    95.     }
    96. }

    Oracle JDBC连接参数封装

    只需要继承父类即可

    1. /**
    2.  * @Description Oracle连接的JDBC参数
    3.  * @Author itdl
    4.  * @Date 2022/08/15 09:50
    5.  */
    6. public class OracleJdbcConnParam extends BaseJdbcConnParam{
    7.     
    8. }

    Oracle JDBC连接实现类

    包括了普通用户的认证和dba用户的认证

    1. /**
    2.  * @Description Oracle获取jdbc连接工具类
    3.  * @Author itdl
    4.  * @Date 2022/08/15 09:52
    5.  */
    6. @Slf4j
    7. public class OracleConnUtil extends AbstractConnUtil<OracleJdbcConnParam> {
    8.     /**
    9.      * 构造函数， 构造工具类对象
    10.      *
    11.      * @param connParam 连接参数
    12.      */
    13.     public OracleConnUtil(OracleJdbcConnParam connParam) {
    14.         super(connParam);
    15.     }
    16.  
    17.     @Override
    18.     protected Connection buildConnection() {
    19.         try {
    20.             Class.forName("oracle.jdbc.driver.OracleDriver");
    21.         } catch (ClassNotFoundException e) {
    22.             e.printStackTrace();
    23.             throw new BizException(ResultCode.ORACLE_DRIVE_LOAD_ERR);
    24.         }
    25.         // 拼接jdbcUrl
    26.         String jdbcUrl = "jdbc:oracle:thin:@//%s:%s/%s";
    27.         final String ip = connParam.getIp();
    28.         final String port = connParam.getPort() + "";
    29.         final String dbName = connParam.getDbName();
    30.         final String username = connParam.getUsername();
    31.         final String password = connParam.getPassword();
    32.  
    33.         // 格式化
    34.         jdbcUrl = String.format(jdbcUrl, ip, port, dbName);
    35.         // 获取连接
    36.         Connection connection;
    37.         try {
    38.             Properties dbProperties = new Properties();
    39.             // 用户名 如果是dba，则后面跟了as sysdba
    40.             String dba = "as sysdba";
    41.             dbProperties.put("password", password);
    42.             dbProperties.put("remarks", "true");
    43.             if (username.trim().endsWith(dba)) {
    44.                 dbProperties.put("user", username.trim().substring(0, username.trim().indexOf(dba) - 1));
    45.                 dbProperties.put("defaultRowPrefetch", "15");
    46.                 dbProperties.put("internal_logon", "sysdba");
    47.                 connection = DriverManager.getConnection(jdbcUrl, dbProperties);
    48.             } else {
    49.                 dbProperties.put("user", username);
    50.                 connection = DriverManager.getConnection(jdbcUrl, dbProperties);
    51.             }
    52.             log.info("=====>>>获取oracle连接成功：username:{}, jdbcUrl: {}", username, jdbcUrl);
    53.             return connection;
    54.         } catch (SQLException e) {
    55.             e.printStackTrace();
    56.             if (e.getMessage().contains("TNS:listener")) {
    57.                 throw new BizException(ResultCode.CONN_LISTENER_UNKNOWN_ERR);
    58.             }
    59.             if (e.getMessage().contains("ORA-01017")) {
    60.                 throw new BizException(ResultCode.CONN_USER_PWD_ERR);
    61.             }
    62.  
    63.             if (e.getMessage().contains("IO 错误: Got minus one from a read call")) {
    64.                 throw new BizException(ResultCode.CONN_CONN_TOO_MANY_ERR);
    65.             }
    66.             throw new BizException(ResultCode.CONN_UNKNOWN_ERR);
    67.         } catch (Exception e) {
    68.             throw new BizException(ResultCode.CONN_UNKNOWN_ERR);
    69.         }
    70.     }
    71. }

    Mysql JDBC连接池封装

    Mysql JDBC连接参数封装

    只需要继承父类即可

    1. /**
    2.  * @Description Mysql连接的JDBC参数
    3.  * @Author itdl
    4.  * @Date 2022/08/15 09:50
    5.  */
    6. public class MysqlJdbcConnParam extends BaseJdbcConnParam{
    7.     
    8. }

    Mysql JDBC连接实现

    需要注意的是连接的属性里面配置useInformationSchema=true，表示可以直接从Connection中获取表和字段的注释。

    1. /**
    2.  * @Description Mysql获取jdbc连接工具类
    3.  * @Author itdl
    4.  * @Date 2022/08/15 09:52
    5.  */
    6. @Slf4j
    7. public class MysqlConnUtil extends AbstractConnUtil<MysqlJdbcConnParam> {
    8.     /**
    9.      * 构造函数， 构造工具类对象
    10.      *
    11.      * @param connParam 连接参数
    12.      */
    13.     public MysqlConnUtil(MysqlJdbcConnParam connParam) {
    14.         super(connParam);
    15.     }
    16.  
    17.     @Override
    18.     protected Connection buildConnection() {
    19.         try {
    20.             Class.forName("com.mysql.cj.jdbc.Driver");
    21.         } catch (ClassNotFoundException e) {
    22.             e.printStackTrace();
    23.             throw new BizException(ResultCode.MYSQL_DRIVE_LOAD_ERR);
    24.         }
    25.  
    26.         // 拼接jdbcUrl
    27.         String jdbcUrl = "jdbc:mysql://%s:%s/%s?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8";
    28.  
    29.         final String ip = connParam.getIp();
    30.         final String port = connParam.getPort() + "";
    31.         final String dbName = connParam.getDbName();
    32.         final String username = connParam.getUsername();
    33.         final String password = connParam.getPassword();
    34.  
    35.         // 格式化
    36.         jdbcUrl = String.format(jdbcUrl, ip, port, dbName);
    37.         // 获取连接
    38.         try {
    39.             Properties dbProperties = new Properties();
    40.             dbProperties.put("user", username);
    41.             dbProperties.put("password", password);
    42.             dbProperties.put("remarks", "true");
    43.             // 设置可以获取tables remarks信息
    44.             dbProperties.setProperty("useInformationSchema", "true");
    45.             Connection connection = DriverManager.getConnection(jdbcUrl,dbProperties);
    46.             log.info("=====>>>获取mysql连接成功：username:{}, jdbcUrl: {}", username, jdbcUrl);
    47.             return connection;
    48.         } catch (SQLException e) {
    49.             e.printStackTrace();
    50.             if (e.getMessage().contains("Unknown database")){
    51.                 throw new BizException(ResultCode.CONN_UNKNOWN_DB_ERR);
    52.             }
    53.             throw new BizException(ResultCode.CONN_USER_PWD_ERR);
    54.         } catch (Exception e) {
    55.             throw new BizException(ResultCode.CONN_UNKNOWN_ERR);
    56.         }
    57.     }
    58. }

    测试代码连接各自数据库

    1. @SpringBootTest(classes = DbConnectionDemoApplication.class)
    2. @RunWith(value = SpringRunner.class)
    3. @Slf4j
    4. class DbConnectionDemoApplicationTests {
    5.  
    6.     private DbConnPool<?> connPool = null;
    7.  
    8.     @Test
    9.     public void testMysqlConn() throws InterruptedException {
    10.         // 创建连接参数
    11.         final MysqlJdbcConnParam connParam = new MysqlJdbcConnParam();
    12.         final String ip = "localhost";
    13.         final Integer port = 3306;
    14.         final String username = "root";
    15.         final String password = "root";
    16.         final String dbname = "test_db";
    17.  
    18.         // 设置参数
    19.         connParam.setDriverName(Driver.class.getName());
    20.         connParam.setIp(ip);
    21.         connParam.setPort(port);
    22.         connParam.setUsername(username);
    23.         connParam.setPassword(password);
    24.         connParam.setDbName(dbname);
    25.  
    26.         // 创建连接池
    27.         connPool = new DbConnPool<>(connParam, 2);
    28.  
    29.         handler01(dbname, DbDialectEnum.MYSQL);
    30.  
    31.         new Thread(() -> handler01(dbname, DbDialectEnum.MYSQL)).start();
    32.         new Thread(() -> handler01(dbname, DbDialectEnum.MYSQL)).start();
    33.  
    34.         Thread.sleep(60 * 1000);
    35.     }
    36.  
    37.  
    38.     @Test
    39.     public void testOracleConn() throws InterruptedException {
    40.         // 创建连接参数
    41.         final OracleJdbcConnParam connParam = new OracleJdbcConnParam();
    42.         final String ip = "你的Oracle的IP地址";
    43.         final Integer port = 1521;
    44.         // 如果是admin账号 用户后面+ as sysdba
    45.         final String username = "用户名";
    46.         final String password = "密码";
    47.         final String dbname = "实例/服务名";
    48.  
    49.         // 设置参数
    50.         connParam.setDriverName(Driver.class.getName());
    51.         connParam.setIp(ip);
    52.         connParam.setPort(port);
    53.         connParam.setUsername(username);
    54.         connParam.setPassword(password);
    55.         connParam.setDbName(dbname);
    56.  
    57.         // 创建连接池
    58.         connPool = new DbConnPool<>(connParam, 2);
    59.  
    60.         final DbDialectEnum dbDialectEnum = DbDialectEnum.ORACLE;
    61.  
    62.         // 处理操作(oracle的schemaName就是用户名)
    63.         handler01(username, dbDialectEnum);
    64.  
    65.         // 新建两个线程获取连接
    66.         new Thread(() -> handler01(username, dbDialectEnum)).start();
    67.         new Thread(() -> handler01(username, dbDialectEnum)).start();
    68.  
    69.         Thread.sleep(60 * 1000);
    70.     }
    71.  
    72.  
    73.     @Test
    74.     public void testHiveConn() throws InterruptedException {
    75.         // 创建连接参数
    76.         final HiveJdbcConnParam connParam = new HiveJdbcConnParam();
    77.         final String ip = "连接的域名";
    78.         final Integer port = 10000;
    79.         // 如果是admin账号 用户后面+ as sysdba
    80.         final String username = "账号@域名";
    81.         final String password = "";
    82.         final String dbname = "数据库名";
    83.         final String principal = "hive/_HOST@域名";
    84.         final String kbr5FilePath = "C:\\workspace\\krb5.conf";
    85.         final String keytabFilePath = "C:\\workspace\\zhouyu.keytab";
    86.  
    87.         // 设置参数
    88.         connParam.setDriverName(Driver.class.getName());
    89.         connParam.setIp(ip);
    90.         connParam.setPort(port);
    91.         connParam.setUsername(username);
    92.         connParam.setPassword(password);
    93.         connParam.setDbName(dbname);
    94.         connParam.setEnableKerberos(true);
    95.         connParam.setPrincipal(principal);
    96.         connParam.setKbr5FilePath(kbr5FilePath);
    97.         connParam.setKeytabFilePath(keytabFilePath);
    98.  
    99.         // 创建连接池
    100.         connPool = new DbConnPool<>(connParam, 2);
    101.  
    102.         final DbDialectEnum dbDialectEnum = DbDialectEnum.HIVE;
    103.  
    104.         // 处理操作(oracle的schemaName就是用户名)
    105.         handler01(username, dbDialectEnum);
    106.  
    107.         // 新建两个线程获取连接
    108.         new Thread(() -> handler01(username, dbDialectEnum)).start();
    109.         new Thread(() -> handler01(username, dbDialectEnum)).start();
    110.  
    111.         Thread.sleep(10 * 60 * 1000);
    112.     }
    113.  
    114.  
    115.     @Test
    116.     public void testMaxComputeConn() throws InterruptedException {
    117.         // 创建连接参数
    118.         final MaxComputeJdbcConnParam connParam = new MaxComputeJdbcConnParam();
    119.         String accessId = "你的阿里云accessId";
    120.         String accessKey = "你的阿里云accessKey";
    121.         String endpoint = "http://service.cn-chengdu.maxcompute.aliyun.com/api";
    122.         String projectName = "项目名=数据库名";
    123.  
    124.         // 设置参数
    125.         connParam.setDriverName(Driver.class.getName());
    126.         connParam.setAliyunAccessId(accessId);
    127.         connParam.setAliyunAccessKey(accessKey);
    128.         connParam.setEndpoint(endpoint);
    129.         connParam.setProjectName(projectName);
    130.  
    131.         // 创建连接池
    132.         connPool = new DbConnPool<>(connParam, 2);
    133.  
    134.         final DbDialectEnum dbDialectEnum = DbDialectEnum.MAX_COMPUTE;
    135.  
    136.         // 处理操作(oracle的schemaName就是用户名)
    137.         handler01(projectName, dbDialectEnum);
    138.  
    139.         // 新建两个线程获取连接
    140.         new Thread(() -> handler01(projectName, dbDialectEnum)).start();
    141.         new Thread(() -> handler01(projectName, dbDialectEnum)).start();
    142.  
    143.         Thread.sleep(60 * 1000);
    144.     }
    145.  
    146.  
    147.     private void handler01(String schemaName, DbDialectEnum dbDialectEnum) {
    148.         final Connection connection = connPool.getConnection();
    149.  
    150.         // 构建工具类
    151.         final SqlUtil sqlUtil = new SqlUtil(connection, dbDialectEnum.getCode());
    152.  
    153.         // 获取表和注释
    154.         final List<TableMetaInfo> tables = sqlUtil.getTables(schemaName);
    155.         log.info("===============获取所有表和注释开始===================");
    156.         log.info(tables.toString());
    157.         log.info("===============获取所有表和注释结束===================");
    158.  
    159.         // 获取字段和注释
    160.         final String tableName = tables.get(0).getTableName();
    161.         final List<TableColumnMetaInfo> columns = sqlUtil.getColumnsByTableName(tableName);
    162.         log.info("===============获取第一个表的字段和注释开始===================");
    163.         log.info(columns.toString());
    164.         log.info("===============获取第一个表的字段和注释结束===================");
    165.  
    166.  
    167.         final PageResult<LinkedHashMap<String, Object>> pageResult = sqlUtil.pageQueryMap("select * from " + tableName, 1, 10);
    168.         log.info("===============SQL分页查询开始===================");
    169.         log.info("总数：{}", pageResult.getTotal());
    170.         log.info("记录数：{}", JSONObject.toJSONString(pageResult.getRows()));
    171.         log.info("===============SQL分页查询结束===================");
    172.  
    173.         connPool.freeConnection(connection);
    174.     }
    175.  
    176.     @After
    177.     public void close(){
    178.         if (connPool != null){
    179.             connPool.close();
    180.             log.info("==================连接池成功关闭================");
    181.         }
    182.     }
    183. }
  • 相关阅读:
    Go 学习笔记（88） — 字符串拼接方法和性能、字符串组成、UTF-8 编码方案、字符串之间转换
    uniapp输入框组件easyInput初始状态清除符号显示的bug
    鸿蒙HarmonyOS实战-ArkUI动画（页面转场动画）
    微服务项目部署-POS收银系统
    Bean的生命周期及演示
    Spring Data JPA使用自定义查询进行分页
    井字棋游戏
    选专业，适合理科女生的大学专业有哪些？
    PMP备考大全：经典题库（6月第4周）
    程序化交易是科学和艺术的结合
  • 原文地址：https://blog.csdn.net/qq_35267557/article/details/126394020