Matt

import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.security.PrivilegedAction;
import java.sql.*;
import java.util.LinkedList;

/*** 类名： ImpalaJdbcUtils
 * * 描述：impala jdbc的工具类
 * * 自定义连接工具池
 * * **/
public class ImpalaJdbcUtils {
    private static final Logger log = LoggerFactory.getLogger(ImpalaJdbcUtils.class);
    /*** 描述：驱动类 <br>*/
    private static String DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver";
    private static String CONNECTION_URL = "jdbc:hive2://*****/default;principal=impala/";
    /*** 连接池**/
    private static LinkedList<Connection> pool = new LinkedList<Connection>();//初始化10个连接

    static {
        try {
            for (int i = 0; i < 15; i++) {
                Connection conn = getImpalaConnection();
                //得到一个连接
                pool.add(conn);
            }
        } catch (Exception e) {
            log.error("数据库连接池初始化失败，请检查配置", e);
        }
    }
    /*** 从连接池中获取连接**/
    public static Connection getConnection() throws SQLException {
        // 取出连接池中一个连接
        Connection conn = null;
        // 测试连接是否超时，超时重新连接
        try {
            conn = pool.removeFirst();
            //执行以下链接，查看是否异常
            String sql = "show DATABASES";
            PreparedStatement ps = conn.prepareStatement(sql);
            ResultSet rs = ps.executeQuery();
            rs.close();
            ps.close();
        }catch (Exception e){
            log.warn("连接异常,尝试重新生成连接！");
            conn = getImpalaConnection();
        }
        // 删除第一个连接返回
        log.info("取出一个连接剩余 " + pool.size() + "个连接！");
        return conn;
    }

    /*** @Description 释放连接***/
    public static void releaseConnection(Connection conn) {
        pool.add(conn);
        log.info("将连接放回到连接池中,数量:" + pool.size());
    }
    /*** @Description 获取impala连接* **/
    public static Connection getImpalaConnection() {
        try {
            Object o = UserGroupInformation.getLoginUser().doAs((PrivilegedAction<Object>) () -> {
                Connection connection = null;
                try {
                    Class.forName(DRIVER_CLASS);
                } catch (ClassNotFoundException e) {
                    log.error("加载impala驱动失败", e);
                }
                try {
                    connection = DriverManager.getConnection(CONNECTION_URL);
                } catch (SQLException e) {
                    log.error("获取jdbc连接失败", e);
                }
                return connection;
            });
            return (Connection) o;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    /*** 关闭jdbc连接* **/
    public static void disconnect(Connection connection, ResultSet res, PreparedStatement ps) {
        try {
            if (res != null){
                res.close();
            }
            if (ps != null){
                ps.close();
            }
            if (connection != null) {
                connection.close();
                ImpalaJdbcUtils.releaseConnection(connection);
            }
        } catch (SQLException e) {
            log.error("jdbc连接关闭失败", e);
        }
    }

    //不使用连接池
    public Connection getConnectSync(){
        try {
            Object o = UserGroupInformation.getLoginUser().doAs((PrivilegedAction<Object>) () -> {
                Connection connection = null;
                try {
                    Class.forName(DRIVER_CLASS);
                } catch (ClassNotFoundException e) {
                    log.error("加载impala驱动失败", e);
                }
                try {
                    connection = DriverManager.getConnection(CONNECTION_URL);
                } catch (SQLException e) {
                    log.error("获取jdbc连接失败", e);
                }
                return connection;
            });
            return (Connection) o;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}