|
|
@@ -0,0 +1,220 @@
|
|
|
+package com.qucheng.game.data.oldsystem.sink;
|
|
|
+
|
|
|
+import com.alibaba.fastjson2.JSON;
|
|
|
+import com.alibaba.fastjson2.JSONObject;
|
|
|
+import org.apache.commons.dbcp2.BasicDataSource;
|
|
|
+import org.apache.commons.lang3.StringUtils;
|
|
|
+import org.apache.flink.configuration.Configuration;
|
|
|
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
|
|
|
+
|
|
|
+import java.sql.Connection;
|
|
|
+import java.sql.PreparedStatement;
|
|
|
+import java.sql.SQLException;
|
|
|
+import java.util.Collection;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.Set;
|
|
|
+
|
|
|
+import static com.qucheng.game.data.oldsystem.common.qcConfig.*;
|
|
|
+
|
|
|
+public class ads_sink extends RichSinkFunction<JSONObject> {
|
|
|
+
|
|
|
+ // private DruidDataSource druidDataSource = null;
|
|
|
+ //设置连接信息
|
|
|
+
|
|
|
+ PreparedStatement ps;
|
|
|
+ BasicDataSource dataSource;
|
|
|
+ private Connection connection;
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void open(Configuration parameters) throws Exception {
|
|
|
+ super.open(parameters);
|
|
|
+ dataSource = new BasicDataSource();
|
|
|
+ connection = getConnection(dataSource);
|
|
|
+ connection.setAutoCommit(true);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void invoke(JSONObject values, Context context) throws SQLException {
|
|
|
+ //拿到表名和数据
|
|
|
+ String sinkTable = values.getString("tableName");
|
|
|
+ JSONObject data = values.getJSONObject("after");
|
|
|
+ String database = values.getString("db");
|
|
|
+ JSONObject key = values.getJSONObject("key");
|
|
|
+
|
|
|
+ try {
|
|
|
+ //执行具体操作
|
|
|
+ if (values.getString("type").equals("insert") || values.getString("type").equals("update") ) {
|
|
|
+ upsert(connection, database, sinkTable, data,key.toString());
|
|
|
+ } else if (values.getString("type").equals("delete")) {
|
|
|
+ deleteValues(connection, database, sinkTable, key.toString());
|
|
|
+ } else {
|
|
|
+ System.out.println("该数据不属于插入，删除和更新" + values);
|
|
|
+ }
|
|
|
+ } catch (SQLException e) {
|
|
|
+ e.printStackTrace();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ //实现upsert方法。如果无更新则插入
|
|
|
+ public void upsert(Connection connection, String database, String sinkTable, JSONObject data, String key) throws SQLException {
|
|
|
+
|
|
|
+ StringBuffer prefix = new StringBuffer();
|
|
|
+ StringBuffer suffix = new StringBuffer();
|
|
|
+
|
|
|
+ JSONObject dataKey = JSON.parseObject(key);
|
|
|
+
|
|
|
+ for (Map.Entry<String, Object> entry : dataKey.entrySet()) {
|
|
|
+ String colName = entry.getKey();
|
|
|
+ Object colType = entry.getValue();
|
|
|
+
|
|
|
+ if (colType.getClass().getName().equals("java.lang.String")) {
|
|
|
+ suffix.append("`" + colName + "`" + " = ").append("'").append(colType).append("' and ");
|
|
|
+ } else {
|
|
|
+ suffix.append("`" + colName + "`" + " = ").append(colType).append(" and ");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ for (Map.Entry<String, Object> entry : data.entrySet()) {
|
|
|
+ String colName = entry.getKey();
|
|
|
+ Object colType = entry.getValue();
|
|
|
+
|
|
|
+ if (entry.getValue() == null) {
|
|
|
+ prefix.append("`" + colName + "`" + " = ").append(colType).append(" ,");
|
|
|
+ }else if(colType.getClass().getName().equals("java.lang.String")){
|
|
|
+ prefix.append("`" + colName + "`" + " = '").append(colType).append("' ,");
|
|
|
+ }else {
|
|
|
+ prefix.append("`" + colName + "`" + " = '").append(colType).append("' ,");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ //减去最后的and
|
|
|
+ String suffixSql = suffix.substring(0, suffix.length() - 4);
|
|
|
+
|
|
|
+ //减去最后的逗号
|
|
|
+ String prefixSql = prefix.substring(0, prefix.length() - 1);
|
|
|
+
|
|
|
+ String sql = "update " + "`"+database+"`" + "." + "`"+sinkTable +"`" + " set " + prefixSql + " where " + suffixSql;
|
|
|
+// System.out.println(sql);
|
|
|
+
|
|
|
+ //2.预编译SQL
|
|
|
+ ps = connection.prepareStatement(sql);
|
|
|
+
|
|
|
+ //3.执行
|
|
|
+ ps.executeUpdate();
|
|
|
+
|
|
|
+ if (ps.getUpdateCount() == 0) {
|
|
|
+ StringBuffer strSqlValue = new StringBuffer();
|
|
|
+
|
|
|
+ Set<String> columns = data.keySet();
|
|
|
+ Collection<Object> values = data.values();
|
|
|
+
|
|
|
+ for (Object col : values) {
|
|
|
+ if (col == null) {
|
|
|
+ strSqlValue.append(col).append(",");
|
|
|
+ }else if (col.getClass().getName().equals("java.lang.String")) {
|
|
|
+ strSqlValue.append("'").append(col).append("',");
|
|
|
+ } else {
|
|
|
+ strSqlValue.append(col).append(",");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ String join = "`" + StringUtils.join(columns, "`,`") + "`";
|
|
|
+
|
|
|
+
|
|
|
+ String insertsql = "insert into " + "`"+database+"`" + "." + "`"+sinkTable +"`"+ "(" +
|
|
|
+ join+ ") values (" +strSqlValue.substring(0, strSqlValue.length() - 1) + ");";
|
|
|
+
|
|
|
+ ps = connection.prepareStatement(insertsql);
|
|
|
+// System.out.println(insertsql);
|
|
|
+
|
|
|
+ //3.执行
|
|
|
+ ps.execute();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ //执行删除操作
|
|
|
+ //delete from glm_test_cdc1.test2 where id = 1 and user_id = '1'
|
|
|
+ public void deleteValues(Connection connection, String database, String sinkTable, String key) throws SQLException {
|
|
|
+
|
|
|
+ JSONObject dataKey = JSON.parseObject(key);
|
|
|
+ StringBuffer strSqlValue = new StringBuffer();
|
|
|
+
|
|
|
+ for (Map.Entry<String, Object> entry : dataKey.entrySet()) {
|
|
|
+ String colName = entry.getKey();
|
|
|
+ Object colType = entry.getValue();
|
|
|
+
|
|
|
+ if (colType.getClass().getName().equals("java.lang.String")) {
|
|
|
+ strSqlValue.append("`" + colName + "`" + " = ").append("'").append(colType).append("' and ");
|
|
|
+ } else {
|
|
|
+ strSqlValue.append("`" + colName + "`" + " = ").append(colType).append(" and ");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ String deleteSql = strSqlValue.substring(0, strSqlValue.length() - 4);
|
|
|
+
|
|
|
+ //拼接删除语句
|
|
|
+ String sql = "delete from " + database + "." + sinkTable + " where " + deleteSql;
|
|
|
+
|
|
|
+ //2.预编译SQL
|
|
|
+ ps = connection.prepareStatement(sql);
|
|
|
+
|
|
|
+ //3.执行
|
|
|
+ ps.execute();
|
|
|
+
|
|
|
+// System.out.println("-------delete-------");
|
|
|
+
|
|
|
+// System.out.println(sql);
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+ private static Connection getConnection(BasicDataSource dataSource) {
|
|
|
+ dataSource.setDriverClassName(MYSQL_DRIVER);
|
|
|
+ //注意，替换成自己本地的 mysql 数据库地址和用户名、密码
|
|
|
+ dataSource.setUrl(MYSQL_URL_BIGDATA); //test为数据库名
|
|
|
+ dataSource.setUsername(MYSQL_USERNAME_BIGDATA); //数据库用户名
|
|
|
+ dataSource.setPassword(MYSQL_PASSWORD_BIGDATA); //数据库密码
|
|
|
+ //设置连接池的一些参数
|
|
|
+ // 设置初始化连接池时池中连接的数量
|
|
|
+ dataSource.setInitialSize(10);
|
|
|
+ dataSource.setMaxTotal(20);
|
|
|
+ // 设置空闲时的最小连接数，必须介于 0 和最大连接数之间，默认为 0
|
|
|
+ dataSource.setMinIdle(1);
|
|
|
+ dataSource.setValidationQuery("select 1");
|
|
|
+ dataSource.setTestWhileIdle(true);
|
|
|
+ dataSource.setTestOnBorrow(false);
|
|
|
+ dataSource.setTestOnReturn(false);
|
|
|
+ // 设置空闲连接回收器每隔 30s 运行一次
|
|
|
+ dataSource.setTimeBetweenEvictionRunsMillis(3 * 1000L);
|
|
|
+ // 设置池中连接空闲 30min 被回收，默认值即为 30 min
|
|
|
+ dataSource.setMinEvictableIdleTimeMillis(30 * 60 * 1000L);
|
|
|
+
|
|
|
+ Connection con = null;
|
|
|
+ try {
|
|
|
+ con = dataSource.getConnection();
|
|
|
+ System.out.println("创建连接池：" + con);
|
|
|
+ } catch (Exception e) {
|
|
|
+ System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());
|
|
|
+ }
|
|
|
+ return con;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void close() throws Exception {
|
|
|
+ super.close();
|
|
|
+
|
|
|
+ //关闭连接和释放资源
|
|
|
+ if (connection != null) {
|
|
|
+ connection.close();
|
|
|
+ }
|
|
|
+ if (ps != null) {
|
|
|
+ ps.close();
|
|
|
+ }
|
|
|
+ if (dataSource != null) {
|
|
|
+ dataSource.close();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+}
|
