62、Flink深入:Flink中通用MySQLUtil工具类

1. 开发目的

在使用SpringBoot后端开发中,我们如果需要对MySQL进行增删查改,可以很方便的使用Mybatis进行操作。但是在大数据中,如果想要对MySQL进行操作,就没有那么方便,特别当flink新一代流式计算框架兴起后,在老版本中没有读取和写入MySQL的连接源,虽然在后续新版本中有以及社区开发的其他补充项目中有source源和sink源了(比如flink-cdc和写入MySQL的sink方法),但当中间需要读取维度数据时,还是不方便。此时一个较为方便的工具类就能很方便的使用,能达到节省开发时间、减小开发难度等目的。

2. 导入依赖

以MySQL8.x版本为例

        <!--MySQL驱动包 mysql8版本-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
 
        <!-- 日志打印的jar包 -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <scope>provided</scope>
        </dependency>
 
        <!-- json解析包,fastjson包 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
 
        <!--commons-beanutils 是 Apache 开源组织提供的用于操作 JAVA BEAN 的工具包。使用 commons-beanutils,我们可以很方便的对 bean 对象的属性进行操作-->
        <dependency>
            <groupId>commons-beanutils</groupId>
            <artifactId>commons-beanutils</artifactId>
            <scope>provided</scope>
        </dependency>
 
        <!--Guava 工程包含了若干被 Google 的 Java 项目广泛依赖的核心库,方便开发-->
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <scope>provided</scope>
        </dependency>
 
        <!-- 数据库连接池和jdbc操作模板 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
            <version>${springboot.version}</version>
        </dependency>
        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
            <version>2.6.1</version>
        </dependency>

对应版本号:

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <scala.binary.version>2.11</scala.binary.version>
    <scala.version>2.11.8</scala.version>
    <flink.binary.version>1.10</flink.binary.version>
    <flink.version>1.10.0</flink.version>
    <alink.version>1.4.0</alink.version>
    <log4j.version>1.2.17</log4j.version>
    <slf4j.version>1.7.21</slf4j.version>
    <mysql.version>8.0.21</mysql.version>
    <fastjson.version>1.2.75</fastjson.version>
    <huaweicloud.dws.jdbc.version>8.1.0</huaweicloud.dws.jdbc.version>
    <commons.beanutils.version>1.9.4</commons.beanutils.version>
    <guava.version>29.0-jre</guava.version>
    <okhttp.version>3.6.0</okhttp.version>
    <springboot.version>2.0.2.RELEASE</springboot.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <avro.version>1.10.0</avro.version>
</properties>

3. 代码

注意:下述代码中使用了自定义的ModelUtil工具类,该工具类的具体介绍可以参考博主的另一篇博文:Flink(60):Flink中通用ModelUtil工具类

3.1. 方法说明

构造方法:

MySQLUtil(String url, String username, String password, int maxConnect, int minConnect) :根据传入的MySQL各项信息和连接池连接的数量来构建连接池
MySQLUtil(String url, String username, String password) :根据传入的MySQL各项信息,使用默认的连接池配置(最大连接2个,最小连接1个)

初始化JdbcTemplate:

 initJdbcTemplate(String url, String username, String password, int maxConnect, int minConnect) :在构造方法中调用,使用HikariPool连接池初始化JdbcTemplate,并赋值给成员变量,后续使用

处理特殊字符:

disposeSpecialCharacter(Object object) :对传入的数据中的特殊字符进行处理,比如当我们拼接SQL时,使用的是单引号拼接,但当数据中有单引号就会报错,可以对这些特殊字符进行处理

查询方法:

List<T> queryList(String sql, Class<T> clz, boolean underScoreToCamel) :通过输入的SQL语句查询MySQL表中的数据,并将数据转换成传入的clz对应的对象

插入方法:

insert(String tableName, boolean underScoreToCamel, Object object) : 通过输入的表名,以及是否对字段名进行下划线驼峰命名转换,将传入的数据(object对象,可以是bean对象,也可以是JSONObject对象)插入到对应的表中

删除方法:

delete(String tableName, Map<String, Object> fieldNameAndValue) : 通过传入的表名以及删除的条件(Map集合),删除对应表中对应的数据
delete(String tableName, boolean underScoreToCamel, Object object, String... fields) : 通过传入的表名,以及是否对字段名进行下划线驼峰命名转换,并使用传入对象中给定的字段进行匹配,删除匹配到的数据

更新方法:

update(String tableName, boolean underScoreToCamel, Object object, Map<String, Object> fieldNameAndValue) :通过传入的表名,以及是否对字段名进行下划线驼峰命名转换,将对应的object对象中的数据更新到表中,使用传入Map集合作为匹配条件
update(String tableName, boolean underScoreToCamel, Object object, String... fields) : 通过传入的表名,以及是否对字段名进行下划线驼峰命名转换,将对应的object对象中的数据更新到表中,使用传入的字段集合作为更新条件

