<dev>

1. mock

pom.xml

@@ -5,8 +5,8 @@
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>com.zorkdata</groupId>
-    <artifactId>com.zorkdata.tools.mock.MockStreamxLogAvro</artifactId>
-    <version>1.0-SNAPSHOT</version>
+    <artifactId>mockDysLogData</artifactId>
+    <version>1.0</version>
     <repositories>
         <!-- Using Local Nexus Maven Repository -->
         <repository>
@@ -133,6 +133,9 @@
                             <!-- 应用的main class-->
                             <mainClass>com.zorkdata.tools.mock.MockLogMergeData</mainClass>
                         </manifest>
+                        <manifestEntries>
+                            <Class-Path>resources/</Class-Path>
+                        </manifestEntries>
                     </archive>
                     <!--  过滤掉不希望包含在jar中的文件-->
                     <excludes>

src/main/assembly/package.xml

@@ -7,9 +7,7 @@
     <!-- Adds dependencies to zip package under lib directory -->
     <dependencySets>
         <dependencySet>
-            <!--
-               不使用项目的artifact，第三方jar不要解压，打包进zip文件的lib目录
-           -->
+            <!-- 不使用项目的artifact，第三方jar不要解压，打包进zip文件的lib目录 -->
             <useProjectArtifact>false</useProjectArtifact>
             <outputDirectory>lib</outputDirectory>
             <unpack>false</unpack>

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

@@ -18,6 +18,7 @@ public class Config {
     public String kafkaServers;
     public int kafkaBathSize;
     public String topicName;
+    public int sendSize;
 
     public static final Config INSTANCE = new Config();
 
@@ -26,10 +27,11 @@ public class Config {
      */
     public Config() {
         try {
-            Properties properties = PropertiesUtil.getProperties("/config.properties");
+            Properties properties = PropertiesUtil.getPropertiesByUserDir("/config.properties");
             kafkaServers = properties.getProperty("kafka.servers");
             kafkaBathSize = Integer.parseInt(properties.getProperty("kafka.batch.size"));
             topicName = properties.getProperty("kafka.topic.name");
+            sendSize = Integer.parseInt(properties.getProperty("send.size"));
         } catch (Exception e) {
             logger.error(e.toString());
             System.exit(1);

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

@@ -3,6 +3,7 @@ package com.zorkdata.tools.mock;
 import com.zorkdata.tools.kafka.CommonProducer;
 import com.zorkdata.tools.kafka.CommonProducerPool;
 import com.zorkdata.tools.kafka.Config;
+import lombok.extern.slf4j.Slf4j;
 
 import java.text.SimpleDateFormat;
 import java.util.*;
@@ -11,6 +12,7 @@ import java.util.*;
  * @author: LiaoMingtao
  * @date: 2020/9/29
  */
+@Slf4j
 public class MockLogMergeData {
     private static String req = "{\n" +
             "\t\"@timestamp\": \"2020-09-28T19:20:33.876Z\",\n" +
@@ -200,25 +202,31 @@ public class MockLogMergeData {
     // }
 
     public static void main(String[] args) throws InterruptedException {
-        int size = 1000;
         Config config = new Config();
         String topicName = config.getTopicName();
+        log.info("kafka.servers={}", config.getKafkaServers());
+        log.info("kafka.topic.name={}", config.getTopicName());
+        log.info("kafka.batch.size={}", config.getKafkaBathSize());
+        log.info("send.size={}", config.getSendSize());
+        int size = config.getSendSize() / 2;
         for (int i = 0; i < size; i++) {
-            UUID uuid = UUID.randomUUID();
             Date date1 = new Date();
             SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSSSS");
+            UUID uuid = UUID.randomUUID();
             String date1String = simpleDateFormat.format(date1);
             Thread.sleep(getSleepTime());
             String reqString = String.format(req, date1String, uuid.toString());
             Date date2 = new Date();
             String date2String = simpleDateFormat.format(date2);
             String respString = String.format(resp, date2String, uuid.toString());
-            // System.out.printf(reqString);
-            // System.out.printf(respString);
             CommonProducer producer = CommonProducerPool.getInstance().getProducer();
             producer.sendJson(topicName, reqString);
             producer.sendJson(topicName, respString);
         }
+        log.info("kafka.servers={}", config.getKafkaServers());
+        log.info("kafka.topic.name={}", config.getTopicName());
+        log.info("kafka.batch.size={}", config.getKafkaBathSize());
+        log.info("send.size={}", config.getSendSize());
     }
 
     // public static void main(String[] args) throws InterruptedException {

src/main/java/com/zorkdata/tools/utils/PropertiesUtil.java

 package com.zorkdata.tools.utils;
 
+import lombok.extern.slf4j.Slf4j;
+import org.apache.log4j.PropertyConfigurator;
+
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.util.Properties;
 
 /**
  * @author DeleMing
  */
+@Slf4j
 public class PropertiesUtil {
 
+    private PropertiesUtil() {
+
+    }
+
     /**
      * 根据文件名获取该properties对象
      *
      * @param propertiesFileName 配置文件名称
      * @return properties
      */
-    public static Properties getProperties(String propertiesFileName) throws Exception {
+    public static Properties getProperties(String propertiesFileName) {
         Properties properties = new Properties();
         InputStream inputStream = null;
         try {
             inputStream = PropertiesUtil.class.getResourceAsStream(propertiesFileName);
             properties.load(inputStream);
+        } catch (IOException e) {
+            log.error("读取配置文件失败:", e);
+            System.exit(1);
+        } finally {
+            if (inputStream != null) {
+                try {
+                    inputStream.close();
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+        return properties;
+    }
+
+    public static Properties getPropertiesByUserDir(String propertiesFileName) {
+        Properties properties = null;
+        InputStream inputStream = null;
+        try {
+            PropertyConfigurator.configure(System.getProperty("user.dir") + propertiesFileName);
+            properties = new Properties();
+            // 读取SRC下配置文件 --- 属于读取内部文件
+            // 读取系统外配置文件 (即Jar包外文件) --- 外部工程引用该Jar包时需要在工程下创建config目录存放配置文件
+            String filePath = System.getProperty("user.dir") + propertiesFileName;
+            inputStream = new BufferedInputStream(new FileInputStream(filePath));
+            properties.load(inputStream);
+        } catch (IOException e) {
+            log.error("读取配置文件失败:", e);
+            System.exit(1);
         } finally {
             if (inputStream != null) {
-                inputStream.close();
+                try {
+                    inputStream.close();
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
             }
         }
         return properties;

src/main/resources/config.properties

-#kafka.servers=kafka1:9092,kafka2:9092,kafka3:9092
+# kafka 
 kafka.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
-#kafka.servers=kafka-1:19092,kafka-2:19092,kafka-3:19092
-# ,kafka02:9092,kafka03:9092
-kafka.batch.size=1
-kafka.topic.name=gutoai
-key.serializer=org.apache.kafka.common.serialization.StringSerializer
-value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
-#value.serializer=org.apache.kafka.common.serialization.StringSerializer
-producer.theard.num=15
-log.size = 10000
-log.topic=flinkx
+kafka.batch.size=100
+kafka.topic.name=guotai
+
+# 
+send.size=10000
\ No newline at end of file