新增方法

pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>com.zorkdata</groupId>
-    <artifactId>mock-data</artifactId>
+    <artifactId>com.zorkdata.tools.mock.MockStreamxLogAvro</artifactId>
     <version>1.0-SNAPSHOT</version>
     <repositories>
         <!-- Using Local Nexus Maven Repository -->
@@ -144,7 +144,7 @@
                             <!--生成的manifest中classpath的前缀，因为要把第三方jar放到lib目录下，所以classpath的前缀是lib/-->
                             <classpathPrefix>lib/</classpathPrefix>
                             <!-- 应用的main class-->
-                            <mainClass>com.zorkdata.tools.mock.MockFilebeatDataToKafka</mainClass>
+                            <mainClass>com.zorkdata.tools.mock.MockStreamxLogAvro</mainClass>
                         </manifest>
                     </archive>
                     <!--  过滤掉不希望包含在jar中的文件-->

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

@@ -60,7 +60,7 @@ public class Producer {
         try {
             byte[] bytes = AvroSerializerFactory.getLogAvorSerializer().serializingLog(logTypeName, timestamp, source,
                     offset, dimensions, metrics, normalFields);
-            producer.send(new ProducerRecord<String, byte[]>(topic, "", bytes));
+            producer.send(new ProducerRecord<String, byte[]>(topic, null, bytes));
         } catch (Exception e) {
             e.printStackTrace();
         }
@@ -101,6 +101,17 @@ public class Producer {
         }
     }
 
+    public void sendMetric(String topic, String metricSetName, String timestamp, Map<String, String> dimensions,
+                           Map<String, Double> metrics) {
+        try {
+            byte[] bytes = AvroSerializerFactory.getMetricAvorSerializer().serializingMetric(metricSetName, timestamp,
+                    dimensions, metrics);
+            producer.send(new ProducerRecord<String, byte[]>(topic, "", bytes));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
 
 
 

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

@@ -76,7 +76,7 @@ public class MockStreamxLogAvro {
     }
 
     public static void main(String[] args) throws Exception {
-        long size = 10000 * 10000 * 10;
+        long size = 100000000;
 
         for (int i = 0; i < size; i++) {
             String logTypeName = "streamx_log_avro";
@@ -91,7 +91,7 @@ public class MockStreamxLogAvro {
 
             //System.out.println(printData(logTypeName, timestamp, source, offset, dimensions, measures, normalFields));
             Producer producer = ProducerPool.getInstance().getProducer();
-            producer.sendLog("logavro_1p", logTypeName, timestamp, source, offset, dimensions,
+            producer.sendLog("log2metric10000w", 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.alibaba.fastjson.JSONObject;
+import com.zorkdata.tools.kafka.Producer;
+import com.zorkdata.tools.kafka.ProducerPool;
+import com.zorkdata.tools.utils.DateUtil;
+import com.zorkdata.tools.utils.PropertiesUtil;
+import com.zorkdata.tools.utils.StringUtil;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+
+/**
+ * @author: LiaoMingtao
+ * @date: 2020/6/28
+ */
+public class MockStreamxMetricAvro {
+    private static long getSize(String propertiesName) throws Exception {
+
+        Properties properties = PropertiesUtil.getProperties(propertiesName);
+        long logSize = StringUtil.getLong(properties.getProperty("log.size", "5000").trim(), 1);
+        return logSize;
+    }
+
+    public static String printData(String logTypeName, String timestamp, String source, String offset,
+                                   Map<String, String> dimensions, Map<String, Double> metrics, Map<String, String> normalFields) {
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put("logTypeName", logTypeName);
+        jsonObject.put("timestamp", timestamp);
+        jsonObject.put("source", source);
+        jsonObject.put("offset", offset);
+        jsonObject.put("dimensions", dimensions);
+        jsonObject.put("measures", metrics);
+        jsonObject.put("normalFields", normalFields);
+        return jsonObject.toString();
+    }
+
+    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<>();
+
+        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() {
+        Random random = new Random(codes.length);
+        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;
+
+        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();
+
+            //System.out.println(printData(logTypeName, timestamp, source, offset, dimensions, measures, normalFields));
+            Producer producer = ProducerPool.getInstance().getProducer();
+            producer.sendMetric("metric2metric20000w", metricSetName, timestamp, dimensions, metrics);
+        }
+        Thread.sleep(1000);
+    }
+}