upsert方法:

upsertByPrimaryKey(String tableName, boolean underScoreToCamel, Object object, Map<String, Object> fieldNameAndValue) : 根据主键使用ON DUPLICATE KEY UPDATE语法对对应的表中的数据进行更新,也可以对传入的数据中的字段是否下划线驼峰命名转换,并将传入的Map集合作为update条件
upsertByPrimaryKey(String tableName, boolean underScoreToCamel, Object object, String... fields) : 根据主键使用ON DUPLICATE KEY UPDATE语法对对应的表中的数据进行更新,也可以对传入的数据中的字段是否下划线驼峰命名转换,并将传入的字段集合作为update条件(字段集合必须在object中有数据)
upsert(String tableName, boolean underScoreToCamel, Object object, String... fields) :先使用上述update方法,如果返回变更的条数等于0,就执行上述的insert方法,如果大于0就不执行
upsert(String tableName, boolean underScoreToCamel, Object object, Map<String, Object> fieldNameAndValue) : 先使用上述update方法,如果返回变更的条数等于0,就执行上述的insert方法,如果大于0就不执行

3.2. 具体实现

package com.yishou.bigdata.common.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import com.google.common.collect.Lists;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.*;

/**
 * @date: 2022/5/17
 *  @Author ddkk.com  弟弟快看,程序员编程资料站
 * @desc: MySQLUtil
 */
public class MySQLUtil {

    static Logger logger = LoggerFactory.getLogger(MySQLUtil.class);

    /**
     * jdbcTemplate
     */
    private JdbcTemplate jdbcTemplate;

    /**
     * 通过传入的参数创建MySQLUtil对象
     *
     * @param url        mysql的url
     * @param username   mysql的username
     * @param password   mysql的password
     * @param maxConnect 连接池中最大连接数
     * @param minConnect 连接池中最小连接数
     */
    public MySQLUtil(String url, String username, String password, int maxConnect, int minConnect) {
        initJdbcTemplate(url, username, password, maxConnect, minConnect);
    }

    /**
     * 通过传入的参数创建MySQLUtil对象
     *
     * @param url      mysql的url
     * @param username mysql的username
     * @param password mysql的password
     */
    public MySQLUtil(String url, String username, String password) {
        initJdbcTemplate(url, username, password, 2, 1);
    }

    /**
     * 初始化MySQL的jdbcTemplate
     *
     * @param url        mysql的url
     * @param username   mysql的username
     * @param password   mysql的password
     * @param maxConnect 连接池中最大连接数
     * @param minConnect 连接池中最小连接数
     */
    public void initJdbcTemplate(String url, String username, String password, int maxConnect, int minConnect) {
        try {
            HikariDataSource ds = new HikariDataSource();
            Thread.sleep(1000);
            ds.setDriverClassName("com.mysql.cj.jdbc.Driver");
            ds.setJdbcUrl(url);
            ds.setUsername(username);
            ds.setPassword(password);
            ds.setMaximumPoolSize(maxConnect);
            ds.setMinimumIdle(minConnect);
            jdbcTemplate = new JdbcTemplate(ds);

            logger.info(
                    "使用HikariPool连接池初始化JdbcTemplate成功,使用的URL为:{} , 其中最大连接大小为:{} , 最小连接大小为:{} ;",
                    ds.getJdbcUrl(),
                    ds.getMaximumPoolSize(),
                    ds.getMinimumIdle()
            );

        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("创建MySQL数据库的jdbcTemplate失败,抛出的异常信息为:" + e.getMessage());
        }
    }

    /**
     * 获取对应的 JdbcTemplate
     *
     * @return JdbcTemplate
     */
    public JdbcTemplate getJdbcTemplate() {
        return jdbcTemplate;
    }

