代码优化，抽取公共方法，优化clickhousesink写入

1 files changed, 35 insertions, 5 deletions

diff --git a/src/main/java/com/zdjizhi/enums/LogMetadata.java b/src/main/java/com/zdjizhi/enums/LogMetadata.java
index d3a7f49..2949085 100644
--- a/src/main/java/com/zdjizhi/enums/LogMetadata.java
+++ b/src/main/java/com/zdjizhi/enums/LogMetadata.java
@@ -1,6 +1,11 @@
 package com.zdjizhi.enums;
 
 import cn.hutool.core.util.EnumUtil;
+import cn.hutool.core.util.StrUtil;
+
+import java.util.Arrays;
+
+import static com.zdjizhi.common.FlowWriteConfig.CK_DATABASE;
 
 /**
  * @description: \
@@ -10,23 +15,27 @@ import cn.hutool.core.util.EnumUtil;
 public enum LogMetadata {
 
     /*
-     * 日志名称topic,表名
+     * 日志名称，表名，字段
      * */
 
-    CONNECTION_RECORD_LOG("connection_record_log", "connection_record_log"),
-    CONNECTION_SKETCH_RECORD_LOG("connection_sketch_record_log", "connection_sketch_record_log"),
-    DNS_RECORD_LOG("dns_record_log", "dns_record_log"),
+    CONNECTION_RECORD_LOG("connection_record_log", "connection_record_log_local", new String[]{"cap_ip", "recv_ip", "src_ip", "dst_ip", "src_port", "dst_port", "addr_type", "protocol", "fxo_id", "link_status", "dir_status", "total_cs_pkts", "total_sc_pkts", "total_cs_bytes", "total_sc_bytes", "log_gen_time", "aa", "wv", "yy", "user_mask", "conn_start_time", "app_class", "app_id", "http_host", "http_url", "http_cookie", "http_user_agent", "http_method", "http_accept", "http_accept_encoding", "http_referer", "http_rescode", "tls_sni", "tls_cert", "phone_num", "imei", "imsi"}),
+    CONNECTION_RELATION_LOG("connection_relation_log", "connection_relation_log_local", new String[]{"start_time", "end_time", "src_ip", "dst_ip", "sessions", "packets", "bytes"}),
+    CONNECTION_SKETCH_RECORD_LOG("connection_sketch_record_log", "connection_sketch_record_log_local", new String[]{"sled_ip", "sketch_start_time", "sketch_duration", "src_ip", "dst_ip", "sketch_sessions", "sketch_packets", "sketch_bytes"}),
+    DNS_RECORD_LOG("dns_record_log", "dns_record_log_local", new String[]{"capture_time", "recv_ip", "src_ip", "dst_ip", "src_port", "dst_port", "addr_type", "dns_flag", "ttl", "protocol", "fxo_id", "req_type", "qname", "response", "dns_a", "dns_a_num", "dns_cname", "dns_cname_num", "dns_aaaa", "dns_aaaa_num", "dns_mx", "dns_mx_num", "dns_ns", "dns_ns_num"}),
+    DNS_RELATION_LOG("dns_relation_log", "dns_relation_log_local", new String[]{"start_time", "end_time", "record_type", "qname", "record", "sessions"}),
     ;
 
     private String source;
     private String sink;
+    private String[] fields;
 
     LogMetadata() {
     }
 
-    LogMetadata(String source, String sink) {
+    LogMetadata(String source, String sink, String[] fields) {
         this.source = source;
         this.sink = sink;
+        this.fields = fields;
     }
 
     public String getSource() {
@@ -37,10 +46,31 @@ public enum LogMetadata {
         return sink;
     }
 
+    public String[] getFields() {
+        return fields;
+    }
+
     public static String getLogSink(String source) {
         LogMetadata logMetadata = EnumUtil.fromString(LogMetadata.class, source);
         return logMetadata.getSink();
+    }
 
+    public static String[] getLogFields(String tableName) {
+        LogMetadata[] values = LogMetadata.values();
+        for (LogMetadata value : values) {
+            if (value.sink.equals(tableName)) {
+                return value.fields;
+            }
+        }
+        return null;
     }
 
+    public static String preparedSql(String tableName) {
+        String[] fields = LogMetadata.getLogFields(tableName);
+        String[] placeholders = new String[fields.length];
+        Arrays.fill(placeholders, "?");
+
+        return StrUtil.concat(true, "INSERT INTO ", CK_DATABASE, ".", tableName,
+                "(", StrUtil.join(",", fields), ") VALUES (", StrUtil.join(",", placeholders), ")");
+    }
 }