package com.qniao.iot; import cn.hutool.core.bean.BeanUtil; import cn.hutool.core.util.StrUtil; import com.alibaba.druid.pool.DruidDataSource; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; import org.apache.flink.streaming.api.functions.sink.SinkFunction; import java.sql.Connection; import java.sql.Date; import java.sql.PreparedStatement; import java.text.SimpleDateFormat; import java.util.List; public class SinkMysqlFunc extends RichSinkFunction> { private DruidDataSource dataSource = null; /** * 初始化方法 在invoke前执行 * * @param parameters * @throws Exception */ @Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 获取数据库连接池配置 此处省略 if (dataSource == null) { dataSource = DruidDataSourceUtil.getInstance(); } } @Override public void invoke(List values, Context context) throws Exception { if (values.size() != 0) { Connection connection = dataSource.getConnection(); PreparedStatement ps = connection.prepareStatement(getSql()); for (Body body : values) { if (body != null) { ps.setInt(1, body.getData_source()); ps.setLong(2, body.getMac()); Integer dataType = body.getData_type(); ps.setInt(3, dataType == 1 ? 0 : 1); ps.setInt(4, dataType == 2 ? 1 : 0); ps.setLong(5, body.getQuantity()); ps.setLong(6, body.getQuantity()); ps.setBigDecimal(7, body.getRunningDuration()); ps.setBigDecimal(8, body.getWaitingDuration()); ps.setBigDecimal(9, null); ps.setBigDecimal(10, null); String createTimeStr = body.getCreate_time(); Date createDate = null; if (StrUtil.isNotEmpty(createTimeStr)) { long time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(createTimeStr).getTime(); createDate = new Date(time); } ps.setDate(11, createDate); Long id = body.getId(); ps.setString(12, id == null ? null : StrUtil.toString(id)); ps.setString(13, body.getDocId()); ps.addBatch(); } } ps.executeBatch(); if (connection != null) { connection.close(); } if (ps != null) { ps.close(); } } } @Override public void close() throws Exception { super.close(); //关闭连接和释放资源 if (dataSource != null) { dataSource.close(); } } private String getSql() { return "insert into qn_cloud_box_event(data_source, machine_iot_mac, machine_pwr_stat, machine_working_stat, acc_job_count,\n" + " curr_job_count, curr_job_duration, curr_waiting_duration, curr_stoping_duration, ig_stat,\n" + " report_time,event_id, doc_id)\n" + "values (?, ?, ?, ?, ?,?, ?, ?, ?, ?, ?, ?, ?)"; } }