    /**
     * 处理传入数据中的特殊字符(例如: 单引号),并将其中数据为空的过滤
     *
     * @param object 传入的数据对象
     * @return 返回的结果
     */
    public String disposeSpecialCharacter(Object object) {

        // 根据传入的情况,将数据转换成json格式(如果传入为string,那就本来是json格式,不需要转)
        String data;
        if (object instanceof String) {
            data = object.toString();
        } else {
            data = JSON.parseObject(JSON.toJSONString(object)).toString();
        }

        // 处理传入数据中的特殊字符(例如: 单引号)
        data = data.replace("'", "''");
        // 将其中为空值的去掉(注意:如果不能转换成json并从中获取数据的,就是从delete中传过来的,只有单纯的value值)
        try {
            JSONObject result = new JSONObject();
            for (Map.Entry<String, Object> entry : JSON.parseObject(data).entrySet()) {
                if (StringUtils.isNotEmpty(entry.getValue().toString())) {
                    result.put(entry.getKey(), entry.getValue());
                }
            }
            data = result.toJSONString();
        } catch (Exception exception) {
            exception.printStackTrace();
            logger.warn("传入的数据为:{};该数据是从delete中传入的,不能转换成json值", object);
        }

        // 返回数据
        return data;

    }
    /**
     * 如果传入的clz中的属性又包含对象,会报错,此时传入JSONObject对象即可
     *
     * @param sql               执行的查询语句
     * @param clz               返回的数据类型
     * @param underScoreToCamel 是否将下划线转换为驼峰命名法
     * @param <T  样例类
     * @return 样例类集合
     */
    public <T> List<T> queryList(String sql, Class<T> clz, boolean underScoreToCamel) {
        try {
            List<Map<String, Object>> mapList = jdbcTemplate.queryForList(sql);
            List<T> resultList = new ArrayList<>();
            for (Map<String, Object> map : mapList) {
                Set<String> keys = map.keySet();
                // 当返回的结果中存在数据,通过反射将数据封装成样例类对象
                T result = clz.newInstance();
                for (String key : keys) {
                    String propertyName = underScoreToCamel ? CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, key) : key;
                    BeanUtils.setProperty(
                            result,
                            propertyName,
                            map.get(key)
                    );
                }
                resultList.add(result);
            }
            return resultList;
        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException(
                    "\r\n从MySQL数据库中 查询 数据失败," +
                            "\r\n抛出的异常信息为:" + exception.getMessage() +
                            "\r\n查询的SQL为:" + sql
            );
        }
    }

    /**
     * 将传入的数据插入到对应的MySQL的表中
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象
     *                          INSERT INTO customer_t1 (c_customer_sk, c_first_name) VALUES (3769, 'Grace');
     */
    public void insert(String tableName, boolean underScoreToCamel, Object object) {

        // 将传入的对象转换成JSONObject格式(并将其中的特殊字符进行替换)
        JSONObject data = JSON.parseObject(disposeSpecialCharacter(object));

        // 从传入的数据中获取出对应的key和value,因为要一一对应,所以使用list
        ArrayList<String> fieldList = Lists.newArrayList(data.keySet());
        ArrayList<String> valueList = new ArrayList<>();
        for (String field : fieldList) {
            valueList.add(data.getString(field));
        }

        // 拼接SQL
        StringBuilder sql = new StringBuilder();
        sql.append(" INSERT INTO ").append(tableName);
        sql.append(" ( ");
        for (String field : fieldList) {
            if (underScoreToCamel) {
                sql.append(CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, field)).append(",");
            } else {
                sql.append(field).append(",");
            }
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(" ) ");
        sql.append(" values ('").append(StringUtils.join(valueList, "','")).append("')");

        // 执行插入操作
        try {
            jdbcTemplate.execute(sql.toString());
        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException(
                    "\r\n向MySQL数据库中 插入 数据失败," +
                            "\r\n抛出的异常信息为:" + exception.getMessage() +
                            "\r\n执行的SQL为:" + sql
            );
        }
    }

    /**
     * 根据主键删除对应数据
     * 注意:传入的字段名要和数据库中一一匹配,即数据库中有下划线,那传入的字段名也要有下划线
     *
     * @param tableName         表名
     * @param fieldNameAndValue 更新时匹配的字段(key)和值(value)(注意:传入的字段名要和数据库中一一匹配,即数据库中有下划线,那传入的字段名也要有下划线)
     * @return 删除时影响的条数
     */
    public int delete(String tableName, Map<String, Object> fieldNameAndValue) {

        // 拼接SQL
        StringBuilder sql = new StringBuilder();
        sql.append(" delete from ").append(tableName);
        if (fieldNameAndValue.size() > 0) {
            sql.append(" WHERE ");
            for (Map.Entry<String, Object> fieldNameAndValueEntry : fieldNameAndValue.entrySet()) {
                sql
                        .append(fieldNameAndValueEntry.getKey())
                        .append(" = ")
                        .append("'")
                        .append(disposeSpecialCharacter(fieldNameAndValueEntry.getValue()))
                        .append("'")
                        .append(" AND ");
            }
            sql.delete(sql.length() - 4, sql.length() - 1);
        } else {
            throw new RuntimeException("从MySQL中删除数据异常,输入的删除条件没有指定字段名和对应的值,会进行全表删除, 拼接的SQL为:" + sql);
        }

        // 执行删除操作
        try {
            return jdbcTemplate.update(sql.toString());
        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException(
                    "\r\n向MySQL数据库中 删除 数据失败," +
                            "\r\n抛出的异常信息为:" + exception.getMessage() +
                            "\r\n执行的SQL为:" + sql
            );
        }
    }

    /**
     * 根据传入的表名、数据、字段名,删除表中对应的数据
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象
     * @param fields            更新时匹配的字段名
     */
    public int delete(String tableName, boolean underScoreToCamel, Object object, String... fields) {

        // 将传入的对象转换成JSONObject格式
        JSONObject data = JSON.parseObject(disposeSpecialCharacter(object));

        // 根据传入的字段,获取要更新的主键值
        HashMap<String, Object> fieldNameAndValue = new HashMap<>();
        for (String field : fields) {
            if (underScoreToCamel) {
                // data中的均为驼峰,获取数据时需要使用驼峰;但是将数据写入到fieldNameAndValue中时,需要全部转换成下划线
                fieldNameAndValue.put(
                        field.contains("_") ? field : CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, field),
                        data.getString(field.contains("_") ? CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, field) : field)
                );
            } else {
                // data中均为下划线,field中也是下划线
                fieldNameAndValue.put(field, data.getString(field));
            }
        }

        // 调用重载函数,删除数据
        return delete(tableName, fieldNameAndValue);

    }

    /**
     * 将传入的数据 更新 到对应的MySQL的表中
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象(既可以包含更新的主键,也可以不包含)
     * @param fieldNameAndValue 更新时匹配的字段和对应的值
     * @return 返回更新的条数
     */
    public int update(String tableName, boolean underScoreToCamel, Object object, Map<String, Object> fieldNameAndValue) {

        // 将传入的对象转换成JSONObject格式,并判断输入的数据是否符合更新条件
        JSONObject data = JSON.parseObject(disposeSpecialCharacter(object));
        if (fieldNameAndValue == null || fieldNameAndValue.size() == 0) {
            throw new RuntimeException("向MySQL中更新数据异常,输入的更新条件没有指定数据,不能更新(这样更新会全表更新),传入的数据为:" + data);
        }

        // 拼接SQL
        StringBuilder sql = new StringBuilder();
        sql.append(" UPDATE ").append(tableName);
        sql.append(" SET ");

        if (underScoreToCamel) {

            // 删除传入对象中要更新的数据
            for (String key : fieldNameAndValue.keySet()) {
                data.remove(key.contains("_") ? CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, key) : key);
            }

            // 拼接要更新的结果值
            for (Map.Entry<String, Object> entry : data.entrySet()) {
                sql
                        .append(CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, entry.getKey()))
                        .append(" = ")
                        .append("'")
                        .append(entry.getValue())
                        .append("'")
                        .append(",");
            }
            sql.deleteCharAt(sql.length() - 1);

            // 拼接判断条件
            sql.append(" WHERE ");
            for (Map.Entry<String, Object> fieldNameAndValueEntry : fieldNameAndValue.entrySet()) {
                String key = fieldNameAndValueEntry.getKey();
                Object value = fieldNameAndValueEntry.getValue();
                sql
                        .append(key.contains("_") ? key : CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, key))
                        .append(" = ")
                        .append("'")
                        .append(value)
                        .append("'")
                        .append(" AND ");
            }

        } else {

            // 删除传入对象中要更新的数据
            for (String key : fieldNameAndValue.keySet()) {
                data.remove(key);
            }

            // 拼接要更新的结果值
            for (Map.Entry<String, Object> entry : data.entrySet()) {
                sql
                        .append(entry.getKey())
                        .append(" = ")
                        .append("'")
                        .append(entry.getValue())
                        .append("'")
                        .append(",");
            }
            sql.deleteCharAt(sql.length() - 1);

            // 拼接判断条件
            sql.append(" WHERE ");
            for (Map.Entry<String, Object> fieldNameAndValueEntry : fieldNameAndValue.entrySet()) {
                String key = fieldNameAndValueEntry.getKey();
                Object value = fieldNameAndValueEntry.getValue();
                sql
                        .append(key)
                        .append(" = ")
                        .append("'")
                        .append(value)
                        .append("'")
                        .append(" AND ");
            }
        }
        sql.delete(sql.length() - 4, sql.length() - 1);

        // 执行更新操作
        try {
            return jdbcTemplate.update(sql.toString());
        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException(
                    "\r\n向MySQL数据库中 更新 数据失败," +
                            "\r\n抛出的异常信息为:" + exception.getMessage() +
                            "\r\n执行的SQL为:" + sql
            );
        }
    }

    /**
     * 将传入的数据 更新 到对应的MySQL的表中
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象
     * @param fields            更新时匹配的字段名(如果underScoreToCamel为true,传入的字段为驼峰,如果underScoreToCamel为false,传入的字段为下划线)
     * @return 返回更新的条数
     */
    public int update(String tableName, boolean underScoreToCamel, Object object, String... fields) {

        // 将传入的对象转换成JSONObject格式
        JSONObject data = JSON.parseObject(disposeSpecialCharacter(object));

        // 根据传入的字段,获取要更新的主键值
        HashMap<String, Object> fieldNameAndValue = new HashMap<>();
        for (String field : fields) {
            if (underScoreToCamel) {
                field = field.contains("_") ? CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, field) : field;
            }
            fieldNameAndValue.put(field, data.getString(field));
        }

        // 调用重载函数,更新数据
        return update(tableName, underScoreToCamel, object, fieldNameAndValue);

    }

    /**
     * 将传入的数据 upsert 到对应的MySQL的表中
     * 会根据MySQL表中的主键进行更新,如果该主键在表中有对应数据,就更新,没有就插入
     * 传入的数据中必须有主键
     * <p>
     * mysql中的upsert语法:
     * INSERT INTO Student(Stud_ID, Name, Email, City) VALUES (4, 'John', 'john@lidihuo.com', 'New York') ON DUPLICATE KEY UPDATE City = 'California';
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象
     * @param fieldNameAndValue 更新时匹配的字段名(如果underScoreToCamel为true,传入的字段为驼峰,如果underScoreToCamel为false,传入的字段为下划线)
     * @return 返回更改的条数
     */
    public int upsertByPrimaryKey(String tableName, boolean underScoreToCamel, Object object, Map<String, Object> fieldNameAndValue) {

        // 判断输入的数据是否符合更新条件
        if (fieldNameAndValue == null || fieldNameAndValue.size() == 0) {
            throw new RuntimeException("向MySQL中更新数据异常,输入的更新条件没有指定数据,不能更新(这样更新会全表更新),传入的数据为:" + object);
        }

        // 将传入的object转换成json类型,并将传入的更新匹配字段和值(即fieldNameAndValue),添加到数据对象中(即data)
        JSONObject data = JSON.parseObject(JSON.toJSONString(object));
        for (Map.Entry<String, Object> entry : fieldNameAndValue.entrySet()) {
            data.put(entry.getKey(), entry.getValue());
        }
        data = JSON.parseObject(disposeSpecialCharacter(data));

        // 拼接SQL
        StringBuilder sql = new StringBuilder();
        sql.append(" INSERT INTO ").append(tableName);

        if (underScoreToCamel) {

            sql.append(" ( ");
            for (String key : data.keySet()) {
                sql.append(CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, key)).append(",");
            }
            sql.deleteCharAt(sql.length() - 1);
            sql.append(" ) ");

            sql.append(" values ");

            sql.append(" ( ");
            for (Object value : data.values()) {
                sql
                        .append("'")
                        .append(CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, value.toString()))
                        .append("'")
                        .append(",");
            }
            sql.deleteCharAt(sql.length() - 1);
            sql.append(" ) ");

            sql.append(" ON DUPLICATE KEY UPDATE ");

            for (Map.Entry<String, Object> entry : fieldNameAndValue.entrySet()) {
                sql
                        .append(CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, entry.getKey()))
                        .append(" = ")
                        .append("'")
                        .append(entry.getValue())
                        .append("'")
                        .append(",");
            }
            sql.deleteCharAt(sql.length() - 1);

        } else {

            sql.append(" ( ");
            for (String key : data.keySet()) {
                sql.append(key).append(",");
            }
            sql.deleteCharAt(sql.length() - 1);
            sql.append(" ) ");

            sql.append(" values ");

            sql.append(" ( ");
            for (Object value : data.values()) {
                sql
                        .append("'")
                        .append(value.toString())
                        .append("'")
                        .append(",");
            }
            sql.deleteCharAt(sql.length() - 1);
            sql.append(" ) ");

            sql.append(" ON DUPLICATE KEY UPDATE ");

            for (Map.Entry<String, Object> entry : fieldNameAndValue.entrySet()) {
                sql
                        .append(entry.getKey())
                        .append(" = ")
                        .append("'")
                        .append(entry.getValue())
                        .append("'")
                        .append(",");
            }
            sql.deleteCharAt(sql.length() - 1);

        }

        // 执行upsert操作
        try {
            return jdbcTemplate.update(sql.toString());
        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException(
                    "\r\n向MySQL数据库中 upsert 数据失败," +
                            "\r\n抛出的异常信息为:" + exception.getMessage() +
                            "\r\n执行的SQL为:" + sql
            );
        }

    }

    /**
     * 将传入的数据 upsert 到对应的MySQL的表中
     * 会根据MySQL表中的主键进行更新,如果该主键在表中有对应数据,就更新,没有就插入
     * 传入的数据中必须有主键
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象
     * @param fields            更新时匹配的字段名(如果underScoreToCamel为true,传入的字段为驼峰,如果underScoreToCamel为false,传入的字段为下划线)
     * @return 返回更改的条数
     */
    public int upsertByPrimaryKey(String tableName, boolean underScoreToCamel, Object object, String... fields) {

        // 将传入的对象转换成JSONObject格式
        JSONObject data = JSON.parseObject(disposeSpecialCharacter(object));

        // 根据传入的字段,获取要更新的主键值
        HashMap<String, Object> fieldNameAndValue = new HashMap<>();
        for (String field : fields) {
            if (underScoreToCamel) {
                field = field.contains("_") ? CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, field) : field;
            }
            fieldNameAndValue.put(field, data.getString(field));
        }

        // 调用重载函数,更新数据
        return upsertByPrimaryKey(tableName, underScoreToCamel, object, fieldNameAndValue);

    }

    /**
     * 将传入的数据 upsert 到对应的MySQL数据库的表中
     * 使用的是先用update进行数据更新,如果更新的条数为0,就进行插入
     * 如果表中包含主键,那在传入的数据中也必须要有主键
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象
     * @param fields            更新时匹配的字段名(如果underScoreToCamel为true,传入的字段为驼峰,如果underScoreToCamel为false,传入的字段为下划线)
     * @return 返回更改的条数
     */
    public int upsert(String tableName, boolean underScoreToCamel, Object object, String... fields) {

        int updateNum = update(tableName, underScoreToCamel, object, fields);

        if (updateNum == 0) {
            insert(tableName, underScoreToCamel, object);
            updateNum = 1;
        }

        return updateNum;
    }

    /**
     * 将传入的数据 upsert 到对应的MySQL数据库的表中
     * 使用的是先用update进行数据更新,如果更新的条数为0,就进行插入
     * 如果表中包含主键,那在传入的数据中也必须要有主键
     *
     * @param tableName         表名
     * @param underScoreToCamel 是否将驼峰转换为下划线
     * @param object            数据对象
     * @param fieldNameAndValue 更新时匹配的字段名(如果underScoreToCamel为true,传入的字段为驼峰,如果underScoreToCamel为false,传入的字段为下划线)
     * @return 返回更改的条数
     */
    public int upsert(String tableName, boolean underScoreToCamel, Object object, Map<String, Object> fieldNameAndValue) {

        int updateNum = update(tableName, underScoreToCamel, object, fieldNameAndValue);

        if (updateNum == 0) {
            insert(tableName, underScoreToCamel, object);
            updateNum = 1;
        }

        return updateNum;
    }

}

