Live Traffic Chart使用Fastjson2测试版本

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.zdjizhi</groupId>
     <artifactId>log-olap-analysis-schema</artifactId>
-    <version>230317-DataSketches</version>
+    <version>230414-FastJson2</version>
 
     <name>log-olap-analysis-schema</name>
     <url>http://www.example.com</url>
@@ -40,6 +40,7 @@
         <hbase.version>2.2.3</hbase.version>
         <nacos.version>1.2.0</nacos.version>
         <zdjz.tools.version>1.0.8</zdjz.tools.version>
+        <fastjson.version>2.0.26</fastjson.version>
         <scope.type>provided</scope.type>
         <!--<scope.type>compile</scope.type>-->
     </properties>
@@ -238,6 +239,12 @@
             <version>3.2.0</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>${fastjson.version}</version>
+        </dependency>
+
     </dependencies>
 </project>
 

properties/default_config.properties

@@ -3,7 +3,7 @@
 session.timeout.ms=60000
 
 #kafka source poll
-max.poll.records=3000
+max.poll.records=5000
 
 #kafka source poll bytes
 max.partition.fetch.bytes=31457280

properties/service_flow_config.properties

@@ -15,41 +15,42 @@ tools.library=D:\\workerspace\\dat
 nacos.server=192.168.44.12:8848
 
 #nacos namespace
-nacos.schema.namespace=prod
+nacos.schema.namespace=livecharts
 
 #nacos data id
-nacos.data.id=liveChart_session.json
+nacos.data.id=liveChart_session_test.json
 
 #--------------------------------Kafka消费组信息------------------------------#
 
 #kafka 接收数据topic
-source.kafka.topic=SESSION-RECORD
+source.kafka.topic=test
 
 #补全数据 输出 topic
-sink.kafka.topic=test-result
+sink.kafka.topic=TRAFFIC-PROTOCOL-TEST
 
 #读取topic,存储该spout id的消费offset信息，可通过该拓扑命名;具体存储offset的位置，确定下次读取不重复的数据；
-group.id=livecharts-test-20220816-1
+group.id=livecharts-test-20230327-3
 
 #--------------------------------topology配置------------------------------#
 
 #consumer 并行度
-source.parallelism=1
+source.parallelism=3
 
 #map函数并行度
-parse.parallelism=1
+parse.parallelism=3
 
 #第一次窗口计算并行度
-first.window.parallelism=1
+first.window.parallelism=3
 
 #第二次窗口计算并行度
-second.window.parallelism=1
+second.window.parallelism=3
 
 #producer 并行度
-sink.parallelism=1
+sink.parallelism=3
 
 #初次随机预聚合窗口时间
 first.count.window.time=5
 
 #二次聚合窗口时间
 second.count.window.time=15
+

src/main/java/com/zdjizhi/topology/StreamAggregateTopology.java

@@ -2,6 +2,7 @@ package com.zdjizhi.topology;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
+import com.alibaba.fastjson2.JSONObject;
 import com.zdjizhi.common.StreamAggregateConfig;
 import com.zdjizhi.utils.functions.keyby.FirstKeyByFunction;
 import com.zdjizhi.utils.functions.keyby.SecondKeyByFunction;
