DruidDSUtil 工具类，连接数据库，执行SQL工具类

package com.sunny.datastream.transform.utils;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 数据库连接 工具类
 *
 * /**
 * # 数据源配置
 spring:
 datasource:
 type: com.alibaba.druid.pool.DruidDataSource
 #driverClassName: com.mysql.cj.jdbc.Driver
 druid:
 # 主库数据源
 master:
 url: jdbc:mysql://localhost:3306/ry-vue?useUnicode=true&characterEncoding=utf8
 username: root
 password: root
 */

public class DruidDSUtil {
    private static DruidDataSource druidDataSource;
    public static DruidDataSource createDataSource() {
        //创建连接池
        druidDataSource = new DruidDataSource();
        // 设置驱动全类名
        druidDataSource.setDriverClassName(GmallConfig.PHOENIX_DRIVER);
        // 设置连接 Url
        druidDataSource.setUrl(GmallConfig.PHOENIX_SERVER);
        druidDataSource.setUsername( GmallConfig.PHOENIX_NAME);     // 用户名  "root"
        druidDataSource.setPassword(GmallConfig.PHOENIX_PASSWORDS);  // 密码 "123456"
        //属性类型是字符串，通过别名的方式配置扩展插件，常用的插件有：
        //监控统计用的filter:stat
        //日志用的filter:log4j
        //防御sql注入的filter:wal
        // druidDataSource.setFilters("log4j");
        // 设置初始化连接池中连接的数量
        druidDataSource.setInitialSize(5);
        //  设置同事活跃的最大连接数
        druidDataSource.setMaxActive(20);
        //  设置空闲时的最小连接数，必须介于 0 和 最大连接数之间，默认为0
        druidDataSource.setMinIdle(1);
        //  设置没有空余连接时的等待时间，超时抛出异常，-1 表示一直等待
        druidDataSource.setMaxWait(-1);
        //  验证连接是否可用 使用的 SQL 语句
        druidDataSource.setValidationQuery("select 1");
        // 指明连接是否被空闲连接回收器(如果有)进行校验，如果检测失败，则连接将被从池中取出
        // 注意默认值为true ,如果没有设置 validationQuery 则报错
        // testWhileIdle is true validationQuery not set
        druidDataSource.setTestWhileIdle(true);
        // 借出连接时，是否测试，设置为false，不测试，否则很影响性能
        druidDataSource.setTestOnBorrow(false);
        // 归还连接时，是否测试
        druidDataSource.setTestOnReturn(false);
        // 设置空闲连接回收器每隔 30s 运行一次
        druidDataSource.setTimeBetweenEvictionRunsMillis(30*1000L);
        // 设置池中连接空闲30 min, 被回收，默认值即为 30 min
        druidDataSource.setMinEvictableIdleTimeMillis(30*60*1000L);
        return druidDataSource;
    }

    public static void main(String[] args) throws SQLException {
        //1、测试创建连接 执行sql
        DruidDataSource dataSource = DruidDSUtil.createDataSource();
        //
        String sql = "select * from userTable";
        DruidPooledConnection connection =null;
        PreparedStatement preparedStatement = null;
        try {
            connection = dataSource.getConnection();
            preparedStatement = connection.prepareStatement(sql);
            preparedStatement.execute();
            connection.commit();

            preparedStatement.close();

            connection.close();
        } catch (SQLException e) {
            System.out.println("数据库异常=>"+e.getMessage());
        } finally {
            if(preparedStatement!=null){
                preparedStatement.close();
            }
            if(connection!=null){
                connection.close();
            }
        }

        //1、测试使用工具类 PhoenixUtil.upserValue 执行sql
        DruidDataSource dataSource1 = DruidDSUtil.createDataSource();
        String sql1 = "select * from userTable";
        DruidPooledConnection connection1 =null;
        try {
            // 创建 JSONObject 匿名类 并赋值。
            JSONObject objectJson = new JSONObject(){
                {
                    put("name","zhangsan");put("age", 25);put("gender", "男");
                }
            };
//            objectJson.put("name","zhangsan");
//            objectJson.put("age", 25);
//            objectJson.put("gender", "男");
            connection1 = dataSource1.getConnection();
            PhoenixUtil.upserValue(connection1,"userTable",objectJson);
            connection1.close();
        } catch (SQLException e) {
            System.out.println("数据库异常=>"+e.getMessage());
        } finally {
            // 打印异常，关闭 connection 连接
            if(connection1!=null){
                connection1.close();
            }
        }

    }
}

package com.sunny.datastream.transform.utils;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import io.netty.util.internal.StringUtil;
import org.apache.commons.lang.StringUtils;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * 执行SQL工具类
 */
public class PhoenixUtil {
    /*
     对于HBASE Inset和 UPdate是一个sql,
     执行sql 工具类 .获取连接 表名 执行的 数据，拼接sql
     */
    public static void upserValue(DruidPooledConnection connection, String sinkTable, JSONObject data) throws SQLException {

        // 1 拼接SQL语句 upsert into db.tn(id,name,sex) value ('1001','zhangsan','male');
        Set<String> colums = data.keySet();
        Collection<Object> values = data.values();
        // StringUtils.join(clumns,",") == clumns.mkString(",") ==> id,name,sex
        String sql = "upsert  into "+GmallConfig.HBASE_SCHEMA +"."+sinkTable +"(" +
                StringUtils.join(colums,",")+") values ('"+
                StringUtils.join(colums,",")+"')";
        // 2 预编译sql
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        // 3 执行
        preparedStatement.execute();
        connection.commit();  // sql 提交
        // 4 资源释放
        preparedStatement.close();
    }
}