public class Producer {

    // 定义TOPIC
    public final static String TOPIC = "bank";

    // 对发送数据条目进行计数，验证发送方和接收方数据的一致性，保证没有重复消费或漏消费
    public static int cal = 0;

    public static void main(String[] args) throws IOException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        // 删除TOPIC及TOPIC下之前生产的log
        AdminClient kafkaAdminClient = KafkaAdminClient.create(props);
        List<String> topics = new ArrayList<>();
        topics.add(TOPIC);
        kafkaAdminClient.deleteTopics(topics);

        // 设置生产者属性
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("group.id", "191250048");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // 从文本读取数据
        BufferedInputStream bis = new BufferedInputStream(new FileInputStream("/home/approdite/IdeaProjects/kafka/bank2.txt"));
        int bufferSize = 10 * 1024 * 1024;
        BufferedReader in = new BufferedReader(new InputStreamReader(bis), bufferSize);
        // 设置log线程，每两秒输出当前生产数据的条目信息
        Thread log = new Thread(() -> {
            int times = 0, last = 0;
            while (true) {
                if (last == cal) {
                    times++;
                    // 连续5次，生产数据的条目没有变化则认为生产结束，log线程停止
                    if (times >= 5) {
                        break;
                    }
                } else {
                    last = cal;
                    times = 0;
                }
                System.err.println("produce message :" + last);
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.err.println("retry over 5 times, producing could be finished!");
        });
        log.start();
        // 以行为单位读取并生产数据
        String line;
        while ((line = in.readLine()) != null && !line.isEmpty()) {
            ++cal;
            producer.send(new ProducerRecord<>(TOPIC, line));
        }
        in.close();
        producer.close();
        kafkaAdminClient.close();
    }
}

public class Consumer {

// 	  // 使用OutputTag将可视化需要的shop做额外的侧输出，用于生成可视化需要的数据
//    public static OutputTag<v_tr_shop_mx> statistic = new OutputTag<v_tr_shop_mx>("id") {
//    };
//	  // 为OutputTag添置数据
//    public static final class Processor extends ProcessFunction<Insert, Insert> {
//
//        @Override
//        public void processElement(Insert insert, Context context, Collector<Insert> collector) {
//            context.output(statistic, (v_tr_shop_mx) insert);
//            collector.collect(insert);
//        }
//    }

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.enableCheckpointing(5000);
        // 设置消费者，从最早开始拉取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("auto.offset.reset", "earliest");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                Producer.TOPIC,
                new SimpleStringSchema(),
                properties);
//        consumer.setStartFromGroupOffsets();
        consumer.setStartFromEarliest();
        final String KEY = "eventType";
        final String BODY = "eventBody";
        // 将数据源设置为kafka消费者
        DataStreamSource<String> sourceStream = env.addSource(consumer).setParallelism(1);
        // 将接受的数据转换为json对象
        SingleOutputStreamOperator<JSONObject> jsonObjectStream = sourceStream.map(JSONObject::parseObject);
        // 表驱动，根据eventType将不同的json对象转化为对应的java类
        String[] keys = new String[]{"shop", "sjyh", "sdrq", "sbyb", "sa", "huanx", "huanb", "gzdf", "grwy", "etc", "duebill", "dsf", "djk", "contract"};
        Class<Insert>[] classes = new Class[]{v_tr_shop_mx.class, dm_v_tr_sjyh_mx.class, dm_v_tr_sdrq_mx.class, dm_v_tr_sbyb_mx.class, dm_v_tr_sa_mx.class, dm_v_tr_huanx_mx.class, dm_v_tr_huanb_mx.class, dm_v_tr_gzdf_mx.class, dm_v_tr_grwy_mx.class, dm_v_tr_etc_mx.class, dm_v_tr_duebill_mx.class, dm_v_tr_dsf_mx.class, dm_v_tr_djk_mx.class, dm_v_tr_contract_mx.class};
        // 根据eventType分流，给不同的java设置自身的sink方法
        for (int i = 0; i < keys.length; i++) {
            int finalI = i;
            SingleOutputStreamOperator<Insert> pojoStream = jsonObjectStream
                    .filter(e -> {
                        try {
                            return e.get(KEY).equals(keys[finalI]);
                        } catch (Exception s) {
                            System.err.println(e);
                            throw s;
                        }
                    })
                    .map(jsonObject ->
                    {
                        try {
                            return jsonObject.getJSONObject(BODY).toJavaObject(classes[finalI]);
                        } catch (JSONException e) {
                            System.err.println(jsonObject);
                            throw e;
                        }
                    });
            pojoStream.addSink(new POJOInsertSink());
        }