4. 如何使用

以在sink中使用该MySQL进行举例。

该代码实现的需求是:前序使用cdc功能从MySQL中采集到了binlog数据,现在将数据写入到另一个MySQL数据库中(该数据库中需要有对应的表,并且表结构和原数据库的表一模一样)

dwsStream
        .keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject element) throws Exception {
                // 根据传入的表名和主键进行分组
                StringBuilder result = new StringBuilder();
                result.append(element.getString("table")).append(" : ");
                for (Map.Entry<String, Object> entry : element.getJSONObject("primary_key").getInnerMap().entrySet()) {
                    result.append("key=>").append(entry.getKey()).append(" ; value=>").append(entry.getValue());
                }
                return result.toString();
            }
        })
        .addSink(new RichSinkFunction<JSONObject>() {

            // 声明MySQLUtil变量
            MySQLUtil mySQLUtil;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 在open方法中创建MySQLUtil工具类的对象
                mySQLUtil = new MySQLUtil(
                        ModelUtil.getConfigValue("mysql.yishou.data.url"),
                        ModelUtil.getConfigValue("mysql.yishou.data.username"),
                        ModelUtil.getConfigValue("mysql.yishou.data.password"),
                        2,
                        1
                );
            }

            @Override
            public void invoke(JSONObject element, Context context) throws Exception {
                long start = 0L;
                long end = 0L;
                try {
                    switch (element.getString("type")) {
                        // 对传入数据进行判断,如果是insert或者update就使用upsert系列方法更新数据
                        case "insert":
                        case "update":
                            start = System.currentTimeMillis();
                            mySQLUtil.upsertByPrimaryKey(
                                    element.getString("data_output_topic"),
                                    false,
                                    element.getJSONObject("data"),
                                    element.getJSONObject("primary_key").getInnerMap()
                            );
                            end = System.currentTimeMillis();
                            logger.info("向MySQL中upsert数据,耗时:{}, 操作的表名为:{}, 数据中的时间戳为:{}", (end - start), element.getString("data_output_topic"), element.getString("ts"));
                            break;
                        // 如果是delete类型数据,就使用delete方法删除数据
                        case "delete":
                            start = System.currentTimeMillis();
                            mySQLUtil.delete(element.getString("data_output_topic"), element.getJSONObject("primary_key").getInnerMap());
                            end = System.currentTimeMillis();
                            logger.info("向MySQL中delete数据,耗时:{}, 操作的表名为:{}, 数据中的时间戳为:{}", (end - start), element.getString("data_output_topic"), element.getString("ts"));
                            break;
                        // 当是其他类型的数据,不进行操作
                        default:
                            logger.error("无匹配,跳过,传入的数据为:{}", element.toJSONString());
                            break;
                    }
                } catch (Exception e) {
                    throw new RuntimeException(" ==> " + element);
                }

            }
        })

