优化代码

src/main/java/com/zorkdata/tools/kafka/CommonProducerPool.java

@@ -19,7 +19,7 @@ public class CommonProducerPool implements Closeable {
      */
     private int index = 0;
 
-    private static CommonProducerPool producerInstance = null;
+    private static CommonProducerPool producerInstance;
 
     public static CommonProducerPool getInstance() {
         if (producerInstance == null) {

src/main/java/com/zorkdata/tools/mock/MockFilebeatData.java

+package com.zorkdata.tools.mock;
+
+import com.alibaba.fastjson.JSONObject;
+import com.zorkdata.tools.oldkafka.Producer;
+import com.zorkdata.tools.oldkafka.ProducerPool;
+
+import java.time.Instant;
+import java.util.Random;
+
+import static java.lang.System.currentTimeMillis;
+
+/**
+ * @author: LiaoMingtao
+ * @date: 2020/6/30
+ */
+public class MockFilebeatData {
+
+    private static String buildMsg() {
+        JSONObject filebeatJson = new JSONObject();
+        JSONObject metadataJson = new JSONObject();
+        metadataJson.put("beat", "filebeat");
+        metadataJson.put("type", "doc");
+        metadataJson.put("version", "6.8.1");
+        JSONObject inputJson = new JSONObject();
+        inputJson.put("type", "log");
+        JSONObject beatJson = new JSONObject();
+        beatJson.put("name", "zorkdata-151");
+        beatJson.put("hostname", "zorkdata-151");
+        beatJson.put("version", "6.8.1");
+        JSONObject hostJson = new JSONObject();
+        hostJson.put("name", "zorkdata-151");
+        hostJson.put("architecture", "x86_64");
+        hostJson.put("id", "8e3dfc85999b4e02bae4adf4b92b909a");
+        hostJson.put("containerized", "false");
+        JSONObject logJson = new JSONObject();
+        logJson.put("file", "{ \"path\": \"/var/log/nginx/access.log\" }");
+        filebeatJson.put("@timestamp", Instant.now().toString());
+        filebeatJson.put("source", "/var/log/nginx/access.log");
+        filebeatJson.put("offset", String.valueOf(currentTimeMillis()));
+        filebeatJson.put("message", "10:06:41.335 功能请求 IP:182.140.129.3 MAC:F8A963586DFF 线程:00004364 通道ID:4 事务ID:16 请求:(0-98)集成客户校验(*) 营业部:(0001)国金证券集中交易(*)\\n66650109|************|XshR9/S5SDE=|8|0||12|7.37.0||||||||||0||0|182.140.129.3;PENGKANG;Administrator;83025;Intel(R)Core(TM)i7-4510UCPU@2.00GHz*4;bfebfbff00040651-GenuineIntel;Windows7 Service Pack 1 (Build 7601);182.140.129.3,0.0.0.0,0.0.0.0;F8A963586DFF,00FF8C535532,A0A8CD0D00B0;TF655AWJ16NG2L,143116404707;07/15/2014;8DC03929-0822-453C-A2D5-EFBE95E359BE;182.140.129.3;;NTFS;0C17-8FD7;C:;113G;HTS725050A7E630;GH2Z;TF655AWJ16NG2L;|||||2,Mar  1 2018,10:22:32|0|||GETLOGINPARAM||7.37,6.01,Mar  1 2018,10:37:07|8噝\\\\\\\\5\\\\\\\\3||||\\\\n10:06:41.491 调用失败 IP:182.140.129.3 MAC:F8A963586DFF 线程:00004364 通道ID:4 事务ID:16 请求:(0-98)集成客户校验(*) 营业部:(0001)国金证券集中交易(*) 耗时A:156 耗时B:0 排队:0\\\\n-4|资金账号或密码错误!|0|||\\\\n10:06:52.678 系统信息 开始关闭交易中心服务。\\\\n10:06:53.303 系统信息 (HS_TCP2.dll)连接守护线程退出!\\\\n10:06:53.335 系统信息 (HS_TCP2.dll)\\\\\\\"刷新约定购回标的证券信息\\\\\\\"线程成功退出!(记录总条数:3536)\\\\n10:06:54.413 系统信息 港股行情服务: 保存代码表(港股)缓存。\\\\n10:06:54.678 系统信息 深沪行情服务: 保存代码表缓存。\\\\n10:06:54.960 系统信息 交易中心服务已经成功关闭。\\\\n10:06:54.960 系统信息 系统正常关闭\\\\n");
+
+        Random random = new Random();
+        int i = random.nextInt(10);
+        filebeatJson.put("appsystem", "test_appsystem" + i * 2);
+        filebeatJson.put("appprogramname", "test_appprogramname" + i * 3);
+        filebeatJson.put("logTypeName", "test_topic_log" + i * 4);
+        filebeatJson.put("servicename", "test_servicename" + i * 5);
+        filebeatJson.put("servicecode", "test_cdde" + i * 6);
+        filebeatJson.put("collectorruleid", "1");
+
+        filebeatJson.put("@metadata", metadataJson);
+        filebeatJson.put("input", inputJson);
+        filebeatJson.put("beat", beatJson);
+        filebeatJson.put("host", hostJson);
+        filebeatJson.put("log", logJson);
+        filebeatJson.put("prospector", inputJson);
+        return filebeatJson.toJSONString();
+    }
+
+
+    public static void main(String[] args) throws InterruptedException {
+        long size = 10000000L * 10;
+        // long size = 1;
+        for (int i = 0; i < size; i++) {
+            String json = buildMsg();
+
+            Producer producer = ProducerPool.getInstance().getProducer();
+            producer.sendLog("analysis1y", json);
+        }
+        Thread.sleep(1000);
+    }
+}

src/main/java/com/zorkdata/tools/mock/MockGrok.java

@@ -25,7 +25,7 @@ import static java.lang.System.currentTimeMillis;
 public class MockGrok {
 
     private static String MESSAGE = "%s - - [29/Jun/2020:16:09:23 +0800] \"%s /webserver/scene/getSceneList.do?menuItemId=1039&sceneGroupId=&templateFlag=%s HTTP/1.0\" %s 3167";
-    private static String MESSAGE1 = "192.168.1.151 - - [29/Jun/2020:16:09:23 +0800] \"GET /webserver/scene/getSceneList.do?menuItemId=1039&sceneGroupId=&templateFlag= HTTP/1.0\" 200 3167";
+    private static String temp = "192.168.1.151 - - [29/Jun/2020:16:09:23 +0800] \"GET /webserver/scene/getSceneList.do?menuItemId=1039&sceneGroupId=&templateFlag= HTTP/1.0\" 200 3167";
 
     private static String buildMessage() {
         Random random = new Random();
@@ -76,12 +76,13 @@ public class MockGrok {
     }
 
     public static void main(String[] args) throws IOException, InterruptedException {
-        long size = 1;
+        long size = 1000;
+        String topicName = "test";
         // long size = 10000000L * 10;
         for (int i = 0; i < size; i++) {
             String json = buildJson(buildMessage());
             CommonProducer producer = CommonProducerPool.getInstance().getProducer();
-            producer.sendLog("test", json);
+            producer.sendLog(topicName, json);
         }
         Thread.sleep(1000);
     }

src/main/java/com/zorkdata/tools/mock/MockStreamxLogAvro.java

 package com.zorkdata.tools.mock;
 
+import com.zorkdata.tools.kafka.CommonProducer;
+import com.zorkdata.tools.kafka.CommonProducerPool;
+import com.zorkdata.tools.oldkafka.Producer;
+import com.zorkdata.tools.oldkafka.ProducerPool;
+import com.zorkdata.tools.utils.DateUtil;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * @author MSI-Gaming
+ */
 public class MockStreamxLogAvro {
+
+    private static String getRandomOffset() {
+        Random random = new Random();
+        long l = random.nextInt(10000);
+        return String.valueOf(l);
+    }
+
+    private static Map<String, String> getRandomDimensions() {
+        Random random = new Random();
+        int i = random.nextInt(10);
+        Map<String, String> dimensions = new HashMap<>(4);
+        dimensions.put("hostname", "zorkdata" + i);
+        dimensions.put("ip", "192.168.1." + i);
+        dimensions.put("appprogramname", "tc50");
+        dimensions.put("appsystem", "tdx");
+        return dimensions;
+    }
+
+    private static String[] codes = {
+            "AO", "AF", "AL", "DZ", "AD", "AI", "AG", "AR", "AM", "AU",
+            "AT", "AZ", "BS", "BH", "BD", "BB", "BY", "BE", "BZ", "BJ"
+    };
+
+    private static String getRandomCountryCode() {
+        return codes[new Random(codes.length).nextInt(codes.length)];
+    }
+
+    private static Map<String, String> getRandomNormalFields() {
+        Map<String, String> normalFields = new HashMap<>();
+        normalFields.put("message", "data update success");
+        normalFields.put("countryCode", getRandomCountryCode());
+        return normalFields;
+    }
+
+    public static void main(String[] args) throws Exception {
+        // long size = 10000000L * 1;
+        long size = 10;
+        // String topicName = "log2metric1y";
+        String topicName = "test";
+        for (int i = 0; i < size; i++) {
+            String logTypeName = "streamx_log_avro";
+            String timestamp = DateUtil.getUTCTimeStr();
+            String source = "/var/log/xiesen.log";
+            String offset = getRandomOffset();
+            Map<String, String> dimensions = getRandomDimensions();
+            Map<String, Double> measures = new HashMap<>(1);
+            Map<String, String> normalFields = getRandomNormalFields();
+            CommonProducer producer = CommonProducerPool.getInstance().getProducer();
+            producer.sendLog(topicName, logTypeName, timestamp, source, offset, dimensions,
+                    measures, normalFields);
+        }
+        Thread.sleep(1000);
+    }
 }

src/main/java/com/zorkdata/tools/mock/MockStreamxMetricAvro.java

+package com.zorkdata.tools.mock;
+
+import com.zorkdata.tools.kafka.CommonProducer;
+import com.zorkdata.tools.kafka.CommonProducerPool;
+import com.zorkdata.tools.oldkafka.Producer;
+import com.zorkdata.tools.oldkafka.ProducerPool;
+import com.zorkdata.tools.utils.DateUtil;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+/**
+ * @author: LiaoMingtao
+ * @date: 2020/6/30
+ */
+public class MockStreamxMetricAvro {
+
+    private static Map<String, String> getRandomDimensions() {
+        Random random = new Random();
+        int i = random.nextInt(10);
+        Map<String, String> dimensions = new HashMap<>();
+
+        dimensions.put("hostname", "zorkdata" + i);
+        dimensions.put("ip", "192.168.1." + i);
+        dimensions.put("appprogramname", "tc50");
+        dimensions.put("appsystem", "tdx");
+
+        return dimensions;
+    }
+
+    private static String[] codes = {
+            "AO", "AF", "AL", "DZ", "AD", "AI", "AG", "AR", "AM", "AU",
+            "AT", "AZ", "BS", "BH", "BD", "BB", "BY", "BE", "BZ", "BJ"
+    };
+
+    private static String getRandomCountryCode() {
+        return codes[new Random(codes.length).nextInt(codes.length)];
+    }
+
+    private static Map<String, String> getRandomNormalFields() {
+        Map<String, String> normalFields = new HashMap<>();
+        normalFields.put("message", "data update success");
+        normalFields.put("countryCode", getRandomCountryCode());
+        return normalFields;
+    }
+
+    private static Map<String, Double> getMetrics() {
+        Map<String, Double> metrics = new HashMap<>(1);
+        Random random = new Random();
+        int i = random.nextInt(10);
+        metrics.put("metric1", i * 0.1);
+        metrics.put("metric2", i * 0.2);
+        metrics.put("metric3", i * 3.65);
+        return metrics;
+    }
+
+
+    public static void main(String[] args) throws Exception {
+        long size = 10000000 * 20;
+        String topicName = "metric2metric20000w";
+        for (int i = 0; i < size; i++) {
+            String metricSetName = "streamx_metric_avro";
+            String timestamp = DateUtil.getUTCTimeStr();
+            Map<String, String> dimensions = getRandomDimensions();
+            Map<String, Double> metrics = getMetrics();
+            CommonProducer producer = CommonProducerPool.getInstance().getProducer();
+            producer.sendMetric(topicName, metricSetName, timestamp, dimensions, metrics);
+        }
+        Thread.sleep(1000);
+    }
+}