@@ -45,25 +46,25 @@ public class StreamAggregateTopology {
                     .setParallelism(StreamAggregateConfig.SOURCE_PARALLELISM).name(StreamAggregateConfig.SOURCE_KAFKA_TOPIC);
 
             //解析原始日志初步聚合计算，增加自定义key 缓解数据倾斜
-            SingleOutputStreamOperator<Tuple3<String, String, Map<String, Object>>> parseDataMap = streamSource.map(new ParseMapFunction())
+            SingleOutputStreamOperator<Tuple3<String, String, JSONObject>> parseDataMap = streamSource.map(new ParseMapFunction())
                     .name("ParseDataMap")
                     .setParallelism(StreamAggregateConfig.PARSE_PARALLELISM);
 
             //初步聚合计算，增加自定义key 缓解数据倾斜
-            WindowedStream<Tuple3<String, String, Map<String, Object>>, String, TimeWindow> firstWindow = parseDataMap.keyBy(new FirstKeyByFunction())
+            WindowedStream<Tuple3<String, String, JSONObject>, String, TimeWindow> firstWindow = parseDataMap.keyBy(new FirstKeyByFunction())
                     .window(TumblingProcessingTimeWindows.of(Time.seconds(StreamAggregateConfig.FIRST_COUNT_WINDOW_TIME)));
 
             //初次聚合计算窗口
-            SingleOutputStreamOperator<Tuple2<String, Map<String, Object>>> metricCountWindow = firstWindow.process(new FirstCountWindowFunction())
+            SingleOutputStreamOperator<Tuple2<String, JSONObject>> metricCountWindow = firstWindow.process(new FirstCountWindowFunction())
                     .name("FirstCountWindow")
                     .setParallelism(StreamAggregateConfig.FIRST_WINDOW_PARALLELISM);
 
             //二次聚合计算，使用业务的key 进行数据汇总
-            WindowedStream<Tuple2<String, Map<String, Object>>, String, TimeWindow> secondWindow = metricCountWindow.keyBy(new SecondKeyByFunction())
+            WindowedStream<Tuple2<String, JSONObject>, String, TimeWindow> secondWindow = metricCountWindow.keyBy(new SecondKeyByFunction())
                     .window(TumblingProcessingTimeWindows.of(Time.seconds(StreamAggregateConfig.SECOND_COUNT_WINDOW_TIME)));
 
             //二次聚合计算窗口
-            SingleOutputStreamOperator<Map<String, Object>> secondCountWindow = secondWindow.process(new SecondCountWindowFunction())
+            SingleOutputStreamOperator<JSONObject> secondCountWindow = secondWindow.process(new SecondCountWindowFunction())
                     .name("SecondCountWindow").setParallelism(StreamAggregateConfig.SECOND_WINDOW_PARALLELISM);
 
             //拆解结果数据按protocol id循环输出

src/main/java/com/zdjizhi/utils/functions/filter/FilterNullFunction.java

@@ -1,17 +0,0 @@
-package com.zdjizhi.utils.functions.filter;
-
-import com.zdjizhi.utils.StringUtil;
-import org.apache.flink.api.common.functions.FilterFunction;
-
-/**
- * @author qidaijie
- * @Package com.zdjizhi.utils.functions
- * @Description:
- * @date 2021/5/2715:01
- */
-public class FilterNullFunction implements FilterFunction<String> {
-    @Override
-    public boolean filter(String message) {
-        return StringUtil.isNotBlank(message);
-    }
-}

src/main/java/com/zdjizhi/utils/functions/keyby/FirstKeyByFunction.java

@@ -1,10 +1,8 @@
 package com.zdjizhi.utils.functions.keyby;
 
-import cn.hutool.core.util.RandomUtil;
-import com.zdjizhi.common.StreamAggregateConfig;
+import com.alibaba.fastjson2.JSONObject;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.tuple.Tuple4;
 
 import java.util.Map;
 
@@ -14,10 +12,10 @@ import java.util.Map;
  * @Description:
  * @date 2021/7/2112:13
  */
-public class FirstKeyByFunction implements KeySelector<Tuple3< String, String, Map<String, Object>>, String> {
+public class FirstKeyByFunction implements KeySelector<Tuple3< String, String, JSONObject>, String> {
 
     @Override
-    public String getKey(Tuple3<String, String, Map<String, Object>> value) throws Exception {
+    public String getKey(Tuple3<String, String, JSONObject> value) throws Exception {
         //以map拼接的key分组
         return value.f0;
     }

src/main/java/com/zdjizhi/utils/functions/keyby/SecondKeyByFunction.java

@@ -1,12 +1,10 @@
 package com.zdjizhi.utils.functions.keyby;
 
-import cn.hutool.core.util.RandomUtil;
-import com.zdjizhi.common.StreamAggregateConfig;
+
+import com.alibaba.fastjson2.JSONObject;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.tuple.Tuple4;
 
-import java.util.Map;
 
 /**
  * @author qidaijie
@@ -14,10 +12,10 @@ import java.util.Map;
  * @Description:
  * @date 2021/7/2112:13
  */
-public class SecondKeyByFunction implements KeySelector<Tuple2<String,Map<String, Object>>, String> {
+public class SecondKeyByFunction implements KeySelector<Tuple2<String, JSONObject>, String> {
 
     @Override
-    public String getKey(Tuple2<String, Map<String, Object>> value) throws Exception {
+    public String getKey(Tuple2<String, JSONObject> value) throws Exception {
         //以map拼接的key分组
         return value.f0;
     }

src/main/java/com/zdjizhi/utils/functions/parse/ParseMapFunction.java

@@ -2,16 +2,15 @@ package com.zdjizhi.utils.functions.parse;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
 import com.zdjizhi.common.StreamAggregateConfig;
-import com.zdjizhi.utils.JsonMapper;
 import com.zdjizhi.utils.StringUtil;
 import com.zdjizhi.utils.general.ParseFunctions;
-import com.zdjizhi.utils.json.JsonParseUtil;
+import com.zdjizhi.utils.meta.MetaDataParse;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.java.tuple.Tuple3;
 
-import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ThreadLocalRandom;
 
@@ -22,21 +21,19 @@ import java.util.concurrent.ThreadLocalRandom;
  * @Description:
  * @date 2021/5/2715:01
  */
-public class ParseMapFunction implements MapFunction<String, Tuple3<String, String, Map<String, Object>>> {
+public class ParseMapFunction implements MapFunction<String, Tuple3<String, String, JSONObject>> {
     private static final Log logger = LogFactory.get();
 
     @Override
     @SuppressWarnings("unchecked")
-    public Tuple3<String, String, Map<String, Object>> map(String message) {
+    public Tuple3<String, String, JSONObject> map(String message) {
         try {
-            ArrayList<String[]> jobList = JsonParseUtil.getTransformsList();
-            HashMap<String, String> dimensionsMap = JsonParseUtil.getDimensionsMap();
             if (StringUtil.isNotBlank(message)) {
-                Map<String, Object> originalLog = (Map<String, Object>) JsonMapper.fromJsonString(message, Map.class);
-                Map<String, Object> dimensionsObj = ParseFunctions.transDimensions(dimensionsMap, originalLog);
+                JSONObject originalLog = JSON.parseObject(message);
+                Map<String, Object> dimensionsObj = ParseFunctions.transDimensions(MetaDataParse.getDimensionsMap(), originalLog);
                 if (ParseFunctions.filterLogs(originalLog)) {
-                    Map<String, Object> metricsLog = ParseFunctions.getMetricsLog(originalLog);
-                    for (String[] strings : jobList) {
+                    JSONObject metricsLog = ParseFunctions.getMetricsLog(originalLog);
+                    for (String[] strings : MetaDataParse.getTransformsList()) {
                         //函数名称
                         String function = strings[0];
                         //结果集字段key
@@ -46,7 +43,7 @@ public class ParseMapFunction implements MapFunction<String, Tuple3<String, Stri
                         //额外的参数的值
                         String parameters = strings[3];
                         //原始日志字段对应的值
-                        Object logsValue = JsonParseUtil.getValue(originalLog, logsKeyName);
+                        Object logsValue = originalLog.get(logsKeyName);
 
                         switch (function) {
                             case "combination":
@@ -64,8 +61,8 @@ public class ParseMapFunction implements MapFunction<String, Tuple3<String, Stri
                                 }
                                 break;
                             case "hierarchy":
-                                String key = JsonParseUtil.getString(dimensionsObj, resultKeyName) + "@" + ThreadLocalRandom.current().nextInt(StreamAggregateConfig.RANDOM_RANGE_NUM);
-                                return new Tuple3<>(key, JsonMapper.toJsonString(dimensionsObj), metricsLog);
+                                String key = dimensionsObj.get(resultKeyName) + "@" + ThreadLocalRandom.current().nextInt(StreamAggregateConfig.RANDOM_RANGE_NUM);
+                                return new Tuple3<>(key, JSONObject.toJSONString(dimensionsObj), metricsLog);
                             default:
                                 break;
                         }

src/main/java/com/zdjizhi/utils/functions/result/ResultFlatMapFunction.java

@@ -1,21 +1,23 @@
 package com.zdjizhi.utils.functions.result;
 
+import cn.hutool.log.Log;
+import cn.hutool.log.LogFactory;
+import com.alibaba.fastjson2.JSONObject;
 import com.zdjizhi.common.StreamAggregateConfig;
-import com.zdjizhi.utils.JsonMapper;
 import com.zdjizhi.utils.StringUtil;
-import com.zdjizhi.utils.json.JsonParseUtil;
+import com.zdjizhi.utils.general.ParseFunctions;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.util.Collector;
 
-import java.util.Map;
-
 /**
  * @author qidaijie
  * @Package com.zdjizhi.utils.functions
  * @Description:
  * @date 2021/7/2114:52
  */
-public class ResultFlatMapFunction implements FlatMapFunction<Map<String, Object>, String> {
+public class ResultFlatMapFunction implements FlatMapFunction<JSONObject, String> {
+    private static final Log logger = LogFactory.get();
+
     private static final String PROTOCOL_ID_KEY = "protocol_stack_id";
     private static final String APP_NAME_KEY = "app_name";
     private static final String HLL_SKETCH_KEY = "client_ip_sketch";
@@ -23,28 +25,26 @@ public class ResultFlatMapFunction implements FlatMapFunction<Map<String, Object
 
     @Override
     @SuppressWarnings("unchecked")
-    public void flatMap(Map<String, Object> jsonObject, Collector<String> out) throws Exception {
-        String protocol = JsonParseUtil.getString(jsonObject, PROTOCOL_ID_KEY);
-        if (jsonObject.containsKey(HLL_SKETCH_KEY)){
-            JsonParseUtil.setValue(jsonObject, HLL_SKETCH_KEY, JsonParseUtil.getHllSketch(jsonObject, HLL_SKETCH_KEY));
+    public void flatMap(JSONObject jsonObject, Collector<String> out) throws Exception {
+        String protocol = jsonObject.getString(PROTOCOL_ID_KEY);
+        if (jsonObject.containsKey(HLL_SKETCH_KEY)) {
+            jsonObject.put(HLL_SKETCH_KEY, ParseFunctions.getHllSketch(jsonObject, HLL_SKETCH_KEY));
         }
-        out.collect(JsonMapper.toJsonString(jsonObject));
+        out.collect(jsonObject.toString());
         jsonObject.remove(APP_NAME_KEY);
 
         StringBuilder stringBuilder = new StringBuilder();
-        if (StringUtil.isNotBlank(protocol)) {
-            String[] protocolIds = protocol.split(StreamAggregateConfig.PROTOCOL_SPLITTER);
-            int protocolIdsNum = protocolIds.length;
-            for (int i = 0; i < protocolIdsNum - 1; i++) {
-                if (StringUtil.isBlank(stringBuilder.toString())) {
-                    stringBuilder.append(protocolIds[i]);
-                    jsonObject.put(PROTOCOL_ID_KEY, stringBuilder.toString());
-                    out.collect(JsonMapper.toJsonString(jsonObject));
-                } else {
-                    stringBuilder.append(".").append(protocolIds[i]);
-                    jsonObject.put(PROTOCOL_ID_KEY, stringBuilder.toString());
-                    out.collect(JsonMapper.toJsonString(jsonObject));
-                }
+        String[] protocolIds = protocol.split(StreamAggregateConfig.PROTOCOL_SPLITTER);
+        int protocolIdsNum = protocolIds.length;
+        for (int i = 0; i < protocolIdsNum - 1; i++) {
+            if (StringUtil.isBlank(stringBuilder.toString())) {
+                stringBuilder.append(protocolIds[i]);
+                jsonObject.put(PROTOCOL_ID_KEY, stringBuilder.toString());
+                out.collect(jsonObject.toString());
+            } else {
+                stringBuilder.append(".").append(protocolIds[i]);
+                jsonObject.put(PROTOCOL_ID_KEY, stringBuilder.toString());
+                out.collect(jsonObject.toString());
             }
         }
     }

src/main/java/com/zdjizhi/utils/functions/statistics/FirstCountWindowFunction.java

@@ -1,21 +1,19 @@
 package com.zdjizhi.utils.functions.statistics;
 
-import com.zdjizhi.utils.JsonMapper;
-import com.zdjizhi.utils.StringUtil;
+import cn.hutool.log.Log;
+import cn.hutool.log.LogFactory;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
 import com.zdjizhi.utils.general.MetricFunctions;
-import com.zdjizhi.utils.general.ParseFunctions;
-import com.zdjizhi.utils.json.JsonParseUtil;
+import com.zdjizhi.utils.meta.MetaDataParse;
 import org.apache.datasketches.hll.HllSketch;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
-import java.util.Map;
 
 /**
  * @author qidaijie
@@ -23,32 +21,31 @@ import java.util.Map;
  * @Description:
  * @date 2021/7/2113:55
  */
-public class FirstCountWindowFunction extends ProcessWindowFunction<Tuple3<String, String, Map<String, Object>>, Tuple2<String, Map<String, Object>>, String, TimeWindow> {
-    private static final Logger logger = LoggerFactory.getLogger(FirstCountWindowFunction.class);
+public class FirstCountWindowFunction extends ProcessWindowFunction<Tuple3<String, String, JSONObject>, Tuple2<String, JSONObject>, String, TimeWindow> {
+    private static final Log logger = LogFactory.get();
 
-    private HashMap<String, Map<String, Object>> cacheMap = new HashMap<>(32);
+    private HashMap<String, JSONObject> cacheMap = new HashMap<>(32);
 
     @Override
     @SuppressWarnings("unchecked")
-    public void process(String key, Context context, Iterable<Tuple3<String, String, Map<String, Object>>> input, Collector<Tuple2<String, Map<String, Object>>> output) {
+    public void process(String key, Context context, Iterable<Tuple3<String, String, JSONObject>> input, Collector<Tuple2<String, JSONObject>> output) {
         try {
-            HashMap<String, String[]> metricsMap = JsonParseUtil.getMetricFunctionsMap();
-            for (Tuple3<String, String, Map<String, Object>> tuple : input) {
+            HashMap<String, String[]> metricsMap = MetaDataParse.getMetricFunctionsMap();
+            for (Tuple3<String, String, JSONObject> tuple : input) {
                 String dimensions = tuple.f1;
-                Map<String, Object> metrics = tuple.f2;
-                if (metrics.size() != 0) {
-                    Map<String, Object> cacheMessage = cacheMap.getOrDefault(dimensions, (Map<String, Object>) JsonMapper.fromJsonString(dimensions, Map.class));
+                JSONObject metrics = tuple.f2;
+                JSONObject cacheMessage = cacheMap.getOrDefault(dimensions, JSON.parseObject(dimensions));
 
-                    for (String resultKeyName : metricsMap.keySet()) {
-                        String[] functions = metricsMap.get(resultKeyName);
-                        String function = functions[0];
-                        String fieldName = functions[1];
-                        functionSet(function, cacheMessage, resultKeyName, JsonParseUtil.getValue(metrics, fieldName));
-                    }
-                    cacheMap.put(dimensions, cacheMessage);
+                for (String resultKeyName : metricsMap.keySet()) {
+                    String[] functions = metricsMap.get(resultKeyName);
+                    String function = functions[0];
+                    String fieldName = functions[1];
+                    functionSet(function, cacheMessage, resultKeyName, metrics.get(fieldName));
                 }
+                cacheMap.put(dimensions, cacheMessage);
             }
 
+
             if (!cacheMap.isEmpty()) {
                 for (String dimensions : cacheMap.keySet()) {
                     output.collect(new Tuple2<>(dimensions, cacheMap.get(dimensions)));
@@ -72,7 +69,7 @@ public class FirstCountWindowFunction extends ProcessWindowFunction<Tuple3<Strin
      * @param resultKeyName  结果字段名称
      * @param fieldNameValue 新加值
      */
-    private static void functionSet(String function, Map<String, Object> cacheMessage, String resultKeyName, Object fieldNameValue) {
+    private static void functionSet(String function, JSONObject cacheMessage, String resultKeyName, Object fieldNameValue) {
         switch (function) {
             case "sum":
                 cacheMessage.put(resultKeyName, MetricFunctions.longSum(cacheMessage.get(resultKeyName), fieldNameValue));

src/main/java/com/zdjizhi/utils/functions/statistics/SecondCountWindowFunction.java

@@ -1,16 +1,17 @@
 package com.zdjizhi.utils.functions.statistics;
 
-import com.zdjizhi.utils.JsonMapper;
-import com.zdjizhi.utils.StringUtil;
+import cn.hutool.log.Log;
+import cn.hutool.log.LogFactory;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
 import com.zdjizhi.utils.general.MetricFunctions;
-import com.zdjizhi.utils.json.JsonParseUtil;
+import com.zdjizhi.utils.general.ParseFunctions;
+import com.zdjizhi.utils.meta.MetaDataParse;
 import org.apache.datasketches.hll.HllSketch;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -21,27 +22,26 @@ import java.util.Map;
  * @Description:
  * @date 2021/7/2113:55
  */
-public class SecondCountWindowFunction extends ProcessWindowFunction<Tuple2<String, Map<String, Object>>, Map<String, Object>, String, TimeWindow> {
-    private static final Logger logger = LoggerFactory.getLogger(SecondCountWindowFunction.class);
+public class SecondCountWindowFunction extends ProcessWindowFunction<Tuple2<String, JSONObject>, JSONObject, String, TimeWindow> {
+    private static final Log logger = LogFactory.get();
 
-    private HashMap<String, Map<String, Object>> cacheMap = new HashMap<>(32);
+    private HashMap<String, JSONObject> cacheMap = new HashMap<>(32);
 
     @Override
     @SuppressWarnings("unchecked")
-    public void process(String key, Context context, Iterable<Tuple2<String, Map<String, Object>>> input, Collector<Map<String, Object>> output) {
+    public void process(String key, Context context, Iterable<Tuple2<String, JSONObject>> input, Collector<JSONObject> output) {
         try {
-            HashMap<String, String[]> metricsMap = JsonParseUtil.getMetricFunctionsMap();
-            for (Tuple2<String, Map<String, Object>> tuple : input) {
+            HashMap<String, String[]> metricsMap = MetaDataParse.getMetricFunctionsMap();
+            for (Tuple2<String, JSONObject> tuple : input) {
                 String dimensions = tuple.f0;
                 Map<String, Object> message = tuple.f1;
                 if (message.size() != 0) {
-                    Map<String, Object> dimensionsObj = (Map<String, Object>) JsonMapper.fromJsonString(dimensions, Map.class);
+                    JSONObject cacheMessage = cacheMap.getOrDefault(dimensions, JSON.parseObject(dimensions));
 
-                    Map<String, Object> cacheMessage = cacheMap.getOrDefault(dimensions, dimensionsObj);
                     for (String resultName : metricsMap.keySet()) {
                         String[] metrics = metricsMap.get(resultName);
                         String function = metrics[0];
-                        functionSet(function, cacheMessage, resultName, JsonParseUtil.getValue(message, resultName));
+                        functionSet(function, cacheMessage, resultName, message.get(resultName));
 
                     }
                     cacheMap.put(dimensions, cacheMessage);
@@ -52,8 +52,8 @@ public class SecondCountWindowFunction extends ProcessWindowFunction<Tuple2<Stri
                 Long endTime = context.window().getEnd() / 1000;
 
                 for (String countKey : cacheMap.keySet()) {
-                    Map<String, Object> resultMap = cacheMap.get(countKey);
-                    JsonParseUtil.setValue(resultMap, JsonParseUtil.getResultTimeKey(), endTime);
+                    JSONObject resultMap = cacheMap.get(countKey);
+                    resultMap.put(MetaDataParse.getResultTimeKey(), endTime);
                     output.collect(resultMap);
                 }
             }

src/main/java/com/zdjizhi/utils/general/MetricFunctions.java

@@ -1,8 +1,9 @@
 package com.zdjizhi.utils.general;
 
 
+import cn.hutool.log.Log;
+import cn.hutool.log.LogFactory;
 import com.zdjizhi.utils.StringUtil;
-import com.zdjizhi.utils.json.JsonTypeUtil;
 import org.apache.datasketches.hll.HllSketch;
 import org.apache.datasketches.hll.Union;
 
@@ -14,6 +15,8 @@ import org.apache.datasketches.hll.Union;
  * @date 2021/7/2015:31
  */
 public class MetricFunctions {
+    private static final Log logger = LogFactory.get();
+
 
     /**
      * Long类型的数据求和
@@ -23,8 +26,8 @@ public class MetricFunctions {
      * @return value1 + value2
      */
     public static Long longSum(Object value1, Object value2) {
-        Long res1 = JsonTypeUtil.checkLongValue(value1);
-        Long res2 = JsonTypeUtil.checkLongValue(value2);
+        Long res1 = checkLongValue(value1);
+        Long res2 = checkLongValue(value2);
 
         return res1 + res2;
     }
@@ -36,7 +39,7 @@ public class MetricFunctions {
      * @return count+1
      */
     public static Long count(Object count) {
-        return JsonTypeUtil.checkLongValue(count) + 1L;
+        return checkLongValue(count) + 1L;
     }
 
     /**
@@ -74,4 +77,30 @@ public class MetricFunctions {
         }
         return HllSketch.heapify(union.getResult().toUpdatableByteArray());
     }
+
+
+    private static long checkLongValue(Object value) {
+        if (value == null) {
+            return 0L;
+        }
+
+        if (value instanceof Long) {
+            return ((Long) value);
+        }
+
+        if (value instanceof Number) {
+            return ((Number) value).longValue();
+        }
+
+        if (value instanceof String) {
+            String str = (String) value;
+            try {
+                return Long.parseLong(str);
+            } catch (NumberFormatException e) {
+                logger.error("Can not cast " + value.getClass() + "to Long,exception is:" + e.getMessage());
+            }
+        }
+
+        return 0L;
+    }
 }

src/main/java/com/zdjizhi/utils/general/ParseFunctions.java

@@ -3,13 +3,15 @@ package com.zdjizhi.utils.general;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
+import com.alibaba.fastjson2.JSONObject;
 import com.jayway.jsonpath.JsonPath;
 import com.zdjizhi.common.StreamAggregateConfig;
-import com.zdjizhi.utils.JsonMapper;
 import com.zdjizhi.utils.StringUtil;
-import com.zdjizhi.utils.json.JsonParseUtil;
+import com.zdjizhi.utils.meta.MetaDataParse;
+import org.apache.datasketches.hll.HllSketch;
 
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -29,18 +31,15 @@ public class ParseFunctions {
      * @param object 原始日志
      * @return true or false
      */
-    public static boolean filterLogs(Map<String, Object> object) {
+    public static boolean filterLogs(JSONObject object) {
         boolean available = false;
-        HashMap<String, String> filtersMap = JsonParseUtil.getFiltersMap();
+        HashMap<String, String> filtersMap = MetaDataParse.getFiltersMap();
         for (String key : filtersMap.keySet()) {
-            switch (key) {
-                case "notempty":
-                    String value = JsonParseUtil.getString(object, filtersMap.get(key));
-                    if (StringUtil.isNotBlank(value)) {
-                        available = true;
-                    }
-                    break;
-                default:
+            if ("notempty".equals(key)) {
+                String value = object.getString(filtersMap.get(key));
+                if (StringUtil.isNotBlank(value)) {
+                    available = true;
+                }
             }
         }
         return available;
@@ -53,11 +52,12 @@ public class ParseFunctions {
      * @param originalLog 原始日志
      * @return 结果维度集
      */
-    public static Map<String, Object> transDimensions(Map<String, String> dimensions, Map<String, Object> originalLog) {
+    public static Map<String, Object> transDimensions(Map<String, String> dimensions, JSONObject originalLog) {
         HashMap<String, Object> dimensionsObj = new HashMap<>(16);
 
-        for (String dimension : dimensions.keySet()) {
-            dimensionsObj.put(dimension, JsonParseUtil.getValue(originalLog, dimensions.get(dimension)));
+        for (String key : dimensions.keySet()) {
+            originalLog.get(dimensions.get(key));
+            dimensionsObj.put(key, originalLog.get(dimensions.get(key)));
         }
 
         return dimensionsObj;
@@ -67,18 +67,17 @@ public class ParseFunctions {
      * 根据原始日志字段，生成schema内指定的metrics指标json。
      *
      * @param originalLog 原始日志json
-     * @return 统计metrics json
+     * @return 统计metrics meta
      */
-    public static Map<String, Object> getMetricsLog(Map<String, Object> originalLog) {
-        Map<String, Object> metricsMap = new HashMap<>(16);
+    public static JSONObject getMetricsLog(JSONObject originalLog) {
+        JSONObject metricsJson = new JSONObject();
 
-        for (String logsKeyName : JsonParseUtil.getMetricsFiledNameList()) {
+        for (String logsKeyName : MetaDataParse.getMetricsFiledNameList()) {
             if (originalLog.containsKey(logsKeyName)) {
-                metricsMap.put(logsKeyName, originalLog.get(logsKeyName));
+                metricsJson.put(logsKeyName, originalLog.get(logsKeyName));
             }
         }
-
-        return metricsMap;
+        return metricsJson;
     }
 
 
@@ -87,26 +86,25 @@ public class ParseFunctions {
      * 获取方法函数中 parameters 字段，结构 "parameters": "abc,/" ;abc为要拼接字段 /为拼接的分隔符
      *
      * @param parameters  参数集
-     * @param message     原始日志
+     * @param originalLog 原始日志
      * @param logsKeyName 原始日志列名
      */
-    public static void combinationUtils(Map<String, Object> dimensions, Map<String, Object> message, String parameters, String resultKeyName, String logsKeyName) {
+    public static void combinationUtils(Map<String, Object> dimensions, JSONObject originalLog, String parameters, String resultKeyName, String logsKeyName) {
         String[] combinationPars = parameters.split(StreamAggregateConfig.FORMAT_SPLITTER);
         String combinationFieldKey = combinationPars[0];
         String separator = combinationPars[1];
-        Object combinationFieldValue = JsonParseUtil.getValue(message, combinationFieldKey);
+        Object combinationFieldValue = originalLog.get(combinationFieldKey);
         if (combinationFieldValue != null) {
-            Object logsFieldValue = JsonParseUtil.getValue(message, logsKeyName);
+            Object logsFieldValue = originalLog.get(logsKeyName);
             String combinationResult = logsFieldValue + separator + combinationFieldValue;
-            JsonParseUtil.setValue(dimensions, resultKeyName, combinationResult);
-            JsonParseUtil.setValue(message, logsKeyName, combinationResult);
+            dimensions.put(resultKeyName, combinationResult);
         }
     }
 
     /**
      * 根据表达式解析json
      * <p>
-     * //* @param message json
+     * //* @param message meta
      *
      * @param expr 解析表达式
      * @return 解析结果
@@ -135,7 +133,7 @@ public class ParseFunctions {
     @Deprecated
     private static Object isJsonValue(Map<String, Object> jsonMap, String param) {
         if (param.contains(StreamAggregateConfig.IS_JSON_KEY_TAG)) {
-            return JsonParseUtil.getValue(jsonMap, param.substring(2));
+            return jsonMap.get(param.substring(2));
         } else {
             return param;
         }
@@ -182,4 +180,23 @@ public class ParseFunctions {
         return result;
     }
 
+    /**
+     * 获取HLLSketch内容
+     *
+     * @param jsonMap 原始日志
+     * @param key     meta key名称
+     * @return HLLSketch数据数组
+     */
+    public static String getHllSketch(JSONObject jsonMap, String key) {
+        try {
+            HllSketch hllSketchResult = (HllSketch) jsonMap.get(key);
+            if (hllSketchResult != null) {
+                return Base64.getEncoder().encodeToString(hllSketchResult.toUpdatableByteArray());
+            }
+        } catch (RuntimeException e) {
+            logger.error("HllSketch data conversion exception,data may be empty! exception:{}", e);
+        }
+        return null;
+    }
+
 }

src/main/java/com/zdjizhi/utils/json/JsonTypeUtil.java

@@ -1,144 +0,0 @@
-package com.zdjizhi.utils.json;
-
-
-import com.zdjizhi.utils.JsonMapper;
-import com.zdjizhi.utils.exception.AnalysisException;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * @author qidaijie
- * @Package PACKAGE_NAME
- * @Description:
- * @date 2021/7/1217:34
- */
-public class JsonTypeUtil {
-    /**
-     * String 类型检验转换方法
-     *
-     * @param value json value
-     * @return String value
-     */
-    public static String checkString(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-        if (value instanceof Map) {
-            return JsonMapper.toJsonString(value);
-        }
-
-        if (value instanceof List) {
-            return JsonMapper.toJsonString(value);
-        }
-
-        return value.toString();
-    }
-
-    /**
-     * array 类型检验转换方法
-     *
-     * @param value json value
-     * @return List value
-     */
-    private static Map checkObject(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-        if (value instanceof Map) {
-            return (Map) value;
-        }
-
-        throw new AnalysisException("can not cast to map, value : " + value);
-    }
-
-    /**
-     * array 类型检验转换方法
-     *
-     * @param value json value
-     * @return List value
-     */
-    private static List checkArray(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-        if (value instanceof List) {
-            return (List) value;
-        }
-
-        throw new AnalysisException("can not cast to List, value : " + value);
-    }
-
-    private static Long checkLong(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-        return TypeUtils.castToLong(value);
-    }
-
-    /**
-     * long 类型检验转换方法,若为空返回基础值
-     *
-     * @param value json value
-     * @return Long value
-     */
-    public static long checkLongValue(Object value) {
-
-        Long longVal = TypeUtils.castToLong(value);
-
-        if (longVal == null) {
-            return 0L;
-        }
-
-        if (longVal < 0L) {
-            return 0L;
-        }
-
-        return longVal;
-    }
-
-    /**
-     * Double 类型校验转换方法
-     *
-     * @param value json value
-     * @return Double value
-     */
-    private static Double checkDouble(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-        return TypeUtils.castToDouble(value);
-    }
-
-
-    private static Integer checkInt(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-        return TypeUtils.castToInt(value);
-    }
-
-
-    /**
-     * int 类型检验转换方法,若为空返回基础值
-     *
-     * @param value json value
-     * @return int value
-     */
-    private static int getIntValue(Object value) {
-
-        Integer intVal = TypeUtils.castToInt(value);
-
-        if (intVal == null) {
-            return 0;
-        }
-        return intVal;
-    }
-
-}

src/main/java/com/zdjizhi/utils/json/TypeUtils.java

@@ -1,172 +0,0 @@
-package com.zdjizhi.utils.json;
-
-import cn.hutool.log.Log;
-import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.StreamAggregateConfig;
-import com.zdjizhi.utils.StringUtil;
-import com.zdjizhi.utils.exception.AnalysisException;
-
-
-/**
- * @author qidaijie
- * @Package PACKAGE_NAME
- * @Description:
- * @date 2021/7/1218:20
- */
-public class TypeUtils {
-    private static final Log logger = LogFactory.get();
-
-    /**
-     * Integer 类型判断方法
-     *
-     * @param value json value
-     * @return Integer value or null
-     */
-    public static Object castToIfFunction(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-        if (value instanceof String) {
-            return value.toString();
-        }
-
-        if (value instanceof Integer) {
-            return ((Number) value).intValue();
-        }
-
-        if (value instanceof Long) {
-            return ((Number) value).longValue();
-        }
-
-        if (value instanceof Boolean) {
-            return (Boolean) value ? 1 : 0;
-        }
-
-        throw new AnalysisException("can not cast to int, value : " + value);
-    }
-
-    /**
-     * Integer 类型判断方法
-     *
-     * @param value json value
-     * @return Integer value or null
-     */
-    static Integer castToInt(Object value) {
-
-        if (value == null) {
-            return null;
-        }
-
-        if (value instanceof Integer) {
-            return (Integer) value;
-        }
-
-        //此判断数值超范围不抛出异常，会截取成对应类型数值
-//        if (value instanceof Number) {
-//            return ((Number) value).intValue();
-//        }
-
-        if (value instanceof String) {
-            String strVal = (String) value;
-            if (StringUtil.isBlank(strVal)) {
-                return null;
-            }
-
-            //将 10,20 类数据转换为10
-            if (strVal.contains(StreamAggregateConfig.FORMAT_SPLITTER)) {
-                strVal = strVal.split(StreamAggregateConfig.FORMAT_SPLITTER)[0];
-            }
-
-            try {
-                return Integer.parseInt(strVal);
-            } catch (NumberFormatException ex) {
-                logger.error("String change Integer Error,The error Str is:" + strVal);
-            }
-        }
-
-        if (value instanceof Boolean) {
-            return (Boolean) value ? 1 : 0;
-        }
-
-        throw new AnalysisException("can not cast to int, value : " + value);
-    }
-
-    /**
-     * Double类型判断方法
-     *
-     * @param value json value
-     * @return double value or null
-     */
-    static Double castToDouble(Object value) {
-
-        if (value instanceof Double) {
-            return (Double) value;
-        }
-
-        //此判断数值超范围不抛出异常，会截取成对应类型数值
-//        if (value instanceof Number) {
-//            return ((Number) value).doubleValue();
-//        }
-
-        if (value instanceof String) {
-            String strVal = (String) value;
-
-            if (StringUtil.isBlank(strVal)) {
-                return null;
-            }
-
-            //将 10,20 类数据转换为10
-            if (strVal.contains(StreamAggregateConfig.FORMAT_SPLITTER)) {
-                strVal = strVal.split(StreamAggregateConfig.FORMAT_SPLITTER)[0];
-            }
-
-            try {
-                return Double.parseDouble(strVal);
-            } catch (NumberFormatException ex) {
-                logger.error("String change Double Error,The error Str is:" + strVal);
-            }
-        }
-
-        throw new AnalysisException("can not cast to double, value : " + value);
-    }
-
-    /**
-     * Long类型判断方法
-     *
-     * @param value json value
-     * @return (Long)value or null
-     */
-    static Long castToLong(Object value) {
-        if (value == null) {
-            return null;
-        }
-
-//        此判断数值超范围不抛出异常，会截取成对应类型数值
-        if (value instanceof Number) {
-            return ((Number) value).longValue();
-        }
-
-        if (value instanceof String) {
-            String strVal = (String) value;
-
-            if (StringUtil.isBlank(strVal)) {
-                return null;
-            }
-
-            //将 10,20 类数据转换为10
-            if (strVal.contains(StreamAggregateConfig.FORMAT_SPLITTER)) {
-                strVal = strVal.split(StreamAggregateConfig.FORMAT_SPLITTER)[0];
-            }
-
-            try {
-                return Long.parseLong(strVal);
-            } catch (NumberFormatException ex) {
-                logger.error("String change Long Error,The error Str is:" + strVal);
-            }
-        }
-
-        throw new AnalysisException("can not cast to long, value : " + value);
-    }
-
-}

src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java

@@ -22,6 +22,7 @@ public class KafkaConsumer {
         properties.put("session.timeout.ms", StreamAggregateConfig.SESSION_TIMEOUT_MS);
         properties.put("max.poll.records", StreamAggregateConfig.MAX_POLL_RECORDS);
         properties.put("max.partition.fetch.bytes", StreamAggregateConfig.MAX_PARTITION_FETCH_BYTES);
+        properties.put("partition.discovery.interval.ms", "10000");
 
         CertUtils.chooseCert(StreamAggregateConfig.SOURCE_KAFKA_SERVERS, properties);
 

src/main/java/com/zdjizhi/utils/meta/MetaDataParse.java

@@ -1,4 +1,4 @@
-package com.zdjizhi.utils.json;
+package com.zdjizhi.utils.meta;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
@@ -10,21 +10,18 @@ import com.alibaba.nacos.api.exception.NacosException;
 import com.jayway.jsonpath.DocumentContext;
 import com.jayway.jsonpath.JsonPath;
 import com.zdjizhi.common.StreamAggregateConfig;
-import com.zdjizhi.utils.JsonMapper;
 import com.zdjizhi.utils.StringUtil;
 
 import java.util.*;
 import java.util.concurrent.Executor;
 
-import org.apache.datasketches.hll.HllSketch;
-
 
 /**
  * 使用FastJson解析json的工具类
  *
  * @author qidaijie
  */
-public class JsonParseUtil {
+public class MetaDataParse {
 
     private static final Log logger = LogFactory.get();
     private static Properties propNacos = new Properties();
@@ -90,89 +87,6 @@ public class JsonParseUtil {
         }
     }
 
-    /**
-     * 获取属性值的方法
-     *
-     * @param jsonMap 原始日志
-     * @param key     josn key名称
-     * @return 属性的值
-     */
-    public static Object getValue(Map<String, Object> jsonMap, String key) {
-        try {
-            return jsonMap.getOrDefault(key, null);
-        } catch (RuntimeException e) {
-            logger.error("Get the JSON value is abnormal,The key is :" + key + "error message is :" + e);
-            return null;
-        }
-    }
-
-    /**
-     * 获取HLLSketch内容
-     *
-     * @param jsonMap 原始日志
-     * @param key     json key名称
-     * @return HLLSketch数据数组
-     */
-    public static byte[] getHllSketch(Map<String, Object> jsonMap, String key) {
-        try {
-            HllSketch hllSketchResult = (HllSketch) jsonMap.getOrDefault(key, null);
-            if (hllSketchResult != null) {
-                return hllSketchResult.toUpdatableByteArray();
-            }
-        } catch (RuntimeException e) {
-            logger.error("HllSketch data conversion exception,data may be empty! exception:{}", e);
-        }
-        return null;
-    }
-
-    /**
-     * long 类型检验转换方法,若为空返回基础值
-     *
-     * @return Long value
-     */
-    public static Long getLong(Map<String, Object> jsonMap, String property) {
-        Object value = jsonMap.getOrDefault(property, null);
-        Long longVal = TypeUtils.castToLong(value);
-
-        if (longVal == null) {
-            return 0L;
-        }
-
-        return longVal;
-    }
-
-    public static String getString(Map<String, Object> jsonMap, String property) {
-        Object value = jsonMap.getOrDefault(property, null);
-        if (value == null) {
-            return null;
-        }
-
-        if (value instanceof Map) {
-            return JsonMapper.toJsonString(value);
-        }
-
-        if (value instanceof List) {
-            return JsonMapper.toJsonString(value);
-        }
-
-        return value.toString();
-    }
-
-    /**
-     * 更新属性值的方法
-     *
-     * @param jsonMap  原始日志json map
-     * @param property 更新的key
-     * @param value    更新的值
-     */
-    public static void setValue(Map<String, Object> jsonMap, String property, Object value) {
-        try {
-            jsonMap.put(property, value);
-        } catch (RuntimeException e) {
-            logger.error("The JSON set value is abnormal,the error message is :", e);
-        }
-    }
-
     /**
      * 通过获取String类型的网关schema链接来获取map，用于生成一个Object类型的对象
      * 用于反射生成schema类型的对象的一个map集合

src/test/java/com/zdjizhi/DatasketchesTest.java

@@ -1,16 +1,15 @@
 package com.zdjizhi;
 
-import com.alibaba.fastjson.JSONObject;
+import cn.hutool.json.JSONUtil;
+import com.alibaba.fastjson2.*;
+import com.zdjizhi.utils.JsonMapper;
 import org.apache.datasketches.hll.HllSketch;
 import org.apache.datasketches.hll.Union;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.Test;
 
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Properties;
-import java.util.Random;
+import java.util.*;
 
 /**
  * @author qidaijie
@@ -80,6 +79,7 @@ public class DatasketchesTest {
 
     @Test
     public void HllSketchDruidTest() {
+        HashMap<String, Object> dataMap = new HashMap<>();
 
         HashSet<String> strings = new HashSet<>();
 
@@ -135,15 +135,39 @@ public class DatasketchesTest {
 
         //CompactByte
         result.setIp_object(sketch_result2.toCompactByteArray());
-        System.out.println(result.toString());
-        sendMessage(result);
+//        System.out.println(result.toString());
+        //sendMessage(JsonMapper.toJsonString(result);
 
 
         //UpdatableByte
         result.setIp_object(sketch_result2.toUpdatableByteArray());
-        System.out.println(result.toString());
-        sendMessage(result);
+//        System.out.println(result.toString());
+        //sendMessage(JsonMapper.toJsonString(result);
 
+        //Hashmap
+        dataMap.put("app_name", "TEST");
+        dataMap.put("protocol_stack_id", "HTTP");
+        dataMap.put("vsys_id", 1);
+        dataMap.put("stat_time", 1681370100);
+        dataMap.put("client_ip_sketch", sketch_result2.toUpdatableByteArray());
+
+        System.out.println("Jackson:" + JsonMapper.toJsonString(dataMap));
+        System.out.println("FastJson2:" + JSONObject.toJSONString(dataMap));
+        System.out.println("Hutool:" + JSONUtil.toJsonStr(dataMap) + "\n\n");
+
+        byte[] toJSONB = JSONB.toBytes(dataMap);
+//        sendMessage(toJSONB);
+        JSONObject jsonObject = JSONB.parseObject(toJSONB);
+        System.out.println("FastJson2 Byte(JSONB):" + jsonObject.toJSONString() + "\n\n");
+
+
+        dataMap.put("client_ip_sketch", Base64.getEncoder().encodeToString(sketch_result2.toUpdatableByteArray()));
+        System.out.println("FastJson2 Byte(Base64):" + JSONObject.toJSONString(dataMap));
+        System.out.println("Hutool Byte(Base64):" + JSONObject.toJSONString(dataMap));
+        System.out.println(JSONUtil.toJsonStr(dataMap));
+
+
+//        sendMessage(JSONObject.toJSONString(dataMap));
     }
 
 
@@ -157,7 +181,7 @@ public class DatasketchesTest {
         return v4_1 + "." + v4_2 + "." + v4_3 + "." + v4_4;
     }
 
-    private static void sendMessage(Result result) {
+    private static void sendMessage(Object message) {
         Properties props = new Properties();
         //kafka地址
         props.put("bootstrap.servers", "192.168.44.12:9092");
@@ -165,11 +189,13 @@ public class DatasketchesTest {
         props.put("retries", 0);
         props.put("linger.ms", 1);
         props.put("buffer.memory", 67108864);
-        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+//        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+//        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
         KafkaProducer<String, Object> kafkaProducer = new KafkaProducer<String, Object>(props);
 
-        kafkaProducer.send(new ProducerRecord<String, Object>("TRAFFIC-PROTOCOL-TEST", JSONObject.toJSONString(result)));
+        kafkaProducer.send(new ProducerRecord<String, Object>("TRAFFIC-PROTOCOL-TEST", message));
 
         kafkaProducer.close();
     }

src/test/java/com/zdjizhi/FunctionTest.java

@@ -3,7 +3,6 @@ package com.zdjizhi;
 import com.jayway.jsonpath.JsonPath;
 import com.zdjizhi.common.StreamAggregateConfig;
 import com.zdjizhi.utils.StringUtil;
-import com.zdjizhi.utils.json.JsonTypeUtil;
 import org.junit.Test;
 
 import java.util.Arrays;
@@ -68,12 +67,4 @@ public class FunctionTest {
             }
         }
     }
-
-    @Test
-    public void longSumTest() {
-        Long res1 = JsonTypeUtil.checkLongValue(123);
-        Long res2 = JsonTypeUtil.checkLongValue("123");
-
-        System.out.println(res1 + res2);
-    }
 }

src/test/java/com/zdjizhi/JsonTest.java

@@ -0,0 +1,145 @@
+package com.zdjizhi;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.jayway.jsonpath.DocumentContext;
+import com.jayway.jsonpath.JsonPath;
+import net.sf.cglib.beans.BeanGenerator;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2023/4/1218:08
+ */
+public class JsonTest {
+
+//    /**
+//     * 在内存中加载反射类用的map
+//     */
+//    private static HashMap<String, Class> map = MetaDataParse.getMapFromHttp(FlowWriteConfig.SCHEMA_HTTP);
+//
+//    /**
+//     * 反射成一个类
+//     */
+//    private static Object mapObject = MetaDataParse.generateObject(map);
+
+    @Test
+    public void fastJson2Test() {
+//        SerializerFeature.DisableCircularReferenceDetect
+//        SerializerFeature.WriteNullStringAsEmpty
+//        SerializerFeature.WriteNullNumberAsZero
+        HashMap<String, Class> classHashMap = new HashMap<>();
+
+        String message = "{\"common_schema_type\":\"HTTP\",\"common_sessions\":1,\"http_request_line\":\"GET sampleFile.html HTTP/1.1\",\"http_host\":\"www.texaslotto.com\",\"http_url\":\"www.texaslotto.com/sampleFile.html\",\"http_user_agent\":\"xPTS/2.0\",\"http_response_line\":\"HTTP/1.1 200 OK\",\"http_isn\":1953597368,\"http_proxy_flag\":0,\"http_version\":\"http1\",\"http_response_latency_ms\":1,\"http_session_duration_ms\":2,\"http_response_content_type\":\"text/html\",\"http_sequence\":80,\"common_protocol_label\":\"ETHERNET.IPv4.UDP.GTP.IPv4.TCP\",\"common_c2s_byte_diff\":17200,\"common_c2s_pkt_diff\":120,\"common_s2c_byte_diff\":16490,\"common_s2c_pkt_diff\":81,\"common_c2s_ipfrag_num\":0,\"common_s2c_ipfrag_num\":0,\"common_first_ttl\":64,\"common_c2s_tcp_unorder_num\":0,\"common_s2c_tcp_unorder_num\":0,\"common_c2s_tcp_lostlen\":0,\"common_s2c_tcp_lostlen\":0,\"common_c2s_pkt_retrans\":0,\"common_s2c_pkt_retrans\":0,\"common_c2s_byte_retrans\":0,\"common_s2c_byte_retrans\":0,\"common_flags\":24720,\"common_flags_identify_info\":\"{\\\"Server is Local\\\":1,\\\"Inbound\\\":201,\\\"C2S\\\":1,\\\"S2C\\\":2}\",\"common_direction\":73,\"common_app_full_path\":\"http\",\"common_app_label\":\"http\",\"common_tcp_client_isn\":1953597368,\"common_tcp_server_isn\":1950649408,\"common_server_ip\":\"192.50.199.25\",\"common_client_ip\":\"192.50.146.197\",\"common_server_port\":80,\"common_client_port\":22533,\"common_stream_dir\":3,\"common_address_type\":4,\"common_address_list\":\"IPv4_TCP<22533-80-192.50.146.197-192.50.199.25>|GTP<111001144-851056526>|IPv4_UDP<2152-2152-192.50.235.220-192.50.135.83>|MAC<000c299b2fa4-000c2915b4f4>\",\"common_start_time\":1680475247,\"common_end_time\":1680475247,\"common_con_duration_ms\":23,\"common_s2c_pkt_num\":81,\"common_s2c_byte_num\":16490,\"common_c2s_pkt_num\":120,\"common_c2s_byte_num\":17200,\"common_establish_latency_ms\":2,\"common_client_location\":\"日本.Unknown.Unknown\",\"common_server_location\":\"日本.Unknown.Unknown\",\"common_service_category\":[6223,6219,5093,5089],\"common_apn\":\"cmiott.owflr.mcto60g.com\",\"common_imsi\":\"460045157091460\",\"common_imei\":\"8626070583005833\",\"common_phone_number\":\"861440152028973\",\"common_tunnel_endpoint_a_desc\":\"test_50_gtp\",\"common_tunnel_endpoint_b_desc\":\"test_50_gtp\",\"common_tunnels\":[{\"tunnels_schema_type\":\"GTP\",\"gtp_a2b_teid\":111001144,\"gtp_b2a_teid\":851056526,\"gtp_endpoint_a_ip\":\"192.50.235.220\",\"gtp_endpoint_b_ip\":\"192.50.135.83\",\"gtp_endpoint_a_port\":2152,\"gtp_endpoint_b_port\":2152},{\"tunnels_schema_type\":\"MULTIPATH_ETHERNET\",\"c2s_source_mac\":\"00:0c:29:9b:2f:a4\",\"c2s_destination_mac\":\"00:0c:29:15:b4:f4\",\"s2c_source_mac\":\"00:0c:29:15:b4:f4\",\"s2c_destination_mac\":\"00:0c:29:9b:2f:a4\"}],\"common_stream_trace_id\":\"578829229323951427\",\"common_l4_protocol\":\"IPv4_TCP\",\"common_sled_ip\":\"192.168.40.161\",\"common_device_id\":\"unknown\",\"common_device_tag\":\"{\\\"tags\\\":[{\\\"tag\\\":\\\"device_group\\\",\\\"value\\\":\\\"group-xxg-7400\\\"},{\\\"tag\\\":\\\"data_center\\\",\\\"value\\\":\\\"center-xxg-7400\\\"}]}\",\"common_t_vsys_id\":1,\"common_policy_id\":0,\"common_service\":2,\"common_action\":0,\"common_vsys_id\":1}";
+
+        JSONObject json = JSON.parseObject(message);
+        Object mapObject = generateObject(classHashMap);
+        Object object = JSON.parseObject(message, mapObject.getClass());
+
+
+        System.out.println(json.get("common_schema_type"));
+        json.put("common_schema_type", "SSH");
+
+
+        System.out.println(json.toJSONString());
+
+
+    }
+
+    private static Class getClassName(String type) {
+        Class clazz;
+
+        switch (type) {
+            case "int":
+                clazz = Integer.class;
+                break;
+            case "string":
+                clazz = String.class;
+                break;
+            case "long":
+                clazz = long.class;
+                break;
+            case "array":
+                clazz = List.class;
+                break;
+            case "double":
+                clazz = double.class;
+                break;
+            case "float":
+                clazz = float.class;
+                break;
+            case "char":
+                clazz = char.class;
+                break;
+            case "byte":
+                clazz = byte.class;
+                break;
+            case "boolean":
+                clazz = boolean.class;
+                break;
+            case "short":
+                clazz = short.class;
+                break;
+            default:
+                clazz = String.class;
+        }
+        return clazz;
+    }
+
+    /**
+     * 根据反射生成对象的方法
+     *
+     * @param properties 反射类用的map
+     * @return 生成的Object类型的对象
+     */
+    private static Object generateObject(Map properties) {
+        BeanGenerator generator = new BeanGenerator();
+        Set keySet = properties.keySet();
+        for (Object aKeySet : keySet) {
+            String key = (String) aKeySet;
+            generator.addProperty(key, (Class) properties.get(key));
+        }
+        return generator.create();
+    }
+
+//    /**
+//     * 通过获取String类型的网关schema链接来获取map，用于生成一个Object类型的对象
+//     *
+//     * @param http 网关schema地址
+//     * @return 用于反射生成schema类型的对象的一个map集合
+//     */
+//    public static HashMap<String, Class> getMapFromHttp(String schema) {
+//        HashMap<String, Class> map = new HashMap<>(16);
+//
+//        DocumentContext parse = JsonPath.parse(schema);
+//
+//        //获取fields，并转化为数组，数组的每个元素都是一个name doc type
+//        com.alibaba.fastjson.JSONObject schemaJson = com.alibaba.fastjson.JSON.parseObject(data.toString());
+//        JSONArray fields = (JSONArray) schemaJson.get("fields");
+//
+//        for (Object field : fields) {
+//            String filedStr = field.toString();
+//            if (checkKeepField(filedStr)) {
+//                String name = JsonPath.read(filedStr, "$.name").toString();
+//                String type = JsonPath.read(filedStr, "$.type").toString();
+//                if (type.contains("{")) {
+//                    type = JsonPath.read(filedStr, "$.type.type").toString();
+//                }
+//                //组合用来生成实体类的map
+//                map.put(name, getClassName(type));
+//            } else {
+//                dropList.add(filedStr);
+//            }
+//        }
+//        return map;
+//    }
+}