5. Hikari连接池各配置说明

 
# Hikari will use the above plus the following to setup connection pooling
spring.datasource.type=com.zaxxer.hikari.HikariDataSource
 
#最小空闲连接,默认值10,小于0或大于maximum-pool-size,都会重置为maximum-pool-size
spring.datasource.hikari.minimum-idle=5
 
#最大连接数,小于等于0会被重置为默认值10;大于零小于1会被重置为minimum-idle的值
spring.datasource.hikari.maximum-pool-size=15
 
#自动提交从池中返回的连接,默认值为true
spring.datasource.hikari.auto-commit=true
 
#空闲连接超时时间,默认值600000(10分钟),大于等于max-lifetime且max-lifetime>0,会被重置为0;不等于0且小于10秒,会被重置为10秒。
#只有空闲连接数大于最大连接数且空闲时间超过该值,才会被释放
spring.datasource.hikari.idle-timeout=30000
 
#连接池名称,默认HikariPool-1
spring.datasource.hikari.pool-name=Hikari
 
#连接最大存活时间.不等于0且小于30秒,会被重置为默认值30分钟.设置应该比mysql设置的超时时间短;单位ms
spring.datasource.hikari.max-lifetime=55000
 
#连接超时时间:毫秒,小于250毫秒,会被重置为默认值30秒
spring.datasource.hikari.connection-timeout=30000
 
#连接测试查询
spring.datasource.hikari.connection-test-query=SELECT 1

6. 注意点

在使用上述方法时,因为是手动拼接SQL,并且会过滤值为空的字段,这样的话,需要在创建MySQL表时,不能设置该字段not null,或者设置该字段为not null但给该字段默认值。

但是在MySQL8.x版本中,对一下类型,比如json类型等不能设置默认值,此时可以在建表的时候先设置 sql_mode 为 空,然后在建表,就可以给这些字段设置默认值了。

默认的sql_mode 的值:

 

设置sql_mode 为 空(注意:不修改配置文件的话,只在此会话中生效):

 

7. 静态MySQLUtil工具类

在上述的工具类中,每次创建都需要进行new MySQLUtil()操作,来设置需要操作的MySQL的具体属性以及连接属性,这在将数据写入到MySQL库中是比较方便的;但当我们需要进行维度匹配时,一个作业中可能很多算子都是用的一个库,这样我们每个算子中都创建一个MySQL连接客户端就比较麻烦了,而且可能也会浪费资源。

这时我们就可以将这个库的MySQL创建成一个静态的工具类,这样在所有作业中就不需要创建了,可以直接使用;而且因为静态的特性,在每台机器上只会创建一个连接池,这样也比较能节省资源(当然,连接池的大小不能跟上述设置成最小为1,最大为2一样,博主在生产环境中一般设置为最小为5,最大为50,这样在一台机器上设置4个并发,经过博主的压测,最高查询次数为1500次/秒)。

具体代码如下所示(因为主要是维度匹配,所以全部都是查询方法):

package com.yishou.bigdata.common.utils;

import com.google.common.base.CaseFormat;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.commons.beanutils.BeanUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @date: 2021/7/5
 *  @Author ddkk.com  弟弟快看,程序员编程资料站
 * @desc: 对数据库R7(MySQL)的工具类
 */
public class MySQLR7Util {

    static Logger logger = LoggerFactory.getLogger(MySQLR7Util.class);

    /**
     * jdbcTemplate
     */
    private static JdbcTemplate jdbcTemplate;

    /**
     * 使用单例模式获取MySQL数据库R7的连接
     *
     * @return jdbc连接
     */
    public static JdbcTemplate getJdbcTemplate() {
        if (jdbcTemplate == null) {
            synchronized (MySQLR7Util.class) {
                if (jdbcTemplate == null) {
                    try {
                        HikariDataSource ds = new HikariDataSource();
                        ds.setDriverClassName("com.mysql.cj.jdbc.Driver");
                        ds.setJdbcUrl(ModelUtil.getConfigValue("mysql.r7.yishou.url"));
                        ds.setUsername(ModelUtil.getConfigValue("mysql.r7.username"));
                        ds.setPassword(ModelUtil.getConfigValue("mysql.r7.password"));
                        ds.setMaximumPoolSize(50);
                        ds.setMinimumIdle(5);
                        jdbcTemplate = new JdbcTemplate(ds);
                        logger.info(
                                "##### 使用HikariPool连接池初始化JdbcTemplate成功,使用的URL为:{} , 其中最大连接大小为:{} , 最小连接大小为:{} ;",
                                ds.getJdbcUrl(),
                                ds.getMaximumPoolSize(),
                                ds.getMinimumIdle()
                        );
                    } catch (Exception e) {
                        e.printStackTrace();
                        throw new RuntimeException("创建 MySQL R7 数据库连接失败");
                    }
                }
            }
        }
        return jdbcTemplate;
    }