//		  // 使用OutputTag分流shop流，设置额外的sink方法，为可视化提供快速查询需要的数据
//        SingleOutputStreamOperator<Insert> pojoStream = jsonObjectStream
//                .filter(e -> e.get(KEY).equals("shop"))
//                .map(jsonObject -> (Insert) jsonObject.getJSONObject(BODY).toJavaObject(v_tr_shop_mx.class)).process(new Processor());
//        DataStream<v_tr_shop_mx> output = pojoStream.getSideOutput(statistic);
//        output.addSink(new StatisticSink());
//        pojoStream.addSink(new POJOInsertSink());
        try {
            env.execute("consume start");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class POJOInsertSink extends RichSinkFunction<Insert> {

    // 数据库连接
    Connection connection = null;

    // 批量插入的下限，不应超过生产数据的批量（16384），
    // 设置较大时，插入效率更高，但插入频率降低不利于可视化，在可视化展示时可合理降低standard值，建议为512
    final int standard = 5120;

    // 为每种POJO维护自己的PreparedStatement，事实上每个POJO都有自己的Sink，所以不会获得其他POJO的PreparedStatement，故使用Map没有意义，可删去。
    Map<Class<?>, PreparedStatement> stats = new ConcurrentHashMap<>();
    
    // 同上，为每种POJO维护自己的PreparedStatement已经加入Batch的条目。
    Map<Class<?>, Integer> count = new ConcurrentHashMap<>();

    // 连接到clickhouse数据库，取消自动提交
    @Override
    public void open(Configuration parameters) {
        try {
            super.open(parameters);
            Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
            // compress 默认为 true，但当前版本的clickhouse对compress支持不佳可能会报错，这里设置为false
            // String url = "jdbc:clickhouse://localhost:8123/dm?compress=false";
            String url = "jdbc:clickhouse://124.222.139.8:8123/dm?compress=false";
            connection = DriverManager.getConnection(url);
            connection.setAutoCommit(false);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 关闭数据库连接，但是此时可能还存在PreparedStatement没有达到批量插入的standard，此时也将其执行
    @Override
    public void close() throws Exception {
        for (Map.Entry<Class<?>, PreparedStatement> entry : stats.entrySet()) {
            if (count.get(entry.getKey()) > 0) {
                long startTime = System.currentTimeMillis();
                int size = entry.getValue().executeBatch().length;
                connection.commit();
                long endTime = System.currentTimeMillis();
//                System.out.println("批量插入完毕用时：" + (endTime - startTime) + " -- 插入数据(" + entry.getKey().getSimpleName() + ") = " + size);
                entry.getValue().clearBatch();
            }
        }
        super.close();
        if (connection != null) {
            connection.close();
        }
    }

    @Override
    public void invoke(Insert pojo, Context context) {
        try {
            // 获得POJO对应的PreparedStatement和已准备条目
            Class<?> pclass = pojo.getClass();
            stats.putIfAbsent(pclass, pojo.stat(connection));
            PreparedStatement preparedStatement = stats.get(pclass);
            pojo.insert(preparedStatement);
            preparedStatement.addBatch();
            count.putIfAbsent(pclass, 0);
            int cnt = count.get(pclass) + 1;
            // 当条目数大于等于standard，则批量执行插入语句
            if (cnt >= standard) {
                long startTime = System.currentTimeMillis();
                int size = preparedStatement.executeBatch().length;
                connection.commit();
                long endTime = System.currentTimeMillis();
//                System.out.println("批量插入完毕用时：" + (endTime - startTime) + " -- 插入数据(" + pclass.getSimpleName() + ") = " + size);
                preparedStatement.clearBatch();
                cnt = 0;
            }
            count.put(pclass, cnt);
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

public interface Insert {
    // 构造POJO的PreparedStatement
    PreparedStatement stat(Connection con);

    // 为PreparedStatement赋值
    PreparedStatement insert(PreparedStatement preparedStatement);
}

// 根据格式化后的SQL构造
public class POJOBuilder {
    
    public static void main(String[] args) throws IOException {
        // 读取ch.sql
        File sql = new File(ClassLoader.getSystemResource("ch.sql").getPath());
        if (sql.exists()) {
            BufferedReader in = new BufferedReader(new FileReader(sql));
            File pojo;
            String line;
            BufferedWriter out = null;
            String className = null;
            String fullName = null;
            // 维护POJO的属性和类型之间的对应关系
            Map<String, Class<?>> properties = null;
            while ((line = in.readLine()) != null) {
                line = line.trim().toLowerCase();
                // 读到建表语句，说明应当新写一个POJO类，此时记录POJO的类名，并设置BufferedWriter指向
                if (line.startsWith("create table")) {
                    line = line.substring(12).trim();
                    fullName = line;
                    className = line.substring(line.lastIndexOf('.') + 1);
                    pojo = new File(System.getProperties().getProperty("user.dir") + File.separator + "src" + File.separator + "main" + File.separator + "java" + File.separator + "pojo" + File.separator + className + ".java");
                    if ((!pojo.exists() || pojo.exists() && pojo.delete()) && pojo.createNewFile()) {
                        out = new BufferedWriter(new FileWriter(pojo));
                    }
                    properties = new HashMap<>();
                } else if (line.contains("string")) {
                    // 读到建表语句中的String，说明该属性是String类型
                    String propertiesName = line.split("\\s+")[0];
                    assert properties != null;
                    properties.put(propertiesName, String.class);
                } else if (line.contains("decimal")) {
                    // 读到建表语句中的decimal，说明该属性是Double类型
                    String propertiesName = line.split("\\s+")[0];
                    assert properties != null;
                    properties.put(propertiesName, Double.class);
                } else if (line.contains("int")) {
                    // 读到建表语句中的int，说明该属性是Integer类型
                    String propertiesName = line.split("\\s+")[0];
                    assert properties != null;
                    properties.put(propertiesName, Integer.class);
                } else if (line.startsWith(")")) {
                    // 读到），说明建表语句结束，获取POJO信息完毕，此时开始将信息写入POJO类
                    StringBuilder content = new StringBuilder();
                    // 定义换行符号和双换号符号
                    String lineSeparator = System.lineSeparator(), doubleLineSeparator = lineSeparator + lineSeparator;
                    // init，POJO的包、依赖关系及类名
                    {
                        assert out != null;
                        content.append("package pojo;").append(doubleLineSeparator);
                        content.append("import lombok.Data;").append(lineSeparator);
                        content.append("import java.sql.Connection;").append(lineSeparator);
                        content.append("import java.sql.PreparedStatement;").append(lineSeparator);
                        content.append("import java.sql.SQLException;").append(doubleLineSeparator);
                        content.append("@Data").append(lineSeparator);
                        content.append("public class ").append(className).append(" implements Insert {").append(doubleLineSeparator);
                    }
                    // properties，POJO的属性，并为double和int值赋初值（避免json空值可能造成的问题）
                    {
                        for (Map.Entry<String, Class<?>> entry : properties.entrySet()) {
                            content.append("\tpublic ").append(entry.getValue().getSimpleName()).append(" ").append(entry.getKey());
                            if (entry.getValue().equals(Integer.class)) {
                                content.append(" = 0");
                            } else if (entry.getValue().equals(Double.class)) {
                                content.append(" = 0.0");
                            }
                            content.append(";").append(doubleLineSeparator);
                        }
                    }
                    // stat & insert，实现insert接口
                    {
                        content.append("\t@Override").append(lineSeparator);
                        content.append("\tpublic PreparedStatement stat(Connection con) {").append(lineSeparator);
                        content.append("\t\ttry {").append(lineSeparator);
                        content.append("\t\t\treturn con.prepareStatement(\"INSERT INTO ").append(fullName).append(" (");
                        int index = 0;
                        // 根据属性拼接PreparedStatement需要的的sql语句
                        for (String propertiesName : properties.keySet()) {
                            content.append(propertiesName);
                            if (++index != properties.size()) content.append(", ");
                            else content.append(") VALUES (");
                        }
                        while (index-- > 1) {
                            content.append("?, ");
                        }
                        content.append("?)\");").append(lineSeparator);
                        content.append("\t\t} catch (SQLException e) {").append(lineSeparator);
                        content.append("\t\t\te.printStackTrace();").append(lineSeparator);
                        content.append("\t\t}").append(lineSeparator);
                        content.append("\t\treturn null;").append(lineSeparator);
                        content.append("\t}").append(doubleLineSeparator);

                        // 根据属性为PreparedStatement赋值
                        content.append("\t@Override").append(lineSeparator);
                        content.append("\tpublic PreparedStatement insert(PreparedStatement preparedStatement) {").append(lineSeparator);
                        content.append("\t\ttry {").append(lineSeparator);
                        for (Map.Entry<String, Class<?>> entry : properties.entrySet()) {
                            if (entry.getValue().equals(Integer.class))
                                content.append("\t\t\tpreparedStatement.setInt").append('(').append(++index).append(", ").append(entry.getKey()).append(");").append(lineSeparator);
                            else
                                content.append("\t\t\tpreparedStatement.set").append(entry.getValue().getSimpleName()).append('(').append(++index).append(", ").append(entry.getKey()).append(");").append(lineSeparator);
                        }
                        content.append("\t\t\treturn preparedStatement;").append(lineSeparator);
                        content.append("\t\t} catch (SQLException e) {").append(lineSeparator);
                        content.append("\t\t\te.printStackTrace();").append(lineSeparator);
                        content.append("\t\t}").append(lineSeparator);
                        content.append("\t\treturn preparedStatement;").append(lineSeparator);
                        content.append("\t}").append(lineSeparator);
                    }
                    // finish
                    {
                        content.append("}").append(lineSeparator);
                        out.write(content.toString());
                        out.close();
                    }
                }
            }
        }
    }
}

package pojo;

import lombok.Data;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

@Data
public class v_tr_shop_mx implements Insert {

	public String tran_date;

	public String tran_channel;

	public String shop_code;

	public String hlw_tran_type;

	public String tran_time;

	public String shop_name;

	public String order_code;

	public String uid;

	public Double score_num = 0.0;

	public String current_status;

	public String pay_channel;

	public Double tran_amt = 0.0;

	public String legal_name;

	public String etl_dt;

	@Override
	public PreparedStatement stat(Connection con) {
		try {
			return con.prepareStatement("INSERT INTO dm.v_tr_shop_mx (tran_date, tran_channel, shop_code, hlw_tran_type, tran_time, shop_name, order_code, uid, score_num, current_status, pay_channel, tran_amt, legal_name, etl_dt) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)");
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public PreparedStatement insert(PreparedStatement preparedStatement) {
		try {
			preparedStatement.setString(1, tran_date);
			preparedStatement.setString(2, tran_channel);
			preparedStatement.setString(3, shop_code);
			preparedStatement.setString(4, hlw_tran_type);
			preparedStatement.setString(5, tran_time);
			preparedStatement.setString(6, shop_name);
			preparedStatement.setString(7, order_code);
			preparedStatement.setString(8, uid);
			preparedStatement.setDouble(9, score_num);
			preparedStatement.setString(10, current_status);
			preparedStatement.setString(11, pay_channel);
			preparedStatement.setDouble(12, tran_amt);
			preparedStatement.setString(13, legal_name);
			preparedStatement.setString(14, etl_dt);
			return preparedStatement;
		} catch (SQLException e) {
			e.printStackTrace();
		}
		return preparedStatement;
	}
}

@Deprecated
public class StatisticSink extends RichSinkFunction<v_tr_shop_mx> {

    Connection connection = null;

    // 每intervals仅执行一次update
    long intervals = 2000;

    long lastModify = 0;

    PreparedStatement insert;

    PreparedStatement update;

    // 未执行的update的统计
    int count = 0;

    // 上一个shop的日期
    String lastDay;

    @Override
    public void open(Configuration parameters) {
        try {
            super.open(parameters);
            Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
            String url = "jdbc:clickhouse://localhost:8123/dm?compress=false";
            connection = DriverManager.getConnection(url);
            connection.setAutoCommit(false);
            // 预编译sql语句
            insert = connection.prepareStatement("INSERT INTO dm.statistic_mx(tran_date, count) VALUES (?,?)");
            update = connection.prepareStatement("ALTER TABLE dm.statistic_mx UPDATE count = ((SELECT count from dm.statistic_mx where tran_date = (?)) + (?)) WHERE tran_date = (?)");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        // 如果结束时仍有update未执行，则执行之
        if (count > 0) {
            update.setInt(2, count);
            update.execute();
            connection.commit();
        }
        super.close();
        if (connection != null) {
            connection.close();
        }
    }

    @Override
    public void invoke(v_tr_shop_mx shop, Context context) {
        try {
            String date = shop.tran_date;
            if (lastDay == null) {
            	// 第一次语句应为insert
                lastDay = date;
                insert.setString(1, date);
                insert.setInt(2, 1);
                insert.executeUpdate();
                update.setString(1, date);
                update.setString(3, date);

                connection.commit();

                count = 0;
                lastModify = System.currentTimeMillis();
            } else if (!lastDay.equals(date)) {
                // 新的shop类日期与之前不同，则应当组装update并执行，并且执行insert语句
                update.setString(1, lastDay);
                update.setString(3, lastDay);
                update.setInt(2, count);
                update.executeUpdate();

                insert.setString(1, date);
                insert.setInt(2, 1);
                insert.executeUpdate();

                connection.commit();

                count = 0;
                lastDay = date;
                lastModify = System.currentTimeMillis();
            } else if (System.currentTimeMillis() - lastModify > intervals) {
                // shop的日期与之前相同，且intervals内未执行过update语句，此时执行update语句并重置count
                count++;
                update.setInt(2, count);
                update.executeUpdate();
                connection.commit();

                lastModify = System.currentTimeMillis();
                count = 0;
            } else {
                // shop的日期与之前相同，且intervals内执行过update语句，此时不执行update语句，仅增加count计数
                count++;
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

@RequestMapping("/select")
@RestController
public class selectController {

    Connection connection = null;

    PreparedStatement preparedStatement;

    @PostConstruct
    void init() {
        try {
            Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
            String url = "jdbc:clickhouse://124.222.139.8:8123/dm?compress=false";
            // String url = "jdbc:clickhouse://localhost:8123/dm?compress=false";
            connection = DriverManager.getConnection(url);
            preparedStatement = connection.prepareStatement("SELECT count(1), tran_date FROM dm.v_tr_shop_mx group by tran_date order by tran_date");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/all")
    @CrossOrigin
    public List<String[]> select() {
        try {
            List<String[]> res = new ArrayList<>();
            ResultSet set = preparedStatement.executeQuery();
            while (set.next()) {
                res.add(new String[]{set.getString(2), set.getString(1)});
            }
            return res;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

}