    /**
     * 处理传入数据中的特殊字符(例如: 单引号)
     *
     * @param data 传入的数据
     * @return 返回的结果
     */
    public static String disposeSpecialCharacter(String data) {

        // 处理其中的单引号
        data = data.replace("'", "''");

        // 返回结果
        return data;

    }

    /**
     * 通过传入的表名,主键key,主键value,样例类 和 需要的值,获取对应的数据
     *
     * @param tableName    表名
     * @param primaryKey   主键字段
     * @param primaryValue 主键值
     * @param clz          返回的数据类型
     * @param fields       需要的字段名
     * @param <T>          样例类
     * @return 样例类集合
     */
    public static <T> List<T> queryListByKey(String tableName, String primaryKey, String primaryValue, Class<T> clz, String... fields) {

        // 拼接SQL
        String sql = " select " +
                Arrays.stream(fields).map(String::valueOf).collect(Collectors.joining(",")) +
                " from " + tableName +
                " where " + primaryKey + " = '" + disposeSpecialCharacter(primaryValue) + "'";

        // 执行SQL并返回结果
        return queryList(sql, clz);

    }

    /**
     * 如果传入的clz中的属性又包含对象,会报错,此时传入JSONObject对象即可
     *
     * @param sql 执行的查询语句
     * @param clz 返回的数据类型
     * @param <T> 样例类
     * @return 样例类集合
     */
    public static <T> List<T> queryList(String sql, Class<T> clz) {
        try {
            List<Map<String, Object>> mapList = MySQLR7Util.getJdbcTemplate().queryForList(sql);
            List<T> resultList = new ArrayList<>();
            for (Map<String, Object> map : mapList) {
                Set<String> keys = map.keySet();
                // 当返回的结果中存在数据,通过反射将数据封装成样例类对象
                T result = clz.newInstance();
                for (String key : keys) {
                    BeanUtils.setProperty(
                            result,
                            key,
                            map.get(key)
                    );
                }
                resultList.add(result);
            }
            return resultList;
        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException(
                    "\r\n从 MySQL R7 数据库中 查询 数据失败," +
                            "\r\n抛出的异常信息为:" + exception.getMessage() +
                            "\r\n查询的SQL为:" + sql
            );
        }
    }

    /**
     * 如果传入的clz中的属性又包含对象,会报错,此时传入JSONObject对象即可
     *
     * @param sql               执行的查询语句
     * @param underScoreToCamel 是否将下划线转换为驼峰命名法
     * @param clz               返回的数据类型
     * @param <T  样例类
     * @return 样例类集合
     */
    public static <T> List<T> queryList(String sql, boolean underScoreToCamel, Class<T> clz) {
        try {
            List<Map<String, Object>> mapList = MySQLR7Util.getJdbcTemplate().queryForList(sql);
            List<T> resultList = new ArrayList<>();
            for (Map<String, Object> map : mapList) {
                Set<String> keys = map.keySet();
                // 当返回的结果中存在数据,通过反射将数据封装成样例类对象
                T result = clz.newInstance();
                for (String key : keys) {
                    String propertyName = underScoreToCamel ? CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, key) : key;
                    BeanUtils.setProperty(
                            result,
                            propertyName,
                            map.get(key)
                    );
                }
                resultList.add(result);
            }
            return resultList;
        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException(
                    "\r\n从 MySQL R7 数据库中 查询 数据失败," +
                            "\r\n抛出的异常信息为:" + exception.getMessage() +
                            "\r\n查询的SQL为:" + sql
            );
        }
    }

}