1.适配IP定位库v4/v6合并后的加载逻辑（GAL-436）。feature/tsg-23.11

2.适配bifang23.11接口（TSG-17675）。
3.移除广播流。
4.修复静态阈值判断部分的BUG。

40 files changed, 656 insertions, 1953 deletions

diff --git a/pom.xml b/pom.xml
index 7ee04ac..c55ca23 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,9 +6,14 @@
 
     <groupId>com.zdjizhi</groupId>
     <artifactId>flink-dos-detection</artifactId>
-    <version>1.0-SNAPSHOT</version>
+    <version>23.11</version>
+
+    <name>flink-dos-detection</name>
+    <url>http://www.example.com</url>
+
 
     <properties>
+        <galaxy.tools.version>1.2.1</galaxy.tools.version>
         <flink.version>1.13.1</flink.version>
         <hive.version>2.1.1</hive.version>
         <hadoop.version>2.7.1</hadoop.version>
@@ -152,7 +157,6 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-kafka_2.12</artifactId>
             <version>${flink.version}</version>
-            <!--<scope>provided</scope>-->
         </dependency>
 
         <!-- CLI dependencies -->
@@ -160,7 +164,7 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-clients_2.12</artifactId>
             <version>${flink.version}</version>
-            <!--<scope>provided</scope>-->
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.hadoop</groupId>
@@ -203,7 +207,7 @@
             <groupId>org.apache.hbase</groupId>
             <artifactId>hbase-client</artifactId>
             <version>2.2.3</version>
-            <!--<scope>provided</scope>-->
+            <scope>provided</scope>
             <exclusions>
                 <exclusion>
                     <artifactId>slf4j-log4j12</artifactId>
@@ -258,26 +262,6 @@
             <version>5.3.3</version>
         </dependency>
 
-        <dependency>
-            <groupId>com.zdjizhi</groupId>
-            <artifactId>galaxy</artifactId>
-            <version>1.1.3</version>
-            <exclusions>
-                <exclusion>
-                    <artifactId>slf4j-log4j12</artifactId>
-                    <groupId>org.slf4j</groupId>
-                </exclusion>
-                <exclusion>
-                    <artifactId>log4j-over-slf4j</artifactId>
-                    <groupId>org.slf4j</groupId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.google.guava</groupId>
-                    <artifactId>guava</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-
 
         <dependency>
             <groupId>com.alibaba.fastjson2</groupId>
@@ -354,6 +338,12 @@
         </dependency>
 
 
+
+        <dependency>
+            <groupId>com.geedgenetworks</groupId>
+            <artifactId>galaxy</artifactId>
+            <version>${galaxy.tools.version}</version>
+        </dependency>
     </dependencies>
 
 
diff --git a/src/main/java/com/zdjizhi/common/CustomFile.java b/src/main/java/com/zdjizhi/common/CustomFile.java
deleted file mode 100644
index 701024c..0000000
--- a/src/main/java/com/zdjizhi/common/CustomFile.java
+++ /dev/null
@@ -1,26 +0,0 @@
-package com.zdjizhi.common;
-
-import java.io.Serializable;
-
-public class CustomFile implements Serializable {
-
-    String fileName;
-
-    byte[] content;
-
-    public String getFileName() {
-        return fileName;
-    }
-
-    public void setFileName(String fileName) {
-        this.fileName = fileName;
-    }
-
-    public byte[] getContent() {
-        return content;
-    }
-
-    public void setContent(byte[] content) {
-        this.content = content;
-    }
-}
diff --git a/src/main/java/com/zdjizhi/common/DosDetectionThreshold.java b/src/main/java/com/zdjizhi/common/DosDetectionThreshold.java
index 401673a..dc0825b 100644
--- a/src/main/java/com/zdjizhi/common/DosDetectionThreshold.java
+++ b/src/main/java/com/zdjizhi/common/DosDetectionThreshold.java
@@ -9,110 +9,110 @@ import java.util.Objects;
  * @author wlh
  */
 public class DosDetectionThreshold implements Serializable {
-    private long profileId;
-    private String attackType;
-    private ArrayList<String> serverIpList;
-    private String serverIpAddr;
-    private long packetsPerSec;
-    private long bitsPerSec;
-    private long sessionsPerSec;
-    private int isValid;
-    private int vsysId;
-    private Integer[] superiorIds;
+    private long profile_id;
+    private String attack_type;
+    private ArrayList<String> server_ip_list;
+    private String server_ip_addr;
+    private long packets_per_sec;
+    private long bits_per_sec;
+    private long sessions_per_sec;
+    private int is_valid;
+    private int vsys_id;
+    private Integer[] superior_ids;
 
-    public long getProfileId() {
-        return profileId;
+    public long getProfile_id() {
+        return profile_id;
     }
 
-    public void setProfileId(long profileId) {
-        this.profileId = profileId;
+    public void setProfile_id(long profile_id) {
+        this.profile_id = profile_id;
     }
 
-    public String getAttackType() {
-        return attackType;
+    public String getAttack_type() {
+        return attack_type;
     }
 
-    public void setAttackType(String attackType) {
-        this.attackType = attackType;
+    public void setAttack_type(String attack_type) {
+        this.attack_type = attack_type;
     }
 
-    public ArrayList<String> getServerIpList() {
-        return serverIpList;
+    public ArrayList<String> getServer_ip_list() {
+        return server_ip_list;
     }
 
-    public void setServerIpList(ArrayList<String> serverIpList) {
-        this.serverIpList = serverIpList;
+    public void setServer_ip_list(ArrayList<String> server_ip_list) {
+        this.server_ip_list = server_ip_list;
     }
 
-    public String getServerIpAddr() {
-        return serverIpAddr;
+    public String getServer_ip_addr() {
+        return server_ip_addr;
     }
 
-    public void setServerIpAddr(String serverIpAddr) {
-        this.serverIpAddr = serverIpAddr;
+    public void setServer_ip_addr(String server_ip_addr) {
+        this.server_ip_addr = server_ip_addr;
     }
 
-    public long getPacketsPerSec() {
-        return packetsPerSec;
+    public long getPackets_per_sec() {
+        return packets_per_sec;
     }
 
-    public void setPacketsPerSec(long packetsPerSec) {
-        this.packetsPerSec = packetsPerSec;
+    public void setPackets_per_sec(long packets_per_sec) {
+        this.packets_per_sec = packets_per_sec;
     }
 
-    public long getBitsPerSec() {
-        return bitsPerSec;
+    public long getBits_per_sec() {
+        return bits_per_sec;
     }
 
-    public void setBitsPerSec(long bitsPerSec) {
-        this.bitsPerSec = bitsPerSec;
+    public void setBits_per_sec(long bits_per_sec) {
+        this.bits_per_sec = bits_per_sec;
     }
 
-    public long getSessionsPerSec() {
-        return sessionsPerSec;
+    public long getSessions_per_sec() {
+        return sessions_per_sec;
     }
 
-    public void setSessionsPerSec(long sessionsPerSec) {
-        this.sessionsPerSec = sessionsPerSec;
+    public void setSessions_per_sec(long sessions_per_sec) {
+        this.sessions_per_sec = sessions_per_sec;
     }
 
-    public int getIsValid() {
-        return isValid;
+    public int getIs_valid() {
+        return is_valid;
     }
 
-    public void setIsValid(int isValid) {
-        this.isValid = isValid;
+    public void setIs_valid(int is_valid) {
+        this.is_valid = is_valid;
     }
 
-    public int getVsysId() {
-        return vsysId;
+    public int getVsys_id() {
+        return vsys_id;
     }
 
-    public void setVsysId(int vsysId) {
-        this.vsysId = vsysId;
+    public void setVsys_id(int vsys_id) {
+        this.vsys_id = vsys_id;
     }
 
-    public Integer[] getSuperiorIds() {
-        return superiorIds;
+    public Integer[] getSuperior_ids() {
+        return superior_ids;
     }
 
-    public void setSuperiorIds(Integer[] superiorIds) {
-        this.superiorIds = superiorIds;
+    public void setSuperior_ids(Integer[] superior_ids) {
+        this.superior_ids = superior_ids;
     }
 
     @Override
     public String toString() {
         return "DosDetectionThreshold{" +
-                "profileId='" + profileId + '\'' +
-                ", attackType='" + attackType + '\'' +
-                ", serverIpList=" + serverIpList +
-                ", serverIpAddr='" + serverIpAddr + '\'' +
-                ", packetsPerSec=" + packetsPerSec +
-                ", bitsPerSec=" + bitsPerSec +
-                ", sessionsPerSec=" + sessionsPerSec +
-                ", isValid=" + isValid +
-                ", vsysId=" + vsysId +
-                ", superiorIds=" + Arrays.toString(superiorIds) +
+                "profile_id=" + profile_id +
+                ", attack_type='" + attack_type + '\'' +
+                ", server_ip_list=" + server_ip_list +
+                ", server_ip_addr='" + server_ip_addr + '\'' +
+                ", packets_per_sec=" + packets_per_sec +
+                ", bits_per_sec=" + bits_per_sec +
+                ", sessions_per_sec=" + sessions_per_sec +
+                ", is_valid=" + is_valid +
+                ", vsys_id=" + vsys_id +
+                ", superior_ids=" + Arrays.toString(superior_ids) +
                 '}';
     }
 }
diff --git a/src/main/java/com/zdjizhi/common/DosVsysId.java b/src/main/java/com/zdjizhi/common/DosVsysId.java
index 0369f69..b5465f2 100644
--- a/src/main/java/com/zdjizhi/common/DosVsysId.java
+++ b/src/main/java/com/zdjizhi/common/DosVsysId.java
@@ -4,7 +4,7 @@ import java.util.Arrays;
 
 public class DosVsysId {
     private Integer id;
-    private Integer[] superiorIds;
+    private Integer[] superior_ids;
 
     public Integer getId() {
         return id;
@@ -14,19 +14,19 @@ public class DosVsysId {
         this.id = id;
     }
 
-    public Integer[] getSuperiorIds() {
-        return superiorIds;
+    public Integer[] getSuperior_ids() {
+        return superior_ids;
     }
 
-    public void setSuperiorIds(Integer[] superiorIds) {
-        this.superiorIds = superiorIds;
+    public void setSuperior_ids(Integer[] superior_ids) {
+        this.superior_ids = superior_ids;
     }
 
     @Override
     public String toString() {
         return "DosVsysId{" +
                 "id=" + id +
-                ", superiorIds=" + Arrays.toString(superiorIds) +
+                ", superior_ids=" + Arrays.toString(superior_ids) +
                 '}';
     }
 }
diff --git a/src/main/java/com/zdjizhi/common/CommonConfig.java b/src/main/java/com/zdjizhi/common/FlowWriteConfig.java
index 9bb50c6..a93496f 100644
--- a/src/main/java/com/zdjizhi/common/CommonConfig.java
+++ b/src/main/java/com/zdjizhi/common/FlowWriteConfig.java
@@ -7,12 +7,11 @@ import org.jasypt.encryption.pbe.StandardPBEStringEncryptor;
  * @author wlh
  * @date 2021/1/6
  */
-public class CommonConfig {
+public class FlowWriteConfig {
 
     /**
      * 定位库默认分隔符
      */
-    public static final String LOCATION_SEPARATOR = ".";
 
     private static StandardPBEStringEncryptor encryptor = new StandardPBEStringEncryptor();
 
@@ -51,7 +50,6 @@ public class CommonConfig {
     public static final int DESTINATION_IP_PARTITION_NUM = CommonConfigurations.getIntProperty("destination.ip.partition.num");
     public static final int DATA_CENTER_ID_NUM = CommonConfigurations.getIntProperty("data.center.id.num");
 
-    public static final String IP_MMDB_PATH = CommonConfigurations.getStringProperty("ip.mmdb.path");
 
     public static final String BIFANG_SERVER_URI = CommonConfigurations.getStringProperty("bifang.server.uri");
     public static final String BIFANG_SERVER_ENCRYPTPWD_PATH = CommonConfigurations.getStringProperty("bifang.server.encryptpwd.path");
@@ -74,37 +72,20 @@ public class CommonConfig {
 
     public static final int SASL_JAAS_CONFIG_FLAG = CommonConfigurations.getIntProperty("sasl.jaas.config.flag");
 
-    public static final String NACOS_SERVER_ADDR = CommonConfigurations.getStringProperty("nacos.server.addr");
+    public static final String NACOS_SERVER = CommonConfigurations.getStringProperty("nacos.server.addr");
     public static final String NACOS_USERNAME = CommonConfigurations.getStringProperty("nacos.username");
-    public static final String NACOS_PASSWORD = CommonConfigurations.getStringProperty("nacos.password");
-    public static final String NACOS_NAMESPACE = CommonConfigurations.getStringProperty("nacos.namespace");
-    public static final String NACOS_DATA_ID = CommonConfigurations.getStringProperty("nacos.data.id");
-    public static final String NACOS_GROUP = CommonConfigurations.getStringProperty("nacos.group");
-    public static final int NACOS_READ_TIMEOUT = CommonConfigurations.getIntProperty("nacos.read.timeout");
+    public static final String NACOS_PIN = CommonConfigurations.getStringProperty("nacos.password");
+    public static final String NACOS_PUBLIC_NAMESPACE = CommonConfigurations.getStringProperty("nacos.namespace");
+    public static final String NACOS_KNOWLEDGEBASE_DATA_ID = CommonConfigurations.getStringProperty("nacos.data.id");
+    public static final String NACOS_PUBLIC_GROUP = CommonConfigurations.getStringProperty("nacos.group");
+    public static final Integer NACOS_CONNECTION_TIMEOUT = CommonConfigurations.getIntProperty("nacos.connection.timeout");
 
 
-    public static final String HOS_TOKEN = CommonConfigurations.getStringProperty("hos.token");
 
-    public static final String CLUSTER_OR_SINGLE = CommonConfigurations.getStringProperty("cluster.or.single");
+    public static final String NACOS_DOS_NAMESPACE = CommonConfigurations.getStringProperty("nacos.dos.namespace");
+    public static final String NACOS_DOS_DATA_ID = CommonConfigurations.getStringProperty("nacos.dos.data.id");
+    public static final String NACOS_DOS_GROUP = CommonConfigurations.getStringProperty("nacos.dos.group");
 
-    public static final String HDFS_URI_NS1 = CommonConfigurations.getStringProperty("hdfs.uri.nn1");
-    public static final String HDFS_URI_NS2 = CommonConfigurations.getStringProperty("hdfs.uri.nn2");
-    public static final String HDFS_PATH = CommonConfigurations.getStringProperty("hdfs.path");
-    public static final String HDFS_USER = CommonConfigurations.getStringProperty("hdfs.user");
+    public static final Integer HTTP_SOCKET_TIMEOUT = CommonConfigurations.getIntProperty("http.socket.timeout");
 
-    public static final String DOWNLOAD_PATH = CommonConfigurations.getStringProperty("download.path");
-
-    public static void main(String[] args) {
-        StandardPBEStringEncryptor encryptor = new StandardPBEStringEncryptor();
-        // 配置加密解密的密码/salt值
-        encryptor.setPassword("galaxy");
-        // 对"raw_password"进行加密：S5kR+Y7CI8k7MaecZpde25yK8NKUnd6p
-//        String password = "galaxy2019";
-        String password = "nacos";
-        String encPwd = encryptor.encrypt(password);
-        System.out.println(encPwd);
-        // 再进行解密：raw_password
-        String rawPwd = encryptor.decrypt(encPwd);
-        System.out.println(rawPwd);
-    }
 }
diff --git a/src/main/java/com/zdjizhi/common/KnowledgeLog.java b/src/main/java/com/zdjizhi/common/pojo/KnowlegeBaseMeta.java
index d72f7df..1701367 100644
--- a/src/main/java/com/zdjizhi/common/KnowledgeLog.java
+++ b/src/main/java/com/zdjizhi/common/pojo/KnowlegeBaseMeta.java
@@ -1,14 +1,24 @@
-package com.zdjizhi.common;
+package com.zdjizhi.common.pojo;
 
-public class KnowledgeLog {
-    public   String id;
-    public   String name;
-    public   String path;
-    public   Long size;
-    public   String format;
-    public   String sha256;
-    public   String version;
-    public   String updateTime;
+import java.io.Serializable;
+
+/**
+ *
+ */
+public class KnowlegeBaseMeta implements Serializable {
+    private String id;
+    private String name;
+    private String sha256;
+    private String format;
+    private String path;
+
+    public KnowlegeBaseMeta(String id, String name, String sha256, String format, String path) {
+        this.id = id;
+        this.name = name;
+        this.sha256 = sha256;
+        this.format = format;
+        this.path = path;
+    }
 
     public String getId() {
         return id;
@@ -26,30 +36,6 @@ public class KnowledgeLog {
         this.name = name;
     }
 
-    public String getPath() {
-        return path;
-    }
-
-    public void setPath(String path) {
-        this.path = path;
-    }
-
-    public Long getSize() {
-        return size;
-    }
-
-    public void setSize(Long size) {
-        this.size = size;
-    }
-
-    public String getFormat() {
-        return format;
-    }
-
-    public void setFormat(String format) {
-        this.format = format;
-    }
-
     public String getSha256() {
         return sha256;
     }
@@ -58,34 +44,31 @@ public class KnowledgeLog {
         this.sha256 = sha256;
     }
 
-    public String getVersion() {
-        return version;
+    public String getFormat() {
+        return format;
     }
 
-    public void setVersion(String version) {
-        this.version = version;
+    public void setFormat(String format) {
+        this.format = format;
     }
 
-    public String getUpdateTime() {
-        return updateTime;
+    public String getPath() {
+        return path;
     }
 
-    public void setUpdateTime(String updateTime) {
-        this.updateTime = updateTime;
+    public void setPath(String path) {
+        this.path = path;
     }
 
-
     @Override
     public String toString() {
-        return "KnowledgeLog{" +
+        return "KnowlegeBaseMeta{" +
                 "id='" + id + '\'' +
                 ", name='" + name + '\'' +
-                ", path='" + path + '\'' +
-                ", size=" + size +
-                ", format='" + format + '\'' +
                 ", sha256='" + sha256 + '\'' +
-                ", version='" + version + '\'' +
-                ", updateTime='" + updateTime + '\'' +
+                ", format='" + format + '\'' +
+                ", path='" + path + '\'' +
                 '}';
     }
 }
+
diff --git a/src/main/java/com/zdjizhi/etl/DosDetection.java b/src/main/java/com/zdjizhi/etl/DosDetection.java
index c8d68d1..87cb76a 100644
--- a/src/main/java/com/zdjizhi/etl/DosDetection.java
+++ b/src/main/java/com/zdjizhi/etl/DosDetection.java
@@ -1,17 +1,19 @@
 package com.zdjizhi.etl;
 
-import cn.hutool.core.math.MathUtil;
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
+import com.geedgenetworks.utils.DateUtils;
+import com.geedgenetworks.utils.StringUtil;
 import com.zdjizhi.common.*;
 import com.zdjizhi.utils.*;
+import com.zdjizhi.utils.connections.nacos.NacosUtils;
 import inet.ipaddr.IPAddress;
 import inet.ipaddr.IPAddressString;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.shaded.guava18.com.google.common.collect.TreeRangeMap;
-import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
 import org.apache.flink.util.Collector;
 
 import java.math.BigDecimal;
@@ -24,12 +26,12 @@ import java.util.concurrent.TimeUnit;
 /**
  * @author wlh
  */
-public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<String, String>, DosEventLog> {
+public class DosDetection extends ProcessFunction<DosSketchLog, DosEventLog> {
 
     private static final Log logger = LogFactory.get();
     private static Map<String, Map<String, DosBaselineThreshold>> baselineMap = new HashMap<>();
     private final static NumberFormat PERCENT_INSTANCE = NumberFormat.getPercentInstance();
-    private HashMap<Integer,HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>>> thresholdRangeMap;
+    private HashMap<Integer, HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>>> thresholdRangeMap;
 
     private final static int BASELINE_SIZE = 144;
     private final static int STATIC_CONDITION_TYPE = 1;
@@ -42,22 +44,20 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
 
     private final static int OTHER_BASELINE_TYPE = 3;
 
+
+
+
     @Override
     public void open(Configuration parameters) {
         ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(2,
                 new BasicThreadFactory.Builder().namingPattern("Dos-Detection-%d").daemon(true).build());
         try {
-
             super.open(parameters);
-            logger.info("begin init");
-            IpUtils.loadIpLook();
-            logger.info("init over");
-
             executorService.scheduleAtFixedRate(() -> thresholdRangeMap = ParseStaticThreshold.createStaticThreshold(), 0,
-                    CommonConfig.STATIC_THRESHOLD_SCHEDULE_MINUTES, TimeUnit.MINUTES);
+                    FlowWriteConfig.STATIC_THRESHOLD_SCHEDULE_MINUTES, TimeUnit.MINUTES);
 
             executorService.scheduleAtFixedRate(() -> baselineMap = ParseBaselineThreshold.readFromHbase(), 0,
-                    CommonConfig.BASELINE_THRESHOLD_SCHEDULE_DAYS, TimeUnit.DAYS);
+                    FlowWriteConfig.BASELINE_THRESHOLD_SCHEDULE_DAYS, TimeUnit.DAYS);
         } catch (Exception e) {
             logger.error("定时器任务执行失败", e);
         }
@@ -65,7 +65,7 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
     }
 
     @Override
-    public void processElement(DosSketchLog value, ReadOnlyContext ctx, Collector<DosEventLog> out) {
+    public void processElement(DosSketchLog value, Context ctx, Collector<DosEventLog> out) throws Exception {
         DosEventLog finalResult = null;
         try {
             String destinationIp = value.getDestination_ip();
@@ -75,13 +75,13 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
             IPAddress destinationIpAddress = new IPAddressString(destinationIp).getAddress();
 
             DosDetectionThreshold threshold = null;
-            if (thresholdRangeMap.containsKey(vsysId)){
+            if (thresholdRangeMap.containsKey(vsysId)) {
                 threshold = thresholdRangeMap.get(vsysId).getOrDefault(attackType, TreeRangeMap.create()).get(destinationIpAddress);
             }
 
             logger.debug("当前判断IP：{}, 类型: {}", key, attackType);
             if (threshold == null && baselineMap.containsKey(key)) {
-                finalResult = getDosEventLogByBaseline(value,key);
+                finalResult = getDosEventLogByBaseline(value, key);
             } else if (threshold == null && !baselineMap.containsKey(key)) {
                 finalResult = getDosEventLogBySensitivityThreshold(value);
             } else if (threshold != null) {
@@ -94,71 +94,62 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
             logger.error("判定失败\n {} \n{}", value, e);
         }
 
-        if (finalResult != null){
+        if (finalResult != null) {
             out.collect(finalResult);
         }
     }
 
-    @Override
-    public void processBroadcastElement(Map<String, String> value, Context ctx, Collector<DosEventLog> out) throws Exception {
-        if (!value.isEmpty()){
-            IpUtils.updateIpLook(value);
-        }
-    }
 
     private DosEventLog getDosEventLogBySensitivityThreshold(DosSketchLog value) {
         long sketchSessions = value.getSketch_sessions();
         Integer staticSensitivityThreshold = NacosUtils.getIntProperty("static.sensitivity.threshold");
         long diff = sketchSessions - staticSensitivityThreshold;
-        return getDosEventLog(value, staticSensitivityThreshold, diff,0, SENSITIVITY_CONDITION_TYPE, SESSIONS_TAG);
+        return getDosEventLog(value, staticSensitivityThreshold, diff, 0, SENSITIVITY_CONDITION_TYPE, SESSIONS_TAG);
     }
 
-    private DosEventLog getDosEventLogByBaseline(DosSketchLog value,String key) {
+    private DosEventLog getDosEventLogByBaseline(DosSketchLog value, String key) {
         String attackType = value.getAttack_type();
         long sketchSessions = value.getSketch_sessions();
         DosBaselineThreshold dosBaselineThreshold = baselineMap.get(key).get(attackType);
         Integer base = getBaseValue(dosBaselineThreshold, value);
         long diff = sketchSessions - base;
-        return getDosEventLog(value, base, diff, 0,BASELINE_CONDITION_TYPE, SESSIONS_TAG);
+        return getDosEventLog(value, base, diff, 0, BASELINE_CONDITION_TYPE, SESSIONS_TAG);
     }
 
     private DosEventLog getDosEventLogByStaticThreshold(DosSketchLog value, DosDetectionThreshold threshold) throws CloneNotSupportedException {
-        long sessionBase = threshold.getSessionsPerSec();
-        long pktBase=threshold.getPacketsPerSec();
-        long bitBase=threshold.getBitsPerSec();
+        long sessionBase = threshold.getSessions_per_sec();
+        long pktBase = threshold.getPackets_per_sec();
+        long bitBase = threshold.getBits_per_sec();
 
         long diffSession = value.getSketch_sessions() - sessionBase;
         long diffPkt = value.getSketch_packets() - pktBase;
         long diffByte = value.getSketch_bytes() - bitBase;
 
-//        Double diffSessionPercent = getDiffPercent(diffSession, sessionBase)*100;
-//        Double diffPktPercent = getDiffPercent(diffPkt, pktBase)*100;
-//        Double diffBitPercent = getDiffPercent(diffByte, bitBase)*100;
-
-        double diffSessionPercent=0.0;
-        double diffPktPercent=0.0;
-        double diffBitPercent=0.0;
-        if (sessionBase != 0 && sessionBase > 0){
-            diffSessionPercent = getDiffPercent(diffSession, sessionBase)*100;
+        double diffSessionPercent = 0.0;
+        double diffPktPercent = 0.0;
+        double diffBitPercent = 0.0;
+        //todo 代码Review发现该部分存在bug，23.11版本做修复，需测试。
+        if (sessionBase > 0) {
+            diffSessionPercent = getDiffPercent(diffSession, sessionBase) * 100;
         }
-        else if (pktBase != 0 && pktBase > 0){
-            diffPktPercent = getDiffPercent(diffPkt, pktBase)*100;
+        if (pktBase > 0) {
+            diffPktPercent = getDiffPercent(diffPkt, pktBase) * 100;
         }
-        else if (bitBase != 0 && bitBase > 0){
-            diffBitPercent = getDiffPercent(diffByte, bitBase)*100;
+        if (bitBase > 0) {
+            diffBitPercent = getDiffPercent(diffByte, bitBase) * 100;
         }
 
         long profileId = 0;
-        DosEventLog result =null;
-
-        if (diffSessionPercent >= diffPktPercent && diffSessionPercent >= diffBitPercent){
-            profileId = threshold.getProfileId();
-            result= getDosEventLog(value, sessionBase, diffSession, profileId, STATIC_CONDITION_TYPE, SESSIONS_TAG);
-        }else if (diffPktPercent >= diffSessionPercent && diffPktPercent >= diffBitPercent){
-            profileId = threshold.getProfileId();
-            result = getDosEventLog(value, pktBase, diffPkt,profileId, STATIC_CONDITION_TYPE, PACKETS_TAG);
-        }else if (diffBitPercent >= diffPktPercent && diffBitPercent >= diffSessionPercent){
-            profileId = threshold.getProfileId();
+        DosEventLog result = null;
+
+        if (diffSessionPercent >= diffPktPercent && diffSessionPercent >= diffBitPercent) {
+            profileId = threshold.getProfile_id();
+            result = getDosEventLog(value, sessionBase, diffSession, profileId, STATIC_CONDITION_TYPE, SESSIONS_TAG);
+        } else if (diffPktPercent >= diffSessionPercent && diffPktPercent >= diffBitPercent) {
+            profileId = threshold.getProfile_id();
+            result = getDosEventLog(value, pktBase, diffPkt, profileId, STATIC_CONDITION_TYPE, PACKETS_TAG);
+        } else if (diffBitPercent >= diffPktPercent && diffBitPercent >= diffSessionPercent) {
+            profileId = threshold.getProfile_id();
             result = getDosEventLog(value, bitBase, diffByte, profileId, STATIC_CONDITION_TYPE, BITS_TAG);
         }
         return result;
@@ -175,15 +166,15 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
             if (severity != Severity.NORMAL) {
                 if (type == BASELINE_CONDITION_TYPE && percent < NacosUtils.getDoubleProperty("baseline.sensitivity.threshold")) {
                     logger.debug("当前server IP:{},类型:{},基线值{}百分比{}未超过基线敏感阈值,日志详情\n{}", destinationIp, attackType, base, percent, value);
-                }else if ((type == BASELINE_CONDITION_TYPE || type == SENSITIVITY_CONDITION_TYPE) && value.getSketch_sessions() < staticSensitivityThreshold){
-                    logger.debug("当前server IP:{},类型:{},基线值{}百分比{}未超过静态敏感阈值,日志详情\n{}",destinationIp, attackType, base, percent, value);
-                }else {
+                } else if ((type == BASELINE_CONDITION_TYPE || type == SENSITIVITY_CONDITION_TYPE) && value.getSketch_sessions() < staticSensitivityThreshold) {
+                    logger.debug("当前server IP:{},类型:{},基线值{}百分比{}未超过静态敏感阈值,日志详情\n{}", destinationIp, attackType, base, percent, value);
+                } else {
 //                    result = getResult(value, base, profileId, severity, percent+1, type, tag);
                     result = getResult(value, base, profileId, severity, percent, type, tag);
-                    if (type == SENSITIVITY_CONDITION_TYPE){
+                    if (type == SENSITIVITY_CONDITION_TYPE) {
                         result.setSeverity(Severity.MAJOR.severity);
                     }
-                    logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,基于{}:{}检测,日志详情\n {}", destinationIp,attackType,base,percent,type,tag,result);
+                    logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,基于{}:{}检测,日志详情\n {}", destinationIp, attackType, base, percent, type, tag, result);
                 }
             } else {
                 logger.debug("当前server IP:{} 未出现 {} 异常,日志详情 {}", destinationIp, attackType, value);
@@ -201,10 +192,10 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
         dosEventLog.setProfile_id(profileId);
         dosEventLog.setAttack_type(value.getAttack_type());
         dosEventLog.setSeverity(severity.severity);
-        dosEventLog.setConditions(getConditions(PERCENT_INSTANCE.format(percent), base, value.getSketch_sessions(), type, tag,dosEventLog));
+        dosEventLog.setConditions(getConditions(PERCENT_INSTANCE.format(percent), base, value.getSketch_sessions(), type, tag, dosEventLog));
 //        dosEventLog.setConditions(getConditions(percent, base, value.getSketch_sessions(), type, tag,dosEventLog));
         dosEventLog.setDestination_ip(value.getDestination_ip());
-        dosEventLog.setDestination_country(IpUtils.ipLookup.countryLookup(value.getDestination_ip()));
+        dosEventLog.setDestination_country(IpLookupUtils.getCountryLookup(value.getDestination_ip()));
         String ipList = value.getSource_ip();
         dosEventLog.setSource_ip_list(ipList);
         dosEventLog.setSource_country_list(getSourceCountryList(ipList));
@@ -228,7 +219,7 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
                         logger.debug("获取到当前IP: {},类型: {} baseline值为0,替换为P95观测值{}", value.getDestination_ip(), value.getAttack_type(), defaultVaule);
                         base = defaultVaule;
                     }
-                    if (sessionRateBaselineType == OTHER_BASELINE_TYPE && base < NacosUtils.getIntProperty("static.sensitivity.threshold")){
+                    if (sessionRateBaselineType == OTHER_BASELINE_TYPE && base < NacosUtils.getIntProperty("static.sensitivity.threshold")) {
                         base = NacosUtils.getIntProperty("static.sensitivity.threshold");
                     }
                 }
@@ -239,24 +230,24 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
         return base;
     }
 
-    private String getConditions(String percent, long base, long sessions, int type, String tag,DosEventLog dosEventLog) {
-        int condition =0;
-        if ("Minor".equals(dosEventLog.getSeverity())){
-            condition=50;
-        }else if ("Warning".equals(dosEventLog.getSeverity())){
-            condition=100;
-        }else if ("Major".equals(dosEventLog.getSeverity())){
-            condition=250;
-        }else if ("Severe".equals(dosEventLog.getSeverity())){
-            condition=500;
-        }else if ("Critical".equals(dosEventLog.getSeverity())){
-            condition =800;
+    private String getConditions(String percent, long base, long sessions, int type, String tag, DosEventLog dosEventLog) {
+        int condition = 0;
+        if ("Minor".equals(dosEventLog.getSeverity())) {
+            condition = 50;
+        } else if ("Warning".equals(dosEventLog.getSeverity())) {
+            condition = 100;
+        } else if ("Major".equals(dosEventLog.getSeverity())) {
+            condition = 250;
+        } else if ("Severe".equals(dosEventLog.getSeverity())) {
+            condition = 500;
+        } else if ("Critical".equals(dosEventLog.getSeverity())) {
+            condition = 800;
         }
         switch (type) {
             case STATIC_CONDITION_TYPE:
                 return "Rate > " +
                         base + " " +
-                        tag + "/s" + "(>"+condition+"%)";
+                        tag + "/s" + "(>" + condition + "%)";
             case BASELINE_CONDITION_TYPE:
                 return tag + " > " +
                         percent + " of baseline";
@@ -276,8 +267,8 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
                 String[] ipArr = sourceIpList.split(",");
                 HashSet<String> countrySet = new HashSet<>();
                 for (String ip : ipArr) {
-                    String country = IpUtils.ipLookup.countryLookup(ip);
-                    if (StringUtil.isNotBlank(country)){
+                    String country = IpLookupUtils.getCountryLookup(ip);
+                    if (StringUtil.isNotBlank(country)) {
                         countrySet.add(country);
                     }
                 }
@@ -304,19 +295,13 @@ public class DosDetection extends BroadcastProcessFunction<DosSketchLog,Map<Stri
         return index;
     }
 
-    public static void main(String[] args) {
-//        System.out.println(new DosDetection().getSourceCountryList("192.0.2.3,138.199.14.31,255.255.255.255,121.14.89.209," +
-//                "23.200.74.224,161.117.68.253"));
-//        DosDetection dosDetection = new DosDetection();
-//        System.out.println(dosDetection.judgeSeverity(dosDetection.getDiffPercent(499, 1000)));
 
-    }
 
     private Double getDiffPercent(long diff, long base) {
         try {
             return BigDecimal.valueOf((float) diff / base).setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue();
-        }catch (Exception e){
-            logger.info("当前阈值为0,进行下一阈值条件判断",e);
+        } catch (Exception e) {
+            logger.info("当前阈值为0,进行下一阈值条件判断", e);
             return 0.0;
         }
 
diff --git a/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java b/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
index bfd4ad8..7ffd1d8 100644
--- a/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/EtlProcessFunction.java
@@ -2,7 +2,7 @@ package com.zdjizhi.etl;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosSketchLog;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple3;
@@ -83,14 +83,14 @@ public class EtlProcessFunction extends ProcessWindowFunction<DosSketchLog, DosS
                     endTime = newSketchLog.getSketch_start_time() > endTime ? newSketchLog.getSketch_start_time() : endTime;
                     duration = endTime - startTime == 0 ? 5 : endTime - startTime;
                 }else {
-                    if (sourceIpSet.size() < CommonConfig.SOURCE_IP_LIST_LIMIT){
+                    if (sourceIpSet.size() < FlowWriteConfig.SOURCE_IP_LIST_LIMIT){
                         sourceIpSet.add(sourceIp);
                     }
                 }
             }
             String sourceIpList = StringUtils.join(sourceIpSet, ",");
-            return Tuple6.of(sessions/CommonConfig.FLINK_WINDOW_MAX_TIME,packets/CommonConfig.FLINK_WINDOW_MAX_TIME,
-                    bytes*8/CommonConfig.FLINK_WINDOW_MAX_TIME,sourceIpList,startTime,duration);
+            return Tuple6.of(sessions/ FlowWriteConfig.FLINK_WINDOW_MAX_TIME,packets/ FlowWriteConfig.FLINK_WINDOW_MAX_TIME,
+                    bytes*8/ FlowWriteConfig.FLINK_WINDOW_MAX_TIME,sourceIpList,startTime,duration);
         }catch (Exception e){
             logger.error("聚合中间结果集失败 {}",e);
         }
diff --git a/src/main/java/com/zdjizhi/etl/ParseBaselineThreshold.java b/src/main/java/com/zdjizhi/etl/ParseBaselineThreshold.java
index dc37bdf..8d5b9ca 100644
--- a/src/main/java/com/zdjizhi/etl/ParseBaselineThreshold.java
+++ b/src/main/java/com/zdjizhi/etl/ParseBaselineThreshold.java
@@ -2,9 +2,9 @@ package com.zdjizhi.etl;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.CommonConfig;
+import com.geedgenetworks.utils.DateUtils;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosBaselineThreshold;
-import com.zdjizhi.utils.DateUtils;
 import com.zdjizhi.utils.HbaseUtils;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HConstants;
@@ -17,7 +17,7 @@ import java.util.*;
 
 public class ParseBaselineThreshold {
 
-//    private static final Logger logger = LoggerFactory.getLogger(ParseBaselineThreshold.class);
+
     private static final Log logger = LogFactory.get();
     private static ArrayList<String> floodTypeList = new ArrayList<>();
 
@@ -34,23 +34,23 @@ public class ParseBaselineThreshold {
     private static void prepareHbaseEnv() throws IOException {
         org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
 
-        config.set("hbase.zookeeper.quorum", CommonConfig.HBASE_ZOOKEEPER_QUORUM);
+        config.set("hbase.zookeeper.quorum", FlowWriteConfig.HBASE_ZOOKEEPER_QUORUM);
         config.set("hbase.client.retries.number", "3");
         config.set("hbase.bulkload.retries.number", "3");
         config.set("zookeeper.recovery.retry", "3");
         config.set("hbase.defaults.for.version", "2.2.3");
         config.set("hbase.defaults.for.version.skip", "true");
-        config.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, CommonConfig.HBASE_CLIENT_OPERATION_TIMEOUT);
-        config.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, CommonConfig.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
+        config.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, FlowWriteConfig.HBASE_CLIENT_OPERATION_TIMEOUT);
+        config.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, FlowWriteConfig.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
 
-        TableName tableName = TableName.valueOf(CommonConfig.HBASE_BASELINE_TABLE_NAME);
+        TableName tableName = TableName.valueOf(FlowWriteConfig.HBASE_BASELINE_TABLE_NAME);
         Connection conn = ConnectionFactory.createConnection(config);
         table = conn.getTable(tableName);
         long currentTimeMillis = System.currentTimeMillis();
         scan = new Scan()
                 .setAllowPartialResults(true)
-                .setTimeRange(DateUtils.getSomeDate(new Date(currentTimeMillis), Math.negateExact(CommonConfig.HBASE_BASELINE_TTL)).getTime(), currentTimeMillis)
-                .setLimit(CommonConfig.HBASE_BASELINE_TOTAL_NUM);
+                .setTimeRange(DateUtils.getSomeDate(new Date(currentTimeMillis), Math.negateExact(FlowWriteConfig.HBASE_BASELINE_TTL)).getTime(), currentTimeMillis)
+                .setLimit(FlowWriteConfig.HBASE_BASELINE_TOTAL_NUM);
         logger.info("连接hbase成功，正在读取baseline数据");
     }
 
@@ -84,29 +84,4 @@ public class ParseBaselineThreshold {
         }
         return baselineMap;
     }
-
-    public static void main(String[] args) {
-        long currentTimeMillis = System.currentTimeMillis();
-        long p200D = DateUtils.getSomeDate(new Date(currentTimeMillis), Math.negateExact(CommonConfig.HBASE_BASELINE_TTL)).getTime();
-        System.out.println(p200D);
-        System.out.println(currentTimeMillis);
-        System.out.println(currentTimeMillis - p200D);
-
-
-        Map<String, Map<String, DosBaselineThreshold>> baselineMap = readFromHbase();
-        Set<String> keySet = baselineMap.keySet();
-        for (String key : keySet) {
-            Map<String, DosBaselineThreshold> stringTuple2Map = baselineMap.get(key);
-            Set<String> strings = stringTuple2Map.keySet();
-            for (String s:strings){
-                DosBaselineThreshold dosBaselineThreshold = stringTuple2Map.get(s);
-                System.out.println(key+"---"+s+"---"+dosBaselineThreshold);
-            }
-        }
-        System.out.println(baselineMap.size());
-    }
-
-
-
-
 }
diff --git a/src/main/java/com/zdjizhi/etl/ParseSketchLog.java b/src/main/java/com/zdjizhi/etl/ParseSketchLog.java
index 6731beb..5f31300 100644
--- a/src/main/java/com/zdjizhi/etl/ParseSketchLog.java
+++ b/src/main/java/com/zdjizhi/etl/ParseSketchLog.java
@@ -1,13 +1,11 @@
 package com.zdjizhi.etl;
 
 import com.alibaba.fastjson2.JSONObject;
-import com.fasterxml.jackson.databind.JavaType;
-import com.zdjizhi.common.CommonConfig;
+import com.geedgenetworks.utils.StringUtil;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosSketchLog;
 import com.zdjizhi.source.DosSketchSource;
 import com.zdjizhi.utils.FlinkEnvironmentUtils;
-//import com.zdjizhi.utils.JsonMapper;
-import com.zdjizhi.utils.StringUtil;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
@@ -24,22 +22,19 @@ import java.util.*;
 public class ParseSketchLog {
 
     private static Logger logger = LoggerFactory.getLogger(ParseSketchLog.class);
-//    private static JsonMapper jsonMapperInstance = JsonMapper.getInstance();
-//    private static JavaType hashmapJsonType = jsonMapperInstance.createCollectionType(HashMap.class, String.class, Object.class);
-//    private static JavaType listType = jsonMapperInstance.createCollectionType(ArrayList.class, HashMap.class);
 
 
-    public static SingleOutputStreamOperator<DosSketchLog> getSketchSource(){
+    public static SingleOutputStreamOperator<DosSketchLog> getSketchSource() {
         return flatSketchSource().assignTimestampsAndWatermarks(createWatermarkStrategy());
     }
 
-    private static SingleOutputStreamOperator<DosSketchLog> flatSketchSource(){
+    private static SingleOutputStreamOperator<DosSketchLog> flatSketchSource() {
         return DosSketchSource.createDosSketchSource().flatMap(new FlatSketchLog());
     }
 
-    private static WatermarkStrategy<DosSketchLog> createWatermarkStrategy(){
+    private static WatermarkStrategy<DosSketchLog> createWatermarkStrategy() {
         return WatermarkStrategy
-                .<DosSketchLog>forBoundedOutOfOrderness(Duration.ofSeconds(CommonConfig.FLINK_WATERMARK_MAX_ORDERNESS))
+                .<DosSketchLog>forBoundedOutOfOrderness(Duration.ofSeconds(FlowWriteConfig.FLINK_WATERMARK_MAX_ORDERNESS))
                 .withTimestampAssigner((event, timestamp) -> event.getSketch_start_time() * 1000);
     }
 
@@ -47,17 +42,15 @@ public class ParseSketchLog {
         @Override
         public void flatMap(String s, Collector<DosSketchLog> collector) {
             try {
-                if (StringUtil.isNotBlank(s)){
+                if (StringUtil.isNotBlank(s)) {
                     HashMap<String, Object> sketchSource = JSONObject.parseObject(s, HashMap.class);
-//                    HashMap<String, Object> sketchSource = jsonMapperInstance.fromJson(s, hashmapJsonType);
                     long sketchStartTime = Long.parseLong(sketchSource.get("sketch_start_time").toString());
                     long sketchDuration = Long.parseLong(sketchSource.get("sketch_duration").toString());
                     String attackType = sketchSource.get("attack_type").toString();
                     int vsysId = Integer.parseInt(sketchSource.getOrDefault("common_vsys_id", 1).toString());
                     String report_ip_list = JSONObject.toJSONString(sketchSource.get("report_ip_list"));
                     ArrayList<HashMap<String, Object>> reportIpList = JSONObject.parseObject(report_ip_list, ArrayList.class);
-//                    ArrayList<HashMap<String, Object>> reportIpList = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(sketchSource.get("report_ip_list")), listType);
-                    for (HashMap<String, Object> obj : reportIpList) {
+                   for (HashMap<String, Object> obj : reportIpList) {
                         DosSketchLog dosSketchLog = new DosSketchLog();
                         dosSketchLog.setSketch_start_time(sketchStartTime);
                         dosSketchLog.setSketch_duration(sketchDuration);
@@ -74,19 +67,12 @@ public class ParseSketchLog {
                         dosSketchLog.setSketch_packets(sketchPackets);
                         dosSketchLog.setSketch_bytes(sketchBytes);
                         collector.collect(dosSketchLog);
-                        logger.debug("数据解析成功：{}",dosSketchLog.toString());
+                        logger.debug("数据解析成功：{}", dosSketchLog.toString());
                     }
                 }
             } catch (Exception e) {
-                logger.error("数据解析错误：{} \n{}",s,e);
+                logger.error("数据解析错误：{} \n{}", s, e);
             }
         }
     }
-
-
-    public static void main(String[] args) throws Exception {
-        flatSketchSource().print();
-        FlinkEnvironmentUtils.streamExeEnv.execute();
-    }
-
 }
diff --git a/src/main/java/com/zdjizhi/etl/ParseStaticThreshold.java b/src/main/java/com/zdjizhi/etl/ParseStaticThreshold.java
index 5fd4fa3..1edf2b4 100644
--- a/src/main/java/com/zdjizhi/etl/ParseStaticThreshold.java
+++ b/src/main/java/com/zdjizhi/etl/ParseStaticThreshold.java
@@ -5,16 +5,18 @@ import cn.hutool.log.LogFactory;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
 //import com.fasterxml.jackson.databind.JavaType;
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosDetectionThreshold;
 import com.zdjizhi.common.DosVsysId;
 import com.zdjizhi.utils.HttpClientUtils;
 //import com.zdjizhi.utils.JsonMapper;
-import com.zdjizhi.utils.NacosUtils;
+
+import com.zdjizhi.utils.connections.nacos.NacosUtils;
 import inet.ipaddr.IPAddress;
 import inet.ipaddr.IPAddressString;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Range;
 import org.apache.flink.shaded.guava18.com.google.common.collect.TreeRangeMap;
+
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.message.BasicHeader;
 
@@ -28,15 +30,9 @@ import java.util.Map;
  * @author wlh
  */
 public class ParseStaticThreshold {
-//    private static Logger logger = LoggerFactory.getLogger(ParseStaticThreshold.class);
     private static final Log logger = LogFactory.get();
     private static String encryptpwd;
 
-//    private static JsonMapper jsonMapperInstance = JsonMapper.getInstance();
-//    private static JavaType hashmapJsonType = jsonMapperInstance.createCollectionType(HashMap.class, String.class, Object.class);
-//    private static JavaType thresholdType = jsonMapperInstance.createCollectionType(ArrayList.class, DosDetectionThreshold.class);
-//    private static JavaType vsysIDType = jsonMapperInstance.createCollectionType(ArrayList.class, DosVsysId.class);
-
     static {
         //加载加密登录密码
         encryptpwd = getEncryptpwd();
@@ -48,20 +44,18 @@ public class ParseStaticThreshold {
     private static String getEncryptpwd() {
         String psw = HttpClientUtils.ERROR_MESSAGE;
         try {
-            URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
+            URIBuilder uriBuilder = new URIBuilder(FlowWriteConfig.BIFANG_SERVER_URI);
             HashMap<String, Object> parms = new HashMap<>();
             parms.put("password", "admin");
-            HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_ENCRYPTPWD_PATH, parms);
+            HttpClientUtils.setUrlWithParams(uriBuilder, FlowWriteConfig.BIFANG_SERVER_ENCRYPTPWD_PATH, parms);
             String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build());
             if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
                 HashMap<String, Object> resposeMap = JSONObject.parseObject(resposeJsonStr, HashMap.class);
-//                HashMap<String, Object> resposeMap = jsonMapperInstance.fromJson(resposeJsonStr, hashmapJsonType);
                 boolean success = (boolean) resposeMap.get("success");
                 String msg = resposeMap.get("msg").toString();
 
                 if (success) {
                     HashMap<String, Object> data = JSONObject.parseObject(JSONObject.toJSONString(resposeMap.get("data")), HashMap.class);
-//                    HashMap<String, Object> data = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(resposeMap.get("data")), hashmapJsonType);
                     psw = data.get("encryptpwd").toString();
                 } else {
                     logger.error(msg);
@@ -75,40 +69,6 @@ public class ParseStaticThreshold {
         return psw;
     }
 
-    /**
-     * 登录bifang服务，获取token
-     *
-     * @return token
-     */
-    private static String loginBifangServer() {
-        String token = HttpClientUtils.ERROR_MESSAGE;
-        try {
-            if (!HttpClientUtils.ERROR_MESSAGE.equals(encryptpwd)) {
-                URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
-                HashMap<String, Object> parms = new HashMap<>();
-                parms.put("username", "admin");
-                parms.put("password", encryptpwd);
-                HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_LOGIN_PATH, parms);
-                String resposeJsonStr = HttpClientUtils.httpPost(uriBuilder.build(), null);
-                if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
-                    HashMap<String, Object> resposeMap = JSONObject.parseObject(resposeJsonStr, HashMap.class);
-//                    HashMap<String, Object> resposeMap = jsonMapperInstance.fromJson(resposeJsonStr, hashmapJsonType);
-                    boolean success = (boolean) resposeMap.get("success");
-                    String msg = resposeMap.get("msg").toString();
-                    if (success) {
-                        HashMap<String, Object> data = JSONObject.parseObject(JSONObject.toJSONString(resposeMap.get("data")), HashMap.class);
-//                        HashMap<String, Object> data = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(resposeMap.get("data")), hashmapJsonType);
-                        token = data.get("token").toString();
-                    } else {
-                        logger.error(msg);
-                    }
-                }
-            }
-        } catch (Exception e) {
-            logger.error("登录失败,未获取到token ", e);
-        }
-        return token;
-    }
 
     /**
      * 获取vsysId配置列表
@@ -118,12 +78,12 @@ public class ParseStaticThreshold {
     private static ArrayList<DosVsysId> getVsysId() {
         ArrayList<DosVsysId> vsysIdList = null;
         try {
-            URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
+            URIBuilder uriBuilder = new URIBuilder(FlowWriteConfig.BIFANG_SERVER_URI);
             HashMap<String, Object> parms = new HashMap<>();
-            parms.put("pageSize", -1);
+            parms.put("page_size", -1);
 //            parms.put("orderBy", "vsysId desc");
             parms.put("type", 1);
-            HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_POLICY_VSYSID_PATH, parms);
+            HttpClientUtils.setUrlWithParams(uriBuilder, FlowWriteConfig.BIFANG_SERVER_POLICY_VSYSID_PATH, parms);
             String token = NacosUtils.getStringProperty("bifang.server.token");
             if (!HttpClientUtils.ERROR_MESSAGE.equals(token)) {
                 BasicHeader authorization = new BasicHeader("Authorization", token);
@@ -131,21 +91,15 @@ public class ParseStaticThreshold {
                 String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization, authorization1);
                 if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
                     HashMap<String, Object> resposeMap = JSONObject.parseObject(resposeJsonStr, HashMap.class);
-//                    HashMap<String, Object> resposeMap = jsonMapperInstance.fromJson(resposeJsonStr, hashmapJsonType);
                     boolean success = (boolean) resposeMap.get("success");
                     String msg = resposeMap.get("msg").toString();
                     if (success) {
                         HashMap<String, Object> data = JSONObject.parseObject(JSONObject.toJSONString(resposeMap.get("data")), HashMap.class);
-//                        HashMap<String, Object> data = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(resposeMap.get("data")), hashmapJsonType);
-                        Object list = data.get("list");
+                       Object list = data.get("list");
                         if (list != null) {
-                            String s = JSONObject.toJSONString(list);
                             List<DosVsysId> dosVsysIds = JSON.parseArray(JSONObject.toJSONString(list), DosVsysId.class);
-//                            vsysIdList= JSONObject.parseObject(JSONObject.toJSONString(list), DosVsysId.class);
                             vsysIdList= (ArrayList)dosVsysIds;
-
-//                            vsysIdList = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(list), vsysIDType);
-                            logger.info("获取到vsysId {}条", vsysIdList.size());
+                          logger.info("获取到vsysId {}条", vsysIdList.size());
                         } else {
                             logger.warn("vsysIdList为空");
                         }
@@ -172,35 +126,37 @@ public class ParseStaticThreshold {
             if (vsysIds != null) {
                 for (DosVsysId dosVsysId : vsysIds) {
                     Integer vsysId = dosVsysId.getId() == null ? 1 : dosVsysId.getId();
-                    Integer[] superiorIds = dosVsysId.getSuperiorIds();
-                    URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
+                    Integer[] superiorIds = dosVsysId.getSuperior_ids();
+                    URIBuilder uriBuilder = new URIBuilder(FlowWriteConfig.BIFANG_SERVER_URI);
                     HashMap<String, Object> parms = new HashMap<>();
-                    parms.put("pageSize", -1);
-                    parms.put("orderBy", "profileId asc");
-                    parms.put("isValid", 1);
-                    parms.put("vsysId", vsysId);
-                    HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_POLICY_THRESHOLD_PATH, parms);
+                    parms.put("page_size", -1);
+                    parms.put("order_by", "profileId asc");
+                    parms.put("is_valid", 1);
+                    parms.put("vsys_id", vsysId);
+                    HttpClientUtils.setUrlWithParams(uriBuilder, FlowWriteConfig.BIFANG_SERVER_POLICY_THRESHOLD_PATH, parms);
                     String token = NacosUtils.getStringProperty("bifang.server.token");
                     if (!HttpClientUtils.ERROR_MESSAGE.equals(token)) {
                         BasicHeader authorization = new BasicHeader("Authorization", token);
                         BasicHeader authorization1 = new BasicHeader("Content-Type", "application/x-www-form-urlencoded");
                         String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization, authorization1);
                         if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
-//                            HashMap<String, Object> resposeMap = jsonMapperInstance.fromJson(resposeJsonStr, hashmapJsonType);
+
                             HashMap<String, Object> resposeMap = JSONObject.parseObject(resposeJsonStr,HashMap.class);
+
+
                             boolean success = (boolean) resposeMap.get("success");
                             String msg = resposeMap.get("msg").toString();
                             if (success) {
-//                                HashMap<String, Object> data = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(resposeMap.get("data")), hashmapJsonType);
-                                HashMap<String, Object> data = JSONObject.parseObject(JSONObject.toJSONString(resposeMap.get("data")), HashMap.class);
+                               HashMap<String, Object> data = JSONObject.parseObject(JSONObject.toJSONString(resposeMap.get("data")), HashMap.class);
                                 Object list = data.get("list");
+
                                 if (list != null) {
-//                                    ArrayList<DosDetectionThreshold> thresholds = jsonMapperInstance.fromJson(jsonMapperInstance.toJson(list), thresholdType);
-//                                    ArrayList<DosDetectionThreshold> thresholds = JSONObject.parseObject(JSONObject.toJSONString(list), ArrayList.class);
-                                    List<DosDetectionThreshold> dosDetectionThresholds = JSON.parseArray(JSONObject.toJSONString(list), DosDetectionThreshold.class);
+                                    System.out.println(list);
+                                   List<DosDetectionThreshold> dosDetectionThresholds = JSON.parseArray(JSONObject.toJSONString(list), DosDetectionThreshold.class);
+                                    System.out.println(dosDetectionThresholds);
                                     ArrayList<DosDetectionThreshold> thresholds = (ArrayList)dosDetectionThresholds;
                                     for (DosDetectionThreshold dosDetectionThreshold : thresholds) {
-                                        dosDetectionThreshold.setSuperiorIds(superiorIds);
+                                        dosDetectionThreshold.setSuperior_ids(superiorIds);
                                         vsysThresholds.add(dosDetectionThreshold);
                                     }
                                     logger.info("获取到vsys id是{}静态阈值配置{}条", vsysId, thresholds.size());
@@ -217,7 +173,6 @@ public class ParseStaticThreshold {
         } catch (Exception e) {
             logger.error("获取静态阈值配置失败,请检查bifang服务或登录配置信息 ", e);
         }
-
         return vsysThresholds;
     }
 
@@ -230,14 +185,19 @@ public class ParseStaticThreshold {
         HashMap<Integer, HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>>> thresholdRangeMap = new HashMap<>(4);
         try {
             ArrayList<DosDetectionThreshold> dosDetectionThreshold = getDosDetectionThreshold();
+
             if (dosDetectionThreshold != null && !dosDetectionThreshold.isEmpty()) {
                 for (DosDetectionThreshold threshold : dosDetectionThreshold) {
-                    String attackType = threshold.getAttackType();
-                    int vsysId = threshold.getVsysId();
+
+                    String attackType = threshold.getAttack_type();
+                    int vsysId = threshold.getVsys_id();
                     HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>> rangeMap = thresholdRangeMap.getOrDefault(vsysId, new HashMap<>());
 
                     TreeRangeMap<IPAddress, DosDetectionThreshold> treeRangeMap = rangeMap.getOrDefault(attackType, TreeRangeMap.create());
-                    ArrayList<String> serverIpList = threshold.getServerIpList();
+
+
+                    ArrayList<String> serverIpList = threshold.getServer_ip_list();
+
                     for (String sip : serverIpList) {
                         IPAddressString ipAddressString = new IPAddressString(sip);
                         if (ipAddressString.isIPAddress()) {
@@ -279,33 +239,38 @@ public class ParseStaticThreshold {
         return thresholdRangeMap;
     }
 
-    public static void main(String[] args) {
-        /*
-        ArrayList<DosDetectionThreshold> dosDetectionThreshold = getDosDetectionThreshold();
-//        dosDetectionThreshold.forEach(System.out::println);
-        getVsysId().forEach(System.out::println);
-        System.out.println("------------------------");
-        */
-        HashMap<Integer, HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>>> staticThreshold = createStaticThreshold();
 
-        System.out.println("------------------------");
 
-        for (Integer integer : staticThreshold.keySet()) {
-            HashMap<String, TreeRangeMap<IPAddress, DosDetectionThreshold>> stringTreeRangeMapHashMap = staticThreshold.get(integer);
-            for (String type : stringTreeRangeMapHashMap.keySet()) {
-                Map<Range<IPAddress>, DosDetectionThreshold> asMapOfRanges = stringTreeRangeMapHashMap.get(type).asMapOfRanges();
-                for (Range<IPAddress> range : asMapOfRanges.keySet()) {
-                    DosDetectionThreshold threshold = asMapOfRanges.get(range);
-                    System.out.println(integer + "---" + type + "---" + range + "---" + threshold);
+    /**
+     * 登录bifang服务，获取token
+     *
+     * @return token
+     */
+    private static String loginBifangServer() {
+        String token = HttpClientUtils.ERROR_MESSAGE;
+        try {
+            if (!HttpClientUtils.ERROR_MESSAGE.equals(encryptpwd)) {
+                URIBuilder uriBuilder = new URIBuilder(FlowWriteConfig.BIFANG_SERVER_URI);
+                HashMap<String, Object> parms = new HashMap<>();
+                parms.put("username", "admin");
+                parms.put("password", encryptpwd);
+                HttpClientUtils.setUrlWithParams(uriBuilder, FlowWriteConfig.BIFANG_SERVER_LOGIN_PATH, parms);
+                String resposeJsonStr = HttpClientUtils.httpPost(uriBuilder.build(), null);
+                if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
+                    HashMap<String, Object> resposeMap = JSONObject.parseObject(resposeJsonStr, HashMap.class);
+                    boolean success = (boolean) resposeMap.get("success");
+                    String msg = resposeMap.get("msg").toString();
+                    if (success) {
+                        HashMap<String, Object> data = JSONObject.parseObject(JSONObject.toJSONString(resposeMap.get("data")), HashMap.class);
+                        token = data.get("token").toString();
+                    } else {
+                        logger.error(msg);
+                    }
                 }
-                System.out.println("------------------------");
             }
-
+        } catch (Exception e) {
+            logger.error("登录失败,未获取到token ", e);
         }
-//        String s = loginBifangServer();
-//        System.out.println(s);
-
+        return token;
     }
-
-
 }
diff --git a/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java b/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java
index 8206cff..a4e5bdd 100644
--- a/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java
+++ b/src/main/java/com/zdjizhi/etl/TrafficServerIpMetrics.java
@@ -2,7 +2,7 @@ package com.zdjizhi.etl;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosMetricsLog;
 import com.zdjizhi.common.DosSketchLog;
 
@@ -26,16 +26,11 @@ class TrafficServerIpMetrics {
     }
 
     private static long timeFloor(long sketchStartTime){
-        return sketchStartTime / CommonConfig.FLINK_WINDOW_MAX_TIME * CommonConfig.FLINK_WINDOW_MAX_TIME;
+        return sketchStartTime / FlowWriteConfig.FLINK_WINDOW_MAX_TIME * FlowWriteConfig.FLINK_WINDOW_MAX_TIME;
     }
 
     private static int getPartitionNumByIp(String destinationIp){
-        return Math.abs(destinationIp.hashCode()) % CommonConfig.DESTINATION_IP_PARTITION_NUM;
-    }
-
-    public static void main(String[] args) {
-        System.out.println(getPartitionNumByIp("146.177.223.43"));
-        System.out.println("146.177.223.43".hashCode());
+        return Math.abs(destinationIp.hashCode()) % FlowWriteConfig.DESTINATION_IP_PARTITION_NUM;
     }
 
 }
diff --git a/src/main/java/com/zdjizhi/sink/DosEventSink.java b/src/main/java/com/zdjizhi/sink/DosEventSink.java
index 18694ba..87795e6 100644
--- a/src/main/java/com/zdjizhi/sink/DosEventSink.java
+++ b/src/main/java/com/zdjizhi/sink/DosEventSink.java
@@ -1,7 +1,7 @@
 package com.zdjizhi.sink;
 
 import com.alibaba.fastjson2.JSONObject;
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosEventLog;
 //import com.zdjizhi.utils.JsonMapper;
 import com.zdjizhi.utils.KafkaUtils;
@@ -16,8 +16,8 @@ class DosEventSink {
                 .filter(Objects::nonNull)
 //                .map(JsonMapper::toJsonString)
                 .map(JSONObject::toJSONString)
-                .addSink(KafkaUtils.getKafkaSink(CommonConfig.KAFKA_OUTPUT_EVENT_TOPIC_NAME))
-                .setParallelism(CommonConfig.KAFKA_OUTPUT_EVENT_PARALLELISM);
+                .addSink(KafkaUtils.getKafkaSink(FlowWriteConfig.KAFKA_OUTPUT_EVENT_TOPIC_NAME))
+                .setParallelism(FlowWriteConfig.KAFKA_OUTPUT_EVENT_PARALLELISM);
     }
 
 }
diff --git a/src/main/java/com/zdjizhi/sink/OutputStreamSink.java b/src/main/java/com/zdjizhi/sink/OutputStreamSink.java
index f6ebeb0..35db54f 100644
--- a/src/main/java/com/zdjizhi/sink/OutputStreamSink.java
+++ b/src/main/java/com/zdjizhi/sink/OutputStreamSink.java
@@ -3,18 +3,14 @@ package com.zdjizhi.sink;
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
 import com.alibaba.nacos.api.PropertyKeyConst;
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosEventLog;
 import com.zdjizhi.common.DosMetricsLog;
 import com.zdjizhi.common.DosSketchLog;
 import com.zdjizhi.etl.DosDetection;
 import com.zdjizhi.etl.EtlProcessFunction;
 import com.zdjizhi.etl.ParseSketchLog;
-import com.zdjizhi.source.DosSketchSource;
 import com.zdjizhi.utils.FlinkEnvironmentUtils;
-import org.apache.flink.api.common.state.MapStateDescriptor;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.streaming.api.datastream.*;
@@ -22,7 +18,6 @@ import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindo
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.util.OutputTag;
 
-import java.util.Map;
 import java.util.Properties;
 
 /**
@@ -39,44 +34,24 @@ public class OutputStreamSink {
             SingleOutputStreamOperator<DosSketchLog> middleStream = getMiddleStream();
             DosEventSink.dosEventOutputSink(getEventSinkStream(middleStream));
             TrafficServerIpMetricsSink.sideOutputMetricsSink(middleStream);
-            FlinkEnvironmentUtils.streamExeEnv.execute(CommonConfig.STREAM_EXECUTION_JOB_NAME);
+            FlinkEnvironmentUtils.streamExeEnv.execute(FlowWriteConfig.STREAM_EXECUTION_JOB_NAME);
         } catch (Exception e) {
             logger.error("任务启动失败 {}",e);
         }
     }
 
     private static SingleOutputStreamOperator<DosEventLog> getEventSinkStream(SingleOutputStreamOperator<DosSketchLog> middleStream){
-        DataStreamSource<Map<String, String>> broadcastSource=null;
-        Properties nacosProperties = new Properties();
-
-        nacosProperties.put(PropertyKeyConst.SERVER_ADDR,CommonConfig.NACOS_SERVER_ADDR);
-        nacosProperties.setProperty(PropertyKeyConst.USERNAME, CommonConfig.NACOS_USERNAME);
-        nacosProperties.setProperty(PropertyKeyConst.PASSWORD, CommonConfig.NACOS_PASSWORD);
-        nacosProperties.setProperty(PropertyKeyConst.NAMESPACE, CommonConfig.NACOS_NAMESPACE);
-
-        if ("CLUSTER".equals(CommonConfig.CLUSTER_OR_SINGLE)){
-            broadcastSource = DosSketchSource.broadcastSource(nacosProperties);
-        }else {
-            broadcastSource= DosSketchSource.singleBroadcastSource(nacosProperties);
-        }
-
-        MapStateDescriptor<String,Map> descriptor =
-                new MapStateDescriptor<>("descriptorTest", Types.STRING, TypeInformation.of(Map.class));
-
-        BroadcastStream<Map<String, String>> broadcast = broadcastSource.broadcast(descriptor);
-
         return middleStream
-                .connect(broadcast)
-                .process(new DosDetection()).setParallelism(CommonConfig.FLINK_DETECTION_MAP_PARALLELISM);
+                .process(new DosDetection()).setParallelism(FlowWriteConfig.FLINK_DETECTION_MAP_PARALLELISM);
 
     }
 
     private static SingleOutputStreamOperator<DosSketchLog> getMiddleStream(){
         return ParseSketchLog.getSketchSource()
                 .keyBy(new KeysSelector())
-                .window(TumblingEventTimeWindows.of(Time.seconds(CommonConfig.FLINK_WINDOW_MAX_TIME)))
+                .window(TumblingEventTimeWindows.of(Time.seconds(FlowWriteConfig.FLINK_WINDOW_MAX_TIME)))
                 .process(new EtlProcessFunction())
-                .setParallelism(CommonConfig.FLINK_FIRST_AGG_PARALLELISM);
+                .setParallelism(FlowWriteConfig.FLINK_FIRST_AGG_PARALLELISM);
     }
 
     private static class KeysSelector implements KeySelector<DosSketchLog, Tuple3<String, String, Integer>>{
diff --git a/src/main/java/com/zdjizhi/sink/TrafficServerIpMetricsSink.java b/src/main/java/com/zdjizhi/sink/TrafficServerIpMetricsSink.java
index 7c5faa4..0025544 100644
--- a/src/main/java/com/zdjizhi/sink/TrafficServerIpMetricsSink.java
+++ b/src/main/java/com/zdjizhi/sink/TrafficServerIpMetricsSink.java
@@ -1,10 +1,10 @@
 package com.zdjizhi.sink;
 
 import com.alibaba.fastjson2.JSONObject;
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.common.DosMetricsLog;
 import com.zdjizhi.common.DosSketchLog;
-import com.zdjizhi.utils.JsonMapper;
+
 import com.zdjizhi.utils.KafkaUtils;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
@@ -16,10 +16,8 @@ class TrafficServerIpMetricsSink {
     static void sideOutputMetricsSink(SingleOutputStreamOperator<DosSketchLog> outputStream){
         DataStream<DosMetricsLog> sideOutput = outputStream.getSideOutput(outputTag);
 //        sideOutput.map(JsonMapper::toJsonString).addSink(KafkaUtils.getKafkaSink(CommonConfig.KAFKA_OUTPUT_METRIC_TOPIC_NAME))
-        sideOutput.map(JSONObject::toJSONString).addSink(KafkaUtils.getKafkaSink(CommonConfig.KAFKA_OUTPUT_METRIC_TOPIC_NAME))
-                .setParallelism(CommonConfig.KAFKA_OUTPUT_METRIC_PARALLELISM);
-
-
+        sideOutput.map(JSONObject::toJSONString).addSink(KafkaUtils.getKafkaSink(FlowWriteConfig.KAFKA_OUTPUT_METRIC_TOPIC_NAME))
+                .setParallelism(FlowWriteConfig.KAFKA_OUTPUT_METRIC_PARALLELISM);
 
 
     }
diff --git a/src/main/java/com/zdjizhi/source/DosSketchSource.java b/src/main/java/com/zdjizhi/source/DosSketchSource.java
index e1e73b3..238e350 100644
--- a/src/main/java/com/zdjizhi/source/DosSketchSource.java
+++ b/src/main/java/com/zdjizhi/source/DosSketchSource.java
@@ -1,15 +1,14 @@
 package com.zdjizhi.source;
 
-import com.zdjizhi.common.CommonConfig;
-import com.zdjizhi.common.CustomFile;
+
+
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.utils.FlinkEnvironmentUtils;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
 
-import java.util.List;
-import java.util.Map;
 import java.util.Properties;
 
 /**
@@ -21,26 +20,18 @@ public class DosSketchSource {
 
     public static DataStreamSource<String> createDosSketchSource(){
         Properties properties = new Properties();
-        properties.setProperty("bootstrap.servers", CommonConfig.KAFKA_INPUT_BOOTSTRAP_SERVERS);
-        properties.setProperty("group.id", CommonConfig.KAFKA_GROUP_ID);
-        if (CommonConfig.SASL_JAAS_CONFIG_FLAG == 1){
+        properties.setProperty("bootstrap.servers", FlowWriteConfig.KAFKA_INPUT_BOOTSTRAP_SERVERS);
+        properties.setProperty("group.id", FlowWriteConfig.KAFKA_GROUP_ID);
+        if (FlowWriteConfig.SASL_JAAS_CONFIG_FLAG == 1){
             properties.put("security.protocol", "SASL_PLAINTEXT");
             properties.put("sasl.mechanism", "PLAIN");
-            properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""+CommonConfig.SASL_JAAS_CONFIG_USER+"\" password=\""+CommonConfig.SASL_JAAS_CONFIG_PASSWORD+"\";");
+            properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""+ FlowWriteConfig.SASL_JAAS_CONFIG_USER+"\" password=\""+ FlowWriteConfig.SASL_JAAS_CONFIG_PASSWORD+"\";");
         }
 
         return streamExeEnv.addSource(new FlinkKafkaConsumer<String>(
-                CommonConfig.KAFKA_INPUT_TOPIC_NAME,
+                FlowWriteConfig.KAFKA_INPUT_TOPIC_NAME,
                 new SimpleStringSchema(), properties))
-                .setParallelism(CommonConfig.KAFKA_INPUT_PARALLELISM);
-    }
-
-
-    public static DataStreamSource<Map<String, String>> broadcastSource(Properties nacosProperties){
-        return streamExeEnv.addSource(new HttpSource(nacosProperties, CommonConfig.NACOS_DATA_ID, CommonConfig.NACOS_GROUP, CommonConfig.NACOS_READ_TIMEOUT));
+                .setParallelism(FlowWriteConfig.KAFKA_INPUT_PARALLELISM);
     }
 
-    public static DataStreamSource<Map<String, String>>singleBroadcastSource(Properties nacosProperties){
-        return streamExeEnv.addSource(new SingleHttpSource(nacosProperties, CommonConfig.NACOS_DATA_ID, CommonConfig.NACOS_GROUP, CommonConfig.NACOS_READ_TIMEOUT));
-    }
 }
diff --git a/src/main/java/com/zdjizhi/source/HttpSource.java b/src/main/java/com/zdjizhi/source/HttpSource.java
deleted file mode 100644
index 2e465a7..0000000
--- a/src/main/java/com/zdjizhi/source/HttpSource.java
+++ /dev/null
@@ -1,260 +0,0 @@
-package com.zdjizhi.source;
-
-import cn.hutool.core.io.FileUtil;
-import cn.hutool.core.io.IoUtil;
-import cn.hutool.core.io.file.FileReader;
-import cn.hutool.crypto.digest.DigestUtil;
-import cn.hutool.json.JSONObject;
-import com.alibaba.nacos.api.NacosFactory;
-import com.alibaba.nacos.api.PropertyKeyConst;
-import com.alibaba.nacos.api.config.ConfigService;
-import com.alibaba.nacos.api.config.listener.Listener;
-import com.alibaba.nacos.api.exception.NacosException;
-import com.fasterxml.jackson.databind.JavaType;
-import com.google.common.base.Joiner;
-import com.jayway.jsonpath.JsonPath;
-import com.zdjizhi.common.CommonConfig;
-import com.zdjizhi.common.CustomFile;
-import com.zdjizhi.common.KnowledgeLog;
-import com.zdjizhi.utils.*;
-import org.apache.commons.io.IOUtils;
-import org.apache.flink.configuration.Configuration;
-import org.apache.http.Header;
-import org.apache.http.message.BasicHeader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.*;
-import java.text.SimpleDateFormat;
-import java.util.*;
-import java.util.concurrent.Executor;
-
-
-public class HttpSource extends RichHttpSourceFunction<Map<String, String>> {
-    private static final Logger logger = LoggerFactory.getLogger(HttpSource.class);
-
-    private static final String EXPR = "$.[?(@.version=='latest' && @.name in ['ip_v4_built_in','ip_v6_built_in','ip_v4_user_defined','ip_v6_user_defined'])].['name','sha256','format','path']";
-
-    private static Map<String, String> knowledgeMetaCache = new HashMap<>();
-
-    private static HashMap<String, String> knowledgeUpdateCache;
-
-    private static final int TRY_TIMES = 3;
-
-    private static HttpClientUtils2 httpClientUtils;
-
-    //连接nacos的配置
-    private Properties nacosProperties;
-
-    //nacos data id
-    private String NACOS_DATA_ID;
-
-    //nacos group
-    private String NACOS_GROUP;
-
-    //nacos 连接超时时间
-    private long NACOS_READ_TIMEOUT;
-
-    private ConfigService configService;
-
-    private static Header header;
-
-    //运行状态cancel时置为false
-    private boolean isRunning = true;
-    //是否下发，默认不发送
-    private boolean isSending = false;
-
-
-    public HttpSource(Properties nacosProperties, String NACOS_DATA_ID, String NACOS_GROUP, long NACOS_READ_TIMEOUT) {
-        this.nacosProperties = nacosProperties;
-        this.NACOS_DATA_ID = NACOS_DATA_ID;
-        this.NACOS_GROUP = NACOS_GROUP;
-        this.NACOS_READ_TIMEOUT = NACOS_READ_TIMEOUT;
-    }
-
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        httpClientUtils = new HttpClientUtils2();
-        //初始化元数据缓存
-        knowledgeMetaCache = new HashMap<>(16);
-        //初始化定位库缓存
-        knowledgeUpdateCache = new HashMap<>(16);
-
-        header = new BasicHeader("token", CommonConfig.HOS_TOKEN);
-        //连接nacos配置
-        try {
-            configService = NacosFactory.createConfigService(nacosProperties);
-        }catch (NacosException e){
-            logger.error("Get Schema config from Nacos error,The exception message is :{}", e.getMessage());
-        }
-
-        //初始化知识库
-        initKnowledge();
-        logger.info("连接nacos：" + nacosProperties.getProperty(PropertyKeyConst.SERVER_ADDR));
-
-
-    }
-    @Override
-    public void run(SourceContext ctx) throws Exception {
-        if (!knowledgeUpdateCache.isEmpty()){
-            ctx.collect(knowledgeUpdateCache);
-                knowledgeUpdateCache.clear();
-        }
-//        }
-
-
-        configService.addListener(NACOS_DATA_ID, NACOS_GROUP, new Listener() {
-            @Override
-            public Executor getExecutor() {
-                return null;
-            }
-
-            @Override
-            public void receiveConfigInfo(String configMsg) {
-                try {
-                    logger.info("receive update config:" + configMsg);
-                    if (StringUtil.isNotBlank(configMsg)) {
-                        ArrayList<Object> metaList = JsonPath.parse(configMsg).read(EXPR);
-                        if (metaList.size() > 0) {
-                            for (Object metadata : metaList) {
-                                JSONObject knowledgeJson = new JSONObject(metadata, false, true);
-                                String fileName = Joiner.on(CommonConfig.LOCATION_SEPARATOR).useForNull("").join(knowledgeJson.getStr("name"),
-                                        knowledgeJson.getStr("format"));
-                                String sha256 = knowledgeJson.getStr("sha256");
-                                String filePath = knowledgeJson.getStr("path");
-                                if (!sha256.equals(knowledgeMetaCache.get(fileName))) {
-                                    knowledgeMetaCache.put(fileName, sha256);
-                                    updateKnowledge(fileName, filePath,sha256);
-                                }
-                            }
-                            if (!knowledgeUpdateCache.isEmpty()){
-                                ctx.collect(knowledgeUpdateCache);
-                                knowledgeUpdateCache.clear();
-                            }
-                        }
-                    }
-
-                } catch (Exception e) {
-                    logger.error("监听nacos配置失败", e);
-                }
-            }
-        });
-
-        while (isRunning) {
-            try {
-                Thread.sleep(10000);
-            }catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        }
-
-    }
-
-    private void initKnowledge(){
-        String configMsg = "";
-        try {
-            configMsg=configService.getConfig(NACOS_DATA_ID, NACOS_GROUP, NACOS_READ_TIMEOUT);
-        } catch (NacosException e) {
-            logger.error("从Nacos获取知识库元数据配置文件异常，异常信息为:{}", e.getMessage());
-        }
-
-        if (StringUtil.isNotBlank(configMsg)){
-            ArrayList<Object> metaList = JsonPath.parse(configMsg).read(EXPR);
-            if (metaList.size() > 0) {
-                for (Object metadata : metaList) {
-                    JSONObject knowledgeJson = new JSONObject(metadata, false, true);
-                    String fileName = Joiner.on(CommonConfig.LOCATION_SEPARATOR).useForNull("").join(knowledgeJson.getStr("name"),
-                            knowledgeJson.getStr("format"));
-                    String sha256 = knowledgeJson.getStr("sha256");
-                    String filePath = knowledgeJson.getStr("path");
-                    byte[] localFileByte = getLocalFile(fileName);
-                    String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                    if (sha256.equals(localFileSha256Hex)){
-                        logger.info("本地文件{}的sha256为:{} ,Nacos内记录为:{} ,sha256相等", fileName, localFileSha256Hex, sha256);
-                        knowledgeMetaCache.put(fileName, sha256);
-                    }else {
-                        logger.info("本地文件{}的sha256为:{} ,Nacos内记录为:{} ,sha256不相等，更新本地文件及缓存", fileName, localFileSha256Hex, sha256);
-                        updateKnowledge(fileName,filePath,sha256);
-                    }
-
-                }
-            }
-        }
-    }
-
-    private void updateKnowledge(String fileName, String filePath,String sha256) {
-        InputStream inputStream = null;
-        int retryNum = 0;
-        try {
-            while (retryNum < TRY_TIMES){
-                inputStream = httpClientUtils.httpGetInputStream(filePath, 90000, header);
-                if (inputStream !=null){
-                    byte[] downloadBytes = IOUtils.toByteArray(inputStream);
-                    String downloadFileSha256Hex = DigestUtil.sha256Hex(downloadBytes);
-                    if (sha256.equals(downloadFileSha256Hex)&& downloadBytes.length > 0 ){
-                        logger.info("通过HOS下载{}的sha256为:{} ,Nacos内记录为:{} ,sha256相等", fileName, sha256);
-                        boolean updateStatus = updateLocalFile(fileName, downloadBytes);
-                        if (updateStatus){
-                            knowledgeMetaCache.put(fileName,sha256);
-                            knowledgeUpdateCache.put(fileName, sha256);
-                            retryNum = TRY_TIMES;
-                        }else {
-                            retryNum++;
-                            //避免频繁请求HOS
-                            Thread.sleep(10000);
-                        }
-                    }else {
-                        logger.error("通过HOS下载{}的sha256为:{} ,Nacos内记录为:{} ,sha256不相等 开始第{}次重试下载文件", fileName, downloadFileSha256Hex, sha256, retryNum);
-                        retryNum++;
-                        //避免频繁请求HOS
-                        Thread.sleep(10000);
-                    }
-                }
-            }
-        } catch (IOException ioException) {
-            ioException.printStackTrace();
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        } finally {
-            IOUtils.closeQuietly(inputStream);
-        }
-    }
-
-
-    private boolean updateLocalFile(String fileName,byte[] downloadBytes) {
-        FileOutputStream outputStream = null;
-        boolean updateStatus = false;
-        try {
-            HdfsUtils.uploadFileByBytes(CommonConfig.HDFS_PATH + fileName, downloadBytes);
-            updateStatus=true;
-        } catch (IOException ioe) {
-            logger.error("更新本地文件{}时发生IO异常,异常信息为:", fileName, ioe.getMessage());
-            ioe.printStackTrace();
-        } catch (RuntimeException e) {
-            logger.error("更新本地文件{}时发生异常,异常信息为:", fileName, e.getMessage());
-            e.printStackTrace();
-        } finally {
-            IOUtils.closeQuietly(outputStream);
-        }
-        return updateStatus;
-    }
-
-    private static byte[] getLocalFile(String name) {
-        byte[] fileBytes = null;
-        try {
-            fileBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + name) ;
-        } catch (RuntimeException | IOException e) {
-            logger.error("IpLookupUtils download MMDB files error, message is:" + e.getMessage());
-            e.printStackTrace();
-        }
-        return fileBytes;
-    }
-
-    @Override
-    public void cancel() {
-        this.isRunning = false;
-    }
-
-
-}
diff --git a/src/main/java/com/zdjizhi/source/HttpSourceFunction.java b/src/main/java/com/zdjizhi/source/HttpSourceFunction.java
deleted file mode 100644
index 8fd58a9..0000000
--- a/src/main/java/com/zdjizhi/source/HttpSourceFunction.java
+++ /dev/null
@@ -1,6 +0,0 @@
-package com.zdjizhi.source;
-
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-
-public interface HttpSourceFunction<OUT> extends SourceFunction<OUT> {
-}
\ No newline at end of file
diff --git a/src/main/java/com/zdjizhi/source/RichHttpSourceFunction.java b/src/main/java/com/zdjizhi/source/RichHttpSourceFunction.java
deleted file mode 100644
index 582aa13..0000000
--- a/src/main/java/com/zdjizhi/source/RichHttpSourceFunction.java
+++ /dev/null
@@ -1,10 +0,0 @@
-package com.zdjizhi.source;
-
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-
-public abstract class RichHttpSourceFunction<OUT> extends AbstractRichFunction implements HttpSourceFunction<OUT> {
-    private static final long serialVersionUID = 1L;
-
-    public RichHttpSourceFunction() {
-    }
-}
\ No newline at end of file
diff --git a/src/main/java/com/zdjizhi/source/SingleHttpSource.java b/src/main/java/com/zdjizhi/source/SingleHttpSource.java
deleted file mode 100644
index 4e4ce87..0000000
--- a/src/main/java/com/zdjizhi/source/SingleHttpSource.java
+++ /dev/null
@@ -1,270 +0,0 @@
-package com.zdjizhi.source;
-
-import cn.hutool.core.io.FileUtil;
-import cn.hutool.core.io.IoUtil;
-import cn.hutool.core.io.file.FileReader;
-import cn.hutool.crypto.digest.DigestUtil;
-import cn.hutool.json.JSONObject;
-import com.alibaba.nacos.api.NacosFactory;
-import com.alibaba.nacos.api.PropertyKeyConst;
-import com.alibaba.nacos.api.config.ConfigService;
-import com.alibaba.nacos.api.config.listener.Listener;
-import com.alibaba.nacos.api.exception.NacosException;
-import com.fasterxml.jackson.databind.JavaType;
-import com.google.common.base.Joiner;
-import com.jayway.jsonpath.JsonPath;
-import com.zdjizhi.common.CommonConfig;
-import com.zdjizhi.common.CustomFile;
-import com.zdjizhi.common.KnowledgeLog;
-import com.zdjizhi.utils.*;
-import org.apache.commons.io.IOUtils;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.http.Header;
-import org.apache.http.message.BasicHeader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.*;
-import java.text.SimpleDateFormat;
-import java.util.*;
-import java.util.concurrent.Executor;
-
-public class SingleHttpSource extends RichHttpSourceFunction<Map<String, String>> {
-
-    private static final Logger logger = LoggerFactory.getLogger(SingleHttpSource.class);
-
-    private static final String EXPR = "$.[?(@.version=='latest' && @.name in ['ip_v4_built_in','ip_v6_built_in','ip_v4_user_defined','ip_v6_user_defined'])].['name','sha256','format','path']";
-
-
-    private static Map<String, String> knowledgeMetaCache = new HashMap<>();
-
-    private static HashMap<String, String> knowledgeUpdateCache;
-
-    private static final int TRY_TIMES = 3;
-
-    private Properties nacosProperties;
-
-    private String NACOS_DATA_ID;
-
-    private String NACOS_GROUP;
-
-    private long NACOS_READ_TIMEOUT;
-
-    private static HttpClientUtils2 httpClientUtils ;
-
-    private ConfigService configService;
-
-    private static Header header;
-
-    private boolean isRunning = true;
-
-
-
-    public SingleHttpSource(Properties nacosProperties, String NACOS_DATA_ID, String NACOS_GROUP, long NACOS_READ_TIMEOUT) {
-        this.nacosProperties = nacosProperties;
-        this.NACOS_DATA_ID = NACOS_DATA_ID;
-        this.NACOS_GROUP = NACOS_GROUP;
-        this.NACOS_READ_TIMEOUT = NACOS_READ_TIMEOUT;
-    }
-
-
-
-    @Override
-    public void open(Configuration parameters) throws Exception {
-        super.open(parameters);
-        httpClientUtils = new HttpClientUtils2();
-        //初始化元数据缓存
-        knowledgeMetaCache = new HashMap<>(16);
-        //初始化定位库缓存
-        knowledgeUpdateCache = new HashMap<>(16);
-
-        header = new BasicHeader("token", CommonConfig.HOS_TOKEN);
-
-        //连接nacos配置
-        try {
-            configService = NacosFactory.createConfigService(nacosProperties);
-        }catch (NacosException e){
-            logger.error("Get Schema config from Nacos error,The exception message is :{}", e.getMessage());
-        }
-
-        //初始化知识库
-        initKnowledge();
-        logger.info("连接nacos：" + nacosProperties.getProperty(PropertyKeyConst.SERVER_ADDR));
-    }
-
-    @Override
-    public void run(SourceContext ctx) throws Exception {
-        if (!knowledgeUpdateCache.isEmpty()){
-            ctx.collect(knowledgeUpdateCache);
-                knowledgeUpdateCache.clear();
-        }
-
-        configService.addListener(NACOS_DATA_ID, NACOS_GROUP, new Listener() {
-            @Override
-            public Executor getExecutor() {
-                return null;
-            }
-
-            @Override
-            public void receiveConfigInfo(String configMsg) {
-                try {
-                    logger.info("receive update config:" + configMsg);
-                    if (StringUtil.isNotBlank(configMsg)) {
-                        ArrayList<Object> metaList = JsonPath.parse(configMsg).read(EXPR);
-                        if (metaList.size() > 0) {
-                            for (Object metadata : metaList) {
-                                JSONObject knowledgeJson = new JSONObject(metadata, false, true);
-                                String fileName = Joiner.on(CommonConfig.LOCATION_SEPARATOR).useForNull("").join(knowledgeJson.getStr("name"),
-                                        knowledgeJson.getStr("format"));
-                                String sha256 = knowledgeJson.getStr("sha256");
-                                String filePath = knowledgeJson.getStr("path");
-                                if (!sha256.equals(knowledgeMetaCache.get(fileName))) {
-                                    knowledgeMetaCache.put(fileName, sha256);
-                                    updateKnowledge(fileName, filePath,sha256);
-                                }
-                            }
-                            if (!knowledgeUpdateCache.isEmpty()){
-                                ctx.collect(knowledgeUpdateCache);
-                                knowledgeUpdateCache.clear();
-                            }
-                        }
-                    }
-
-                } catch (Exception e) {
-                    logger.error("监听nacos配置失败", e);
-                }
-                System.out.println(configMsg);
-            }
-        });
-
-        while (isRunning) {
-            try {
-                Thread.sleep(10000);
-            }catch (InterruptedException e){
-                e.printStackTrace();
-            }
-
-        }
-
-
-
-    }
-
-
-
-    private void initKnowledge(){
-        String configMsg = "";
-        try {
-            configMsg=configService.getConfig(NACOS_DATA_ID, NACOS_GROUP, NACOS_READ_TIMEOUT);
-        } catch (NacosException e) {
-            logger.error("从Nacos获取知识库元数据配置文件异常，异常信息为:{}", e.getMessage());
-        }
-
-        if (StringUtil.isNotBlank(configMsg)){
-            ArrayList<Object> metaList = JsonPath.parse(configMsg).read(EXPR);
-            if (metaList.size() > 0) {
-                for (Object metadata : metaList) {
-                    JSONObject knowledgeJson = new JSONObject(metadata, false, true);
-                    String fileName = Joiner.on(CommonConfig.LOCATION_SEPARATOR).useForNull("").join(knowledgeJson.getStr("name"),
-                            knowledgeJson.getStr("format"));
-                    String sha256 = knowledgeJson.getStr("sha256");
-                    String filePath = knowledgeJson.getStr("path");
-                    byte[] localFileByte = getLocalFile(fileName);
-                    String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                    if (sha256.equals(localFileSha256Hex)){
-                        logger.info("本地文件{}的sha256为:{} ,Nacos内记录为:{} ,sha256相等", fileName, localFileSha256Hex, sha256);
-                        knowledgeMetaCache.put(fileName, sha256);
-                    }else {
-                        logger.info("本地文件{}的sha256为:{} ,Nacos内记录为:{} ,sha256不相等，更新本地文件及缓存", fileName, localFileSha256Hex, sha256);
-                        updateKnowledge(fileName,filePath,sha256);
-                    }
-
-                }
-            }
-        }
-    }
-
-
-
-    private void updateKnowledge(String fileName, String filePath,String sha256) {
-        InputStream inputStream = null;
-        int retryNum = 0;
-        try {
-            while (retryNum < TRY_TIMES){
-                inputStream = httpClientUtils.httpGetInputStream(filePath, 3000, header);
-                if (inputStream !=null){
-                    byte[] downloadBytes = IOUtils.toByteArray(inputStream);
-                    String downloadFileSha256Hex = DigestUtil.sha256Hex(downloadBytes);
-                    if (sha256.equals(downloadFileSha256Hex)&& downloadBytes.length > 0 ){
-                        logger.info("通过HOS下载{}的sha256为:{} ,Nacos内记录为:{} ,sha256相等", fileName, sha256);
-                        boolean updateStatus = updateLocalFile(fileName, downloadBytes);
-                        if (updateStatus){
-                            knowledgeMetaCache.put(fileName,sha256);
-                            knowledgeUpdateCache.put(fileName, sha256);
-                            retryNum = TRY_TIMES;
-                        }else {
-                            retryNum++;
-                            //避免频繁请求HOS
-                            Thread.sleep(10000);
-                        }
-//                            isSending = true;
-                    }else {
-                        logger.error("通过HOS下载{}的sha256为:{} ,Nacos内记录为:{} ,sha256不相等 开始第{}次重试下载文件", fileName, downloadFileSha256Hex, sha256, retryNum);
-                        retryNum++;
-                        //避免频繁请求HOS
-                        Thread.sleep(10000);
-                    }
-                }
-            }
-
-        } catch (IOException ioException) {
-            ioException.printStackTrace();
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        } finally {
-            IOUtils.closeQuietly(inputStream);
-        }
-    }
-
-
-
-    private boolean updateLocalFile(String fileName,byte[] downloadBytes) {
-        FileOutputStream outputStream = null;
-        boolean updateStatus = false;
-        try {
-            FileUtil.mkdir(CommonConfig.DOWNLOAD_PATH);
-            File file = new File(CommonConfig.DOWNLOAD_PATH.concat(File.separator).concat(fileName));
-            outputStream = new FileOutputStream(file);
-            IoUtil.copy(new ByteArrayInputStream(downloadBytes), outputStream);
-            updateStatus=true;
-        } catch (IOException ioe) {
-            logger.error("更新本地文件{}时发生IO异常,异常信息为:", fileName, ioe.getMessage());
-            ioe.printStackTrace();
-        } catch (RuntimeException e) {
-            logger.error("更新本地文件{}时发生异常,异常信息为:", fileName, e.getMessage());
-            e.printStackTrace();
-        } finally {
-            IOUtils.closeQuietly(outputStream);
-        }
-        return updateStatus;
-    }
-
-    private static byte[] getLocalFile(String name) {
-        byte[] fileBytes = null;
-        try {
-            fileBytes=new FileReader(CommonConfig.DOWNLOAD_PATH + name).readBytes();
-        } catch (RuntimeException e) {
-            logger.error("IpLookupUtils download MMDB files error, message is:" + e.getMessage());
-            e.printStackTrace();
-        }
-        return fileBytes;
-    }
-
-
-    @Override
-    public void cancel() {
-        this.isRunning = false;
-    }
-}
-
diff --git a/src/main/java/com/zdjizhi/utils/CollectionUtils.java b/src/main/java/com/zdjizhi/utils/CollectionUtils.java
deleted file mode 100644
index 69d4592..0000000
--- a/src/main/java/com/zdjizhi/utils/CollectionUtils.java
+++ /dev/null
@@ -1,23 +0,0 @@
-package com.zdjizhi.utils;
-
-import java.util.Collection;
-import java.util.HashSet;
-
-/**
- * @author wlh
- * 扩展集合处理工具
- */
-public class CollectionUtils {
-
-    public static<T> Collection<T> takeUniqueLimit(Collection<T> collection, int limit){
-        int i =0;
-        Collection<T> newSet = new HashSet<>();
-        for (T t:collection){
-            if (i < limit){
-                newSet.add(t);
-                i += 1;
-            }
-        }
-        return newSet;
-    }
-}
diff --git a/src/main/java/com/zdjizhi/utils/FileByteUtils.java b/src/main/java/com/zdjizhi/utils/FileByteUtils.java
deleted file mode 100644
index bb1f5aa..0000000
--- a/src/main/java/com/zdjizhi/utils/FileByteUtils.java
+++ /dev/null
@@ -1,24 +0,0 @@
-package com.zdjizhi.utils;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-
-public class FileByteUtils {
-
-    public static byte[] getFileBytes (String filePath) throws IOException {
-        File file = new File(filePath);
-        FileInputStream fis = new FileInputStream(file);
-        ByteArrayOutputStream bos = new ByteArrayOutputStream(1024);
-        byte[] b = new byte[1024];
-        int n;
-        while ((n = fis.read(b)) != -1) {
-            bos.write(b, 0, n);
-        }
-        fis.close();
-        byte[] data = bos.toByteArray();
-        bos.close();
-        return data;
-    }
-}
\ No newline at end of file
diff --git a/src/main/java/com/zdjizhi/utils/FlinkEnvironmentUtils.java b/src/main/java/com/zdjizhi/utils/FlinkEnvironmentUtils.java
index cd628c5..f3d6d11 100644
--- a/src/main/java/com/zdjizhi/utils/FlinkEnvironmentUtils.java
+++ b/src/main/java/com/zdjizhi/utils/FlinkEnvironmentUtils.java
@@ -1,8 +1,6 @@
 package com.zdjizhi.utils;
 
-import com.zdjizhi.common.CommonConfig;
-import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 
@@ -13,7 +11,7 @@ public class FlinkEnvironmentUtils {
     public static StreamExecutionEnvironment streamExeEnv = StreamExecutionEnvironment.getExecutionEnvironment();
 
     static {
-        streamExeEnv.setParallelism(CommonConfig.STREAM_EXECUTION_ENVIRONMENT_PARALLELISM);
+        streamExeEnv.setParallelism(FlowWriteConfig.STREAM_EXECUTION_ENVIRONMENT_PARALLELISM);
 
         /*
         // 每 1000ms 开始一次 checkpoint
diff --git a/src/main/java/com/zdjizhi/utils/HdfsUtils.java b/src/main/java/com/zdjizhi/utils/HdfsUtils.java
deleted file mode 100644
index 4cbc199..0000000
--- a/src/main/java/com/zdjizhi/utils/HdfsUtils.java
+++ /dev/null
@@ -1,75 +0,0 @@
-package com.zdjizhi.utils;
-import cn.hutool.log.Log;
-import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.CommonConfig;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-
-public class HdfsUtils {
-
-    private static final Log logger = LogFactory.get();
-    private static FileSystem fileSystem;
-
-    static {
-        Configuration configuration = new Configuration();
-        try {
-            //配置hdfs相关信息
-            configuration.set("fs.defaultFS","hdfs://ns1");
-            configuration.set("hadoop.proxyuser.root.hosts","*");
-            configuration.set("hadoop.proxyuser.root.groups","*");
-            configuration.set("ha.zookeeper.quorum", CommonConfig.HBASE_ZOOKEEPER_QUORUM);
-            configuration.set("dfs.nameservices","ns1");
-            configuration.set("dfs.ha.namenodes.ns1","nn1,nn2");
-            configuration.set("dfs.namenode.rpc-address.ns1.nn1",CommonConfig.HDFS_URI_NS1);
-            configuration.set("dfs.namenode.rpc-address.ns1.nn2",CommonConfig.HDFS_URI_NS2);
-            configuration.set("dfs.client.failover.proxy.provider.ns1","org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
-            //指定用户
-            System.setProperty("HADOOP_USER_NAME", CommonConfig.HDFS_USER);
-            //创建fileSystem,用于连接hdfs
-            fileSystem = FileSystem.get(configuration);
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    public static boolean isExists(String filePath) throws IOException {
-        return fileSystem.exists(new Path(filePath));
-    }
-
-    public static byte[] getFileBytes(String filePath) throws IOException {
-        try (FSDataInputStream open = fileSystem.open(new Path(filePath))) {
-            byte[] bytes = new byte[open.available()];
-            open.read(0, bytes, 0, open.available());
-            return bytes;
-        } catch (IOException e) {
-            logger.error("An I/O exception when files are download from HDFS. Message is :" + e.getMessage());
-        }
-        return null;
-    }
-
-    public static void uploadFileByBytes(String filePath,byte[] bytes) throws IOException {
-        try (FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path(filePath), true)) {
-            fsDataOutputStream.write(bytes);
-//            fsDataOutputStream.flush();
-        } catch (RuntimeException e) {
-            logger.error("Uploading files to the HDFS is abnormal. Message is :" + e.getMessage());
-        } catch (IOException e) {
-            logger.error("An I/O exception when files are uploaded to HDFS. Message is :" + e.getMessage());
-        }
-    }
-
-    public static void rename(String src, String dst) throws IOException {
-        fileSystem.rename(new Path(src),new Path(dst));
-    }
-
-
-}
diff --git a/src/main/java/com/zdjizhi/utils/HttpClientUtils.java b/src/main/java/com/zdjizhi/utils/HttpClientUtils.java
index d4bf174..b69e38b 100644
--- a/src/main/java/com/zdjizhi/utils/HttpClientUtils.java
+++ b/src/main/java/com/zdjizhi/utils/HttpClientUtils.java
@@ -2,12 +2,16 @@ package com.zdjizhi.utils;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.CommonConfig;
+import com.geedgenetworks.utils.StringUtil;
+
+import com.zdjizhi.common.FlowWriteConfig;
 import org.apache.http.*;
 import org.apache.http.client.ClientProtocolException;
 import org.apache.http.client.HttpRequestRetryHandler;
 import org.apache.http.client.config.RequestConfig;
-import org.apache.http.client.methods.*;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.protocol.HttpClientContext;
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.conn.ConnectTimeoutException;
@@ -48,9 +52,9 @@ public class HttpClientUtils {
     static {
 
         // 设置最大连接数
-        CONN_MANAGER.setMaxTotal(CommonConfig.HTTP_POOL_MAX_CONNECTION);
+        CONN_MANAGER.setMaxTotal(FlowWriteConfig.HTTP_POOL_MAX_CONNECTION);
         // 设置每个连接的路由数
-        CONN_MANAGER.setDefaultMaxPerRoute(CommonConfig.HTTP_POOL_MAX_PER_ROUTE);
+        CONN_MANAGER.setDefaultMaxPerRoute(FlowWriteConfig.HTTP_POOL_MAX_PER_ROUTE);
 
     }
 
@@ -62,11 +66,11 @@ public class HttpClientUtils {
         // 创建Http请求配置参数
         RequestConfig requestConfig = RequestConfig.custom()
                 // 获取连接超时时间
-                .setConnectionRequestTimeout(CommonConfig.HTTP_POOL_REQUEST_TIMEOUT)
+                .setConnectionRequestTimeout(FlowWriteConfig.HTTP_POOL_REQUEST_TIMEOUT)
                 // 请求超时时间
-                .setConnectTimeout(CommonConfig.HTTP_POOL_CONNECT_TIMEOUT)
+                .setConnectTimeout(FlowWriteConfig.HTTP_POOL_CONNECT_TIMEOUT)
                 // 响应超时时间
-                .setSocketTimeout(CommonConfig.HTTP_POOL_RESPONSE_TIMEOUT)
+                .setSocketTimeout(FlowWriteConfig.HTTP_POOL_RESPONSE_TIMEOUT)
                 .build();
 
         /*
diff --git a/src/main/java/com/zdjizhi/utils/IpLocationConfiguration.java b/src/main/java/com/zdjizhi/utils/IpLocationConfiguration.java
deleted file mode 100644
index 1141400..0000000
--- a/src/main/java/com/zdjizhi/utils/IpLocationConfiguration.java
+++ /dev/null
@@ -1,21 +0,0 @@
-package com.zdjizhi.utils;
-import lombok.Data;
-/**
- * @author fy
- * @version 1.0
- * @date 2022/10/19 18:27
- */
-
-
-@Data
-public class IpLocationConfiguration {
-
-    private String ipV4UserDefined;
-
-    private String ipV4BuiltIn;
-
-    private String ipV6UserDefined;
-
-    private String ipV6BuiltIn;
-
-}
diff --git a/src/main/java/com/zdjizhi/utils/IpLookupUtils.java b/src/main/java/com/zdjizhi/utils/IpLookupUtils.java
new file mode 100644
index 0000000..8d84bb9
--- /dev/null
+++ b/src/main/java/com/zdjizhi/utils/IpLookupUtils.java
@@ -0,0 +1,160 @@
+package com.zdjizhi.utils;
+
+import cn.hutool.crypto.digest.DigestUtil;
+import cn.hutool.log.Log;
+import cn.hutool.log.LogFactory;
+import com.alibaba.fastjson2.*;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.config.listener.Listener;
+import com.alibaba.nacos.api.exception.NacosException;
+import com.geedgenetworks.utils.IpLookupV2;
+import com.geedgenetworks.utils.StringUtil;
+import com.google.common.base.Joiner;
+import com.zdjizhi.common.FlowWriteConfig;
+import com.zdjizhi.common.pojo.KnowlegeBaseMeta;
+import com.zdjizhi.utils.connections.http.HttpClientService;
+import com.zdjizhi.utils.connections.nacos.NacosConnection;
+
+import java.io.ByteArrayInputStream;
+import java.util.HashMap;
+import java.util.concurrent.Executor;
+
+/**
+ * @author wangchengcheng
+ * @version 2023/11/10 15:23
+ */
+public class IpLookupUtils {
+    private static final Log logger = LogFactory.get();
+    private static final String ipBuiltInName = "ip_builtin.mmdb";
+    private static final String ipUserDefinedName = "ip_user_defined.mmdb";
+
+    /**
+     * ip定位库
+     */
+    private static IpLookupV2 ipLookup;
+
+    /**
+     * 定位库默认分隔符
+     */
+    private static final String LOCATION_SEPARATOR = ".";
+
+    /**
+     * 最大重试次数
+     */
+    private static final int TRY_TIMES = 5;
+
+    /**
+     * http connections
+     */
+    private static final HttpClientService httpClientService;
+
+    /**
+     * 定位库元数据缓存
+     */
+    private static final HashMap<String, KnowlegeBaseMeta> knowledgeMetaCache = new HashMap<>(16);
+
+    static {
+        JSONPath jsonPath = JSONPath.of(getFilterParameter());
+        httpClientService = new HttpClientService();
+
+        NacosConnection nacosConnection = new NacosConnection();
+        ConfigService schemaService = nacosConnection.getPublicService();
+        try {
+            String configInfo = schemaService.getConfigAndSignListener(FlowWriteConfig.NACOS_KNOWLEDGEBASE_DATA_ID, FlowWriteConfig.NACOS_PUBLIC_GROUP, FlowWriteConfig.NACOS_CONNECTION_TIMEOUT, new Listener() {
+                @Override
+                public Executor getExecutor() {
+                    return null;
+                }
+
+                @Override
+                public void receiveConfigInfo(String configInfo) {
+                    if (StringUtil.isNotBlank(configInfo)) {
+                        updateIpLookup(jsonPath, configInfo);
+                    }
+                }
+            });
+
+            if (StringUtil.isNotBlank(configInfo)) {
+                updateIpLookup(jsonPath, configInfo);
+            }
+        } catch (NacosException e) {
+            logger.error("Get Schema config from Nacos error,The exception message is :" + e.getMessage());
+        }
+    }
+
+    private static void updateIpLookup(JSONPath jsonPath, String configInfo) {
+        String extract = jsonPath.extract(JSONReader.of(configInfo)).toString();
+        if (StringUtil.isNotBlank(extract)) {
+            JSONArray jsonArray = JSON.parseArray(extract);
+            if (jsonArray.size() > 0) {
+                for (int i = 0; i < jsonArray.size(); i++) {
+                    KnowlegeBaseMeta knowlegeBaseMeta = JSONObject.parseObject(jsonArray.getString(i), KnowlegeBaseMeta.class);
+                    String fileName = Joiner.on(LOCATION_SEPARATOR).useForNull("").join(knowlegeBaseMeta.getName(), knowlegeBaseMeta.getFormat());
+                    knowledgeMetaCache.put(fileName, knowlegeBaseMeta);
+                }
+                reloadIpLookup();
+            }
+        }
+    }
+
+    /**
+     * 从HDFS下载文件更新IpLookup
+     */
+    private static void reloadIpLookup() {
+        IpLookupV2.Builder builder = new IpLookupV2.Builder(false);
+        for (String fileName : knowledgeMetaCache.keySet()) {
+            int retryNum = 0;
+            KnowlegeBaseMeta knowlegeBaseMeta = knowledgeMetaCache.get(fileName);
+            String metaSha256 = knowlegeBaseMeta.getSha256();
+            while (retryNum < TRY_TIMES) {
+                System.out.println("download file ：" + fileName + ",HOS path :" + knowlegeBaseMeta.getPath());
+                Long startTime = System.currentTimeMillis();
+                byte[] httpGetByte = httpClientService.httpGetByte(knowlegeBaseMeta.getPath(), FlowWriteConfig.HTTP_SOCKET_TIMEOUT);
+                if (httpGetByte != null && httpGetByte.length > 0) {
+                    String downloadFileSha256 = DigestUtil.sha256Hex(httpGetByte);
+                    if (metaSha256.equals(downloadFileSha256)) {
+                        ByteArrayInputStream inputStream = new ByteArrayInputStream(httpGetByte);
+                        switch (fileName) {
+                            case ipBuiltInName:
+                                builder.loadDataFile(inputStream);
+                                break;
+                            case ipUserDefinedName:
+                                builder.loadDataFilePrivate(inputStream);
+                                break;
+                            default:
+                        }
+                        System.out.println("update " + fileName + " finished, speed :" + (System.currentTimeMillis() - startTime) + "ms");
+                        retryNum = TRY_TIMES;
+                    } else {
+                        logger.error("通过HOS下载{}的sha256为:{} ,Nacos内记录为:{} ,sha256不相等 开始第{}次重试下载文件", fileName, downloadFileSha256, metaSha256, retryNum);
+                        retryNum++;
+                    }
+                } else {
+                    logger.error("通过HOS下载{}的流为空 ,开始第{}次重试下载文件", fileName, retryNum);
+                    retryNum++;
+                }
+            }
+        }
+        ipLookup = builder.build();
+    }
+
+    /**
+     * 根据配置组合生成知识库元数据过滤参数
+     *
+     * @return 过滤参数
+     */
+    private static String getFilterParameter() {
+//       String expr = "$.[?(@.version=='latest' && @.name in ['ip_built_in','ip_user_defined'])].['name','sha256','format','path']";
+
+
+        String expr = "[?(@.version=='latest')][?(@.name in ('ip_builtin','ip_user_defined'))]";
+
+        return expr;
+    }
+
+    public static String getCountryLookup(String ip) {
+        return ipLookup.countryLookup(ip);
+    }
+
+
+}
diff --git a/src/main/java/com/zdjizhi/utils/IpUtils.java b/src/main/java/com/zdjizhi/utils/IpUtils.java
deleted file mode 100644
index 4e02664..0000000
--- a/src/main/java/com/zdjizhi/utils/IpUtils.java
+++ /dev/null
@@ -1,224 +0,0 @@
-package com.zdjizhi.utils;
-
-import cn.hutool.core.io.file.FileReader;
-import cn.hutool.crypto.digest.DigestUtil;
-import com.zdjizhi.common.CommonConfig;
-import com.zdjizhi.common.CustomFile;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.List;
-import java.util.Map;
-
-public class IpUtils {
-    public static IpLookupV2 ipLookup ;
-
-    private static Logger LOG = LoggerFactory.getLogger(IpUtils.class);
-
-
-    /**
-     * IP定位库工具类
-     */
-//    public static IpLookupV2 ipLookup = new IpLookupV2.Builder(false)
-//            .loadDataFileV4(CommonConfig.IP_MMDB_PATH + "ip_v4_built_in.mmdb")
-//            .loadDataFileV6(CommonConfig.IP_MMDB_PATH + "ip_v6_built_in.mmdb")
-//            .loadDataFilePrivateV4(CommonConfig.IP_MMDB_PATH + "ip_v4_user_defined.mmdb")
-//            .loadDataFilePrivateV6(CommonConfig.IP_MMDB_PATH + "ip_v6_user_defined.mmdb")
-//            .build();
-
-
-    public static void loadIpLook(){
-        try {
-            IpLookupV2.Builder builder = new IpLookupV2.Builder(false);
-            if ("CLUSTER".equals(CommonConfig.CLUSTER_OR_SINGLE)) {
-                byte[] ipv4BuiltBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v4_built_in.mmdb");
-                if (ipv4BuiltBytes!=null){
-                    InputStream ipv4BuiltInputStream = new ByteArrayInputStream(ipv4BuiltBytes);
-                    builder.loadDataFileV4(ipv4BuiltInputStream);
-                }
-
-                byte[] ipv6BuiltBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v6_built_in.mmdb");
-                if (ipv6BuiltBytes!=null){
-                    InputStream ipv6BuiltInputStream = new ByteArrayInputStream(ipv6BuiltBytes);
-                    builder.loadDataFileV6(ipv6BuiltInputStream);
-                }
-
-                byte[] ipv4UserBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v4_user_defined.mmdb");
-                if (ipv4UserBytes!=null){
-                    InputStream ipv4UserInputStream = new ByteArrayInputStream(ipv4UserBytes);
-                    builder.loadDataFilePrivateV4(ipv4UserInputStream);
-                }
-
-                byte[] ipv6UserBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v6_user_defined.mmdb");
-                if (ipv6UserBytes!=null){
-                    InputStream ipv6UserInputStream = new ByteArrayInputStream(ipv6UserBytes);
-                    builder.loadDataFilePrivateV6(ipv6UserInputStream);
-                }
-            }else if ("SINGLE".equals(CommonConfig.CLUSTER_OR_SINGLE)){
-                byte[] ipv4BuiltBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v4_built_in.mmdb");
-                if (ipv4BuiltBytes!=null){
-                    InputStream ipv4BuiltInputStream = new ByteArrayInputStream(ipv4BuiltBytes);
-                    builder.loadDataFileV4(ipv4BuiltInputStream);
-                }
-
-                byte[] ipv6BuiltBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v6_built_in.mmdb");
-                if (ipv6BuiltBytes!=null){
-                    InputStream ipv6BuiltInputStream = new ByteArrayInputStream(ipv6BuiltBytes);
-                    builder.loadDataFileV6(ipv6BuiltInputStream);
-                }
-
-                byte[] ipv4UserBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v4_user_defined.mmdb");
-                if (ipv4UserBytes!=null){
-                    InputStream ipv4UserInputStream = new ByteArrayInputStream(ipv4UserBytes);
-                    builder.loadDataFilePrivateV4(ipv4UserInputStream);
-                }
-
-                byte[] ipv6UserBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v6_user_defined.mmdb");
-                if (ipv6UserBytes!=null){
-                    InputStream ipv6UserInputStream = new ByteArrayInputStream(ipv6UserBytes);
-                    builder.loadDataFilePrivateV6(ipv6UserInputStream);
-                }
-            }
-            ipLookup = builder.build();
-
-        }catch (Exception e){
-            LOG.error("加载失败",e);
-        }
-    }
-
-    public static void updateIpLook(Map<String, String> knowledgeFileCache){
-        try{
-            IpLookupV2.Builder builder = new IpLookupV2.Builder(false);
-            if ("CLUSTER".equals(CommonConfig.CLUSTER_OR_SINGLE)) {
-                byte[] ipv4BuiltBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v4_built_in.mmdb");
-                if (ipv4BuiltBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v4_built_in.mmdb")){
-                        String sha256 = knowledgeFileCache.get("ip_v4_built_in.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v4_built_in.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)){
-                            builder.loadDataFileV4(new ByteArrayInputStream(ipv4BuiltBytes));
-                        }
-                    }
-                }
-
-                byte[] ipv6BuiltBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v6_built_in.mmdb");
-                if (ipv6BuiltBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v6_built_in.mmdb")) {
-                        String sha256 = knowledgeFileCache.get("ip_v6_built_in.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v6_built_in.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)) {
-                            builder.loadDataFileV6(new ByteArrayInputStream(ipv6BuiltBytes));
-                        }
-                    }
-                }
-
-                byte[] ipv4UserBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v4_user_defined.mmdb");
-                if (ipv4UserBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v4_user_defined.mmdb")) {
-                        String sha256 = knowledgeFileCache.get("ip_v4_user_defined.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v4_user_defined.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)) {
-                            builder.loadDataFilePrivateV4(new ByteArrayInputStream(ipv4UserBytes));
-                        }
-                    }
-                }
-
-                byte[] ipv6UserBytes = HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + "ip_v6_user_defined.mmdb");
-                if (ipv6UserBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v6_user_defined.mmdb")) {
-                        String sha256 = knowledgeFileCache.get("ip_v6_user_defined.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v6_user_defined.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)) {
-                            builder.loadDataFilePrivateV6(new ByteArrayInputStream(ipv6UserBytes));
-                        }
-                    }
-                }
-            }else if ("SINGLE".equals(CommonConfig.CLUSTER_OR_SINGLE)){
-                byte[] ipv4BuiltBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v4_built_in.mmdb");
-                if (ipv4BuiltBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v4_built_in.mmdb")){
-                        String sha256 = knowledgeFileCache.get("ip_v4_built_in.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v4_built_in.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)){
-                            builder.loadDataFileV4(new ByteArrayInputStream(ipv4BuiltBytes));
-                        }
-                    }
-                }
-
-                byte[] ipv6BuiltBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v6_built_in.mmdb");
-                if (ipv6BuiltBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v6_built_in.mmdb")) {
-                        String sha256 = knowledgeFileCache.get("ip_v6_built_in.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v6_built_in.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)) {
-                            builder.loadDataFileV6(new ByteArrayInputStream(ipv6BuiltBytes));
-                        }
-                    }
-                }
-
-                byte[] ipv4UserBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v4_user_defined.mmdb");
-                if (ipv4UserBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v4_user_defined.mmdb")) {
-                        String sha256 = knowledgeFileCache.get("ip_v4_user_defined.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v4_user_defined.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)) {
-                            builder.loadDataFilePrivateV4(new ByteArrayInputStream(ipv4UserBytes));
-                        }
-                    }
-                }
-
-                byte[] ipv6UserBytes = FileByteUtils.getFileBytes(CommonConfig.DOWNLOAD_PATH + "ip_v6_user_defined.mmdb");
-                if (ipv6UserBytes!=null){
-                    if (knowledgeFileCache.containsKey("ip_v6_user_defined.mmdb")) {
-                        String sha256 = knowledgeFileCache.get("ip_v6_user_defined.mmdb");
-                        byte[] localFileByte = getLocalFile("ip_v6_user_defined.mmdb");
-                        String localFileSha256Hex = DigestUtil.sha256Hex(localFileByte);
-                        if (sha256.equals(localFileSha256Hex)) {
-                            builder.loadDataFilePrivateV6(new ByteArrayInputStream(ipv6UserBytes));
-                        }
-                    }
-                }
-            }
-            ipLookup = builder.build();
-        }catch (Exception e){
-            LOG.error("加载失败",e);
-        }
-
-    }
-
-
-    private static byte[] getLocalFile(String name) {
-        byte[] fileBytes = null;
-        try {
-            fileBytes = "CLUSTER".equals(CommonConfig.CLUSTER_OR_SINGLE) ?
-                    HdfsUtils.getFileBytes(CommonConfig.HDFS_PATH + name) :
-                    new FileReader(CommonConfig.DOWNLOAD_PATH + name).readBytes();
-        } catch (RuntimeException | IOException e) {
-            e.printStackTrace();
-        }
-        return fileBytes;
-    }
-
-    public static void main(String[] args) {
-        System.out.println(ipLookup.countryLookup("49.7.115.37"));
-
-        String ips = "182.168.50.23,182.168.50.45,182.168.56.9,182.168.56.8,92.168.50.58,19.168.56.7,12.168.56.6,2.168.50.40,1.168.50.19,9.168.50.6,2.168.50.4,192.168.56.17,192.168.50.27,192.168.50.26,192.168.50.18,192.168.56.3,192.168.56.10";
-        for (String ip:ips.split(",")){
-            System.out.println(ip+"--"+ipLookup.countryLookup(ip));
-        }
-
-    }
-
-
-
-}
diff --git a/src/main/java/com/zdjizhi/utils/KafkaUtils.java b/src/main/java/com/zdjizhi/utils/KafkaUtils.java
index b0312a5..4298c40 100644
--- a/src/main/java/com/zdjizhi/utils/KafkaUtils.java
+++ b/src/main/java/com/zdjizhi/utils/KafkaUtils.java
@@ -1,6 +1,6 @@
 package com.zdjizhi.utils;
 
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
 
@@ -11,11 +11,11 @@ public class KafkaUtils {
 
     private static Properties getKafkaSinkProperty(){
         Properties properties = new Properties();
-        properties.setProperty("bootstrap.servers", CommonConfig.KAFKA_OUTPUT_BOOTSTRAP_SERVERS);
-        if (CommonConfig.SASL_JAAS_CONFIG_FLAG == 1){
+        properties.setProperty("bootstrap.servers", FlowWriteConfig.KAFKA_OUTPUT_BOOTSTRAP_SERVERS);
+        if (FlowWriteConfig.SASL_JAAS_CONFIG_FLAG == 1){
             properties.put("security.protocol", "SASL_PLAINTEXT");
             properties.put("sasl.mechanism", "PLAIN");
-            properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""+CommonConfig.SASL_JAAS_CONFIG_USER+"\" password=\""+CommonConfig.SASL_JAAS_CONFIG_PASSWORD+"\";");
+            properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""+ FlowWriteConfig.SASL_JAAS_CONFIG_USER+"\" password=\""+ FlowWriteConfig.SASL_JAAS_CONFIG_PASSWORD+"\";");
         }
 
         return properties;
diff --git a/src/main/java/com/zdjizhi/utils/SnowflakeId.java b/src/main/java/com/zdjizhi/utils/SnowflakeId.java
index 8b26f43..0af5582 100644
--- a/src/main/java/com/zdjizhi/utils/SnowflakeId.java
+++ b/src/main/java/com/zdjizhi/utils/SnowflakeId.java
@@ -2,7 +2,7 @@ package com.zdjizhi.utils;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.CommonConfig;
+import com.zdjizhi.common.FlowWriteConfig;
 
 public class SnowflakeId {
 //    private static final Logger logger = LoggerFactory.getLogger(SnowflakeId.class);
@@ -99,7 +99,7 @@ public class SnowflakeId {
     private static ZookeeperUtils zookeeperUtils = new ZookeeperUtils();
 
     static {
-        idWorker = new SnowflakeId(CommonConfig.HBASE_ZOOKEEPER_QUORUM, CommonConfig.DATA_CENTER_ID_NUM);
+        idWorker = new SnowflakeId(FlowWriteConfig.HBASE_ZOOKEEPER_QUORUM, FlowWriteConfig.DATA_CENTER_ID_NUM);
     }
 
     //==============================Constructors=====================================
@@ -108,7 +108,7 @@ public class SnowflakeId {
      * 构造函数
      */
     private SnowflakeId(String zookeeperIp, long dataCenterIdNum) {
-        DistributedLock lock = new DistributedLock(CommonConfig.HBASE_ZOOKEEPER_QUORUM, "disLocks1");
+        DistributedLock lock = new DistributedLock(FlowWriteConfig.HBASE_ZOOKEEPER_QUORUM, "disLocks1");
         try {
             lock.lock();
             int tmpWorkerId = zookeeperUtils.modifyNode("/Snowflake/" + "worker" + dataCenterIdNum, zookeeperIp);
diff --git a/src/main/java/com/zdjizhi/utils/HttpClientUtils2.java b/src/main/java/com/zdjizhi/utils/connections/http/HttpClientService.java
index 1136e6d..280c0a2 100644
--- a/src/main/java/com/zdjizhi/utils/HttpClientUtils2.java
+++ b/src/main/java/com/zdjizhi/utils/connections/http/HttpClientService.java
@@ -1,6 +1,11 @@
-package com.zdjizhi.utils;
+package com.zdjizhi.utils.connections.http;
 
-import com.zdjizhi.common.CommonConfig;
+import cn.hutool.log.Log;
+import cn.hutool.log.LogFactory;
+
+import com.geedgenetworks.utils.StringUtil;
+import com.zdjizhi.common.FlowWriteConfig;
+import com.zdjizhi.utils.exception.FlowWriteException;
 import org.apache.commons.io.IOUtils;
 import org.apache.http.*;
 import org.apache.http.client.ClientProtocolException;
@@ -8,63 +13,35 @@ import org.apache.http.client.HttpRequestRetryHandler;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.protocol.HttpClientContext;
-import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.config.Registry;
 import org.apache.http.config.RegistryBuilder;
 import org.apache.http.conn.ConnectTimeoutException;
 import org.apache.http.conn.ConnectionKeepAliveStrategy;
-import org.apache.http.conn.HttpHostConnectException;
 import org.apache.http.conn.socket.ConnectionSocketFactory;
 import org.apache.http.conn.socket.PlainConnectionSocketFactory;
 import org.apache.http.conn.ssl.NoopHostnameVerifier;
 import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
-import org.apache.http.entity.ByteArrayEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.apache.http.message.BasicHeaderElementIterator;
 import org.apache.http.protocol.HTTP;
 import org.apache.http.util.EntityUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import javax.net.ssl.*;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InterruptedIOException;
-import java.net.URI;
+import java.net.SocketTimeoutException;
 import java.net.UnknownHostException;
-import java.nio.charset.StandardCharsets;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
 import java.security.cert.X509Certificate;
-import java.util.Map;
-
-import static org.apache.kafka.common.requests.FetchMetadata.log;
 
-/**
- * http client工具类
- */
-public class HttpClientUtils2 {
-    /** 全局连接池对象 */
-    private static final PoolingHttpClientConnectionManager CONN_MANAGER = new PoolingHttpClientConnectionManager();
+public class HttpClientService {
 
-    private static Logger logger = LoggerFactory.getLogger(HttpClientUtils2.class);
-    public static final String ERROR_MESSAGE = "-1";
-
-    /*
-     * 静态代码块配置连接池信息
-     */
-    static {
-
-        // 设置最大连接数
-        CONN_MANAGER.setMaxTotal(CommonConfig.HTTP_POOL_MAX_CONNECTION);
-        // 设置每个连接的路由数
-        CONN_MANAGER.setDefaultMaxPerRoute(CommonConfig.HTTP_POOL_MAX_PER_ROUTE);
-
-    }
+    private static final Log log = LogFactory.get();
 
     /**
      * 在调用SSL之前需要重写验证方法，取消检测SSL
@@ -80,9 +57,11 @@ public class HttpClientUtils2 {
                 public X509Certificate[] getAcceptedIssuers() {
                     return null;
                 }
+
                 @Override
                 public void checkClientTrusted(X509Certificate[] xcs, String str) {
                 }
+
                 @Override
                 public void checkServerTrusted(X509Certificate[] xcs, String str) {
                 }
@@ -96,31 +75,33 @@ public class HttpClientUtils2 {
             // 创建ConnectionManager，添加Connection配置信息
             PoolingHttpClientConnectionManager connManager = new PoolingHttpClientConnectionManager(socketFactoryRegistry);
             // 设置最大连接数
-            connManager.setMaxTotal(CommonConfig.HTTP_POOL_MAX_CONNECTION);
+            connManager.setMaxTotal(FlowWriteConfig.HTTP_POOL_MAX_CONNECTION);
             // 设置每个连接的路由数
-            connManager.setDefaultMaxPerRoute(CommonConfig.HTTP_POOL_MAX_PER_ROUTE);
+            connManager.setDefaultMaxPerRoute(FlowWriteConfig.HTTP_POOL_MAX_PER_ROUTE);
             return connManager;
         } catch (KeyManagementException | NoSuchAlgorithmException e) {
-            throw new RuntimeException(e.getMessage());
+            throw new FlowWriteException(e.getMessage());
         }
     }
 
     /**
      * 获取Http客户端连接对象
+     *
+     * @param socketTimeOut 响应超时时间
      * @return Http客户端连接对象
      */
-    private  CloseableHttpClient getHttpClient() {
+    private CloseableHttpClient getHttpClient(int socketTimeOut) {
         // 创建Http请求配置参数
         RequestConfig requestConfig = RequestConfig.custom()
                 // 获取连接超时时间
-                .setConnectionRequestTimeout(CommonConfig.HTTP_POOL_REQUEST_TIMEOUT)
+                .setConnectionRequestTimeout(FlowWriteConfig.HTTP_POOL_REQUEST_TIMEOUT)
                 // 请求超时时间
-                .setConnectTimeout(CommonConfig.HTTP_POOL_CONNECT_TIMEOUT)
+                .setConnectTimeout(FlowWriteConfig.HTTP_POOL_CONNECT_TIMEOUT)
                 // 响应超时时间
-                .setSocketTimeout(CommonConfig.HTTP_POOL_RESPONSE_TIMEOUT)
+                .setSocketTimeout(socketTimeOut)
                 .build();
 
-        /*
+        /**
          * 测出超时重试机制为了防止超时不生效而设置
          *  如果直接放回false,不重试
          *  这里会根据情况进行判断是否重试
@@ -135,25 +116,32 @@ public class HttpClientUtils2 {
             if (exception instanceof SSLHandshakeException) {// 不要重试SSL握手异常
                 return false;
             }
+            if (exception instanceof SocketTimeoutException) {
+                if (exception.getMessage().contains("Read timed out")) {
+                    return false;
+                }
+            }
             if (exception instanceof UnknownHostException) {// 目标服务器不可达
                 return false;
             }
             if (exception instanceof ConnectTimeoutException) {// 连接被拒绝
                 return false;
             }
-            if (exception instanceof HttpHostConnectException) {// 连接被拒绝
-                return false;
-            }
             if (exception instanceof SSLException) {// ssl握手异常
                 return false;
             }
             if (exception instanceof InterruptedIOException) {// 超时
                 return true;
             }
+
+
             HttpClientContext clientContext = HttpClientContext.adapt(context);
             HttpRequest request = clientContext.getRequest();
             // 如果请求是幂等的，就再次尝试
-            return !(request instanceof HttpEntityEnclosingRequest);
+            if (!(request instanceof HttpEntityEnclosingRequest)) {
+                return true;
+            }
+            return false;
         };
 
 
@@ -164,7 +152,7 @@ public class HttpClientUtils2 {
                 HeaderElement he = it.nextElement();
                 String param = he.getName();
                 String value = he.getValue();
-                if (value != null && "timeout".equalsIgnoreCase(param)) {
+                if (value != null && param.equalsIgnoreCase("timeout")) {
                     return Long.parseLong(value) * 1000;
                 }
             }
@@ -183,11 +171,10 @@ public class HttpClientUtils2 {
                 .build();
     }
 
-    // TODO: 2022/10/19  加载知识库
     public InputStream httpGetInputStream(String url, int socketTimeout, Header... headers) {
         InputStream result = null;
         // 获取客户端连接对象
-        CloseableHttpClient httpClient = getHttpClient();// TODO: 2022/10/19 去掉了 socketTimeout
+        CloseableHttpClient httpClient = getHttpClient(socketTimeout);
         // 创建GET请求对象
         HttpGet httpGet = new HttpGet(url);
         if (StringUtil.isNotEmpty(headers)) {
@@ -228,7 +215,47 @@ public class HttpClientUtils2 {
     }
 
 
+    public byte[] httpGetByte(String url, int socketTimeout, Header... headers) {
+        byte[] result = null;
+        // 获取客户端连接对象
+        CloseableHttpClient httpClient = getHttpClient(socketTimeout);
+        // 创建GET请求对象
+        HttpGet httpGet = new HttpGet(url);
+        if (StringUtil.isNotEmpty(headers)) {
+            for (Header h : headers) {
+                httpGet.addHeader(h);
+            }
+        }
+        CloseableHttpResponse response = null;
+
+        try {
+            // 执行请求
+            response = httpClient.execute(httpGet);
+            // 获取响应实体
+            result = IOUtils.toByteArray(response.getEntity().getContent());
+            // 获取响应信息
+            EntityUtils.consume(response.getEntity());
+        } catch (ClientProtocolException e) {
+            log.error("current file: {},Protocol error:{}", url, e.getMessage());
+
+        } catch (ParseException e) {
+            log.error("current file: {}, Parser error:{}", url, e.getMessage());
 
+        } catch (IOException e) {
+            log.error("current file: {},IO error:{}", url, e.getMessage());
 
+        } finally {
+            if (null != response) {
+                try {
+                    EntityUtils.consume(response.getEntity());
+                    response.close();
+                } catch (IOException e) {
+                    log.error("Release Connection error:{}", e.getMessage());
+
+                }
+            }
+            return result;
+        }
+    }
 
 }
diff --git a/src/main/java/com/zdjizhi/utils/connections/nacos/NacosConnection.java b/src/main/java/com/zdjizhi/utils/connections/nacos/NacosConnection.java
new file mode 100644
index 0000000..8de0ae0
--- /dev/null
+++ b/src/main/java/com/zdjizhi/utils/connections/nacos/NacosConnection.java
@@ -0,0 +1,54 @@
+package com.zdjizhi.utils.connections.nacos;
+
+import cn.hutool.log.Log;
+import cn.hutool.log.LogFactory;
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.PropertyKeyConst;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.exception.NacosException;
+import com.zdjizhi.common.FlowWriteConfig;
+
+
+import java.util.Properties;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi.tools.connections.nacos
+ * @Description:
+ * @date 2023/7/2714:49
+ */
+public class NacosConnection {
+    private static final Log logger = LogFactory.get();
+
+    private ConfigService configService;
+
+
+    public ConfigService getDosService() {
+        Properties properties = new Properties();
+        properties.setProperty(PropertyKeyConst.SERVER_ADDR, FlowWriteConfig.NACOS_SERVER);
+        properties.setProperty(PropertyKeyConst.NAMESPACE, FlowWriteConfig.NACOS_DOS_NAMESPACE);
+        properties.setProperty(PropertyKeyConst.USERNAME, FlowWriteConfig.NACOS_USERNAME);
+        properties.setProperty(PropertyKeyConst.PASSWORD, FlowWriteConfig.NACOS_PIN);
+        try {
+            configService = NacosFactory.createConfigService(properties);
+        } catch (NacosException e) {
+            logger.error("NacosException:{}", e);
+        }
+        return configService;
+    }
+
+
+    public ConfigService getPublicService() {
+        Properties properties = new Properties();
+        properties.setProperty(PropertyKeyConst.SERVER_ADDR, FlowWriteConfig.NACOS_SERVER);
+        properties.setProperty(PropertyKeyConst.NAMESPACE, FlowWriteConfig.NACOS_PUBLIC_NAMESPACE);
+        properties.setProperty(PropertyKeyConst.USERNAME, FlowWriteConfig.NACOS_USERNAME);
+        properties.setProperty(PropertyKeyConst.PASSWORD, FlowWriteConfig.NACOS_PIN);
+        try {
+            configService = NacosFactory.createConfigService(properties);
+        } catch (NacosException e) {
+            logger.error("NacosException:{}", e);
+        }
+        return configService;
+    }
+}
diff --git a/src/main/java/com/zdjizhi/utils/NacosUtils.java b/src/main/java/com/zdjizhi/utils/connections/nacos/NacosUtils.java
index 25e6809..d60dd0f 100644
--- a/src/main/java/com/zdjizhi/utils/NacosUtils.java
+++ b/src/main/java/com/zdjizhi/utils/connections/nacos/NacosUtils.java
@@ -1,11 +1,10 @@
-package com.zdjizhi.utils;
+package com.zdjizhi.utils.connections.nacos;
 
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
-import com.alibaba.nacos.api.NacosFactory;
-import com.alibaba.nacos.api.PropertyKeyConst;
 import com.alibaba.nacos.api.config.ConfigService;
 import com.alibaba.nacos.api.config.listener.Listener;
+import com.zdjizhi.common.FlowWriteConfig;
 
 import java.io.IOException;
 import java.io.StringReader;
@@ -13,40 +12,18 @@ import java.util.Properties;
 import java.util.concurrent.Executor;
 
 public class NacosUtils {
-//    private static final Logger logger = LoggerFactory.getLogger(NacosUtils.class);
     private static final Log logger = LogFactory.get();
-    private static Properties nacosProperties = new Properties();
     private static Properties commonProperties = new Properties();
 
-
-    private static final String NACOS_SERVER_ADDR = CommonConfigurations.getStringProperty("nacos.server.addr");
-    private static final String NACOS_STATIC_NAMESPACE = CommonConfigurations.getStringProperty("nacos.static.namespace");
-    private static final String NACOS_USERNAME = CommonConfigurations.getStringProperty("nacos.username");
-    private static final String NACOS_PASSWORD = CommonConfigurations.getStringProperty("nacos.password");
-    private static final String NACOS_STATIC_DATA_ID = CommonConfigurations.getStringProperty("nacos.static.data.id");
-    private static final String NACOS_STATIC_GROUP = CommonConfigurations.getStringProperty("nacos.static.group");
-    private static final long NACOS_READ_TIMEOUT = CommonConfigurations.getLongProperty("nacos.read.timeout");
-
     static {
-        createConfigService();
-    }
-
-    private static void getProperties() {
-        nacosProperties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_SERVER_ADDR);
-        nacosProperties.setProperty(PropertyKeyConst.NAMESPACE, NACOS_STATIC_NAMESPACE);
-        nacosProperties.setProperty(PropertyKeyConst.USERNAME, NACOS_USERNAME);
-        nacosProperties.setProperty(PropertyKeyConst.PASSWORD, NACOS_PASSWORD);
-    }
-
-    private static void createConfigService() {
+        NacosConnection nacosConnection = new NacosConnection();
+        ConfigService dosService = nacosConnection.getDosService();
         try {
-            getProperties();
-            ConfigService configService = NacosFactory.createConfigService(nacosProperties);
-            String config = configService.getConfig(NACOS_STATIC_DATA_ID, NACOS_STATIC_GROUP, NACOS_READ_TIMEOUT);
-            commonProperties.load(new StringReader(config));
+            String config = dosService.getConfig(FlowWriteConfig.NACOS_DOS_DATA_ID, FlowWriteConfig.NACOS_DOS_GROUP, FlowWriteConfig.NACOS_CONNECTION_TIMEOUT);
 
+            commonProperties.load(new StringReader(config));
 
-            configService.addListener(NACOS_STATIC_DATA_ID, NACOS_STATIC_GROUP, new Listener() {
+            dosService.addListener(FlowWriteConfig.NACOS_DOS_DATA_ID, FlowWriteConfig.NACOS_DOS_GROUP, new Listener() {
                 @Override
                 public Executor getExecutor() {
                     return null;
diff --git a/src/main/java/com/zdjizhi/utils/exception/FlowWriteException.java b/src/main/java/com/zdjizhi/utils/exception/FlowWriteException.java
new file mode 100644
index 0000000..44fdef8
--- /dev/null
+++ b/src/main/java/com/zdjizhi/utils/exception/FlowWriteException.java
@@ -0,0 +1,13 @@
+package com.zdjizhi.utils.exception;
+
+
+public class FlowWriteException extends RuntimeException {
+
+    public FlowWriteException() {
+    }
+
+    public FlowWriteException(String message) {
+        super(message);
+    }
+
+}
diff --git a/src/main/resources/common.properties b/src/main/resources/common.properties
index 083dbba..b4b44ec 100644
--- a/src/main/resources/common.properties
+++ b/src/main/resources/common.properties
@@ -74,10 +74,6 @@ destination.ip.partition.num=10000
 
 data.center.id.num=15
 
-#IP mmdb库路径
-ip.mmdb.path=D:\\data\\dat\\bak\\
-#ip.mmdb.path=/home/bigdata/topology/dat/
-#ip.mmdb.path=/home/bigdata/wlh/topology/dos-detection/dat/
 
 #bifang服务访问地址
 bifang.server.uri=http://192.168.44.72:80
@@ -90,10 +86,10 @@ bifang.server.encryptpwd.path=/v1/user/encryptpwd
 bifang.server.login.path=/v1/user/login
 
 #获取vaysId路径信息
-bifang.server.policy.vaysid.path=/v1/system/vsys/
+bifang.server.policy.vaysid.path=/v1/admin/vsys
 
 #获取静态阈值路径信息
-bifang.server.policy.threshold.path=/v1/policy/profile/DoS/detection/threshold
+bifang.server.policy.threshold.path=/v1/policy/profile/dos/detection/threshold
 
 #http请求相关参数
 #最大连接数
@@ -129,30 +125,15 @@ sasl.jaas.config.flag=1
 nacos.server.addr=192.168.44.12:8848
 nacos.username=nacos
 nacos.password=nacos
-nacos.read.timeout=5000
 ############################## Nacos ---知识库配置 ######################################
 nacos.namespace=public
 nacos.data.id=knowledge_base.json
 nacos.group=DEFAULT_GROUP
-
+nacos.connection.timeout=60000
 
 ############################## Nacos ---静态阈值配置 ######################################
-nacos.static.namespace=test
-nacos.static.data.id=dos_detection.properties
-nacos.static.group=Galaxy
-
-############################## hos Token 配置 ######################################
-hos.token=c21f969b5f03d33d43e04f8f136e7682
-
-############################# 选择集群模式或者单机模式 配置 ######################################
-cluster.or.single=CLUSTER
-#cluster.or.single=SINGLE
-
-############################## 集群模式配置文件路径 配置 ######################################
-hdfs.path=/test/TEST/
-hdfs.uri.nn1=192.168.40.151:9000
-hdfs.uri.nn2=192.168.40.152:9000
-hdfs.user=dos
+nacos.dos.namespace=test
+nacos.dos.data.id=dos_detection.properties
+nacos.dos.group=Galaxy
 
-############################## 单机模式配置文件下载路径 配置 ######################################
-download.path=D:\\ttt\\
\ No newline at end of file
+http.socket.timeout=90000
\ No newline at end of file
diff --git a/src/main/resources/core-site.xml b/src/main/resources/core-site.xml
deleted file mode 100644
index c103340..0000000
--- a/src/main/resources/core-site.xml
+++ /dev/null
@@ -1,58 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
-<!--
-  Licensed under the Apache License, Version 2.0 (the "License");
-  you may not use this file except in compliance with the License.
-  You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-  Unless required by applicable law or agreed to in writing, software
-  distributed under the License is distributed on an "AS IS" BASIS,
-  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  See the License for the specific language governing permissions and
-  limitations under the License. See accompanying LICENSE file.
--->
-
-<!-- Put site-specific property overrides in this file. -->
-
-<configuration>
-    <property>  
-        <name>fs.defaultFS</name>  
-        <value>hdfs://ns1</value>  
-    </property>
-    <property>
-        <name>hadoop.tmp.dir</name>
-        <value>file:/home/tsg/olap/hadoop/tmp</value>
-    </property>
-    <property>
-        <name>io.file.buffer.size</name>
-        <value>131702</value>
-    </property>
-    <property>
-	<name>hadoop.proxyuser.root.hosts</name>
-	<value>*</value>
-    </property>
-    <property>
-	<name>hadoop.proxyuser.root.groups</name>
-	<value>*</value>
-    </property>
-    <property>
-	<name>hadoop.logfile.size</name>
-	<value>10000000</value>
-	<description>The max size of each log file</description>
-    </property>
-    <property>
-	<name>hadoop.logfile.count</name>
-	<value>1</value>
-	<description>The max number of log files</description>
-    </property>
-    <property>  
-        <name>ha.zookeeper.quorum</name>  
-        <value>192.168.40.151:2181,192.168.40.152:2181,192.168.40.203:2181</value>  
-    </property>
-    <property>
-        <name>ipc.client.connect.timeout</name>
-        <value>90000</value>
-    </property>
-</configuration>
diff --git a/src/main/resources/hdfs-site.xml b/src/main/resources/hdfs-site.xml
deleted file mode 100644
index e1408d2..0000000
--- a/src/main/resources/hdfs-site.xml
+++ /dev/null
@@ -1,142 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
-<!--
-  Licensed under the Apache License, Version 2.0 (the "License");
-  you may not use this file except in compliance with the License.
-  You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-  Unless required by applicable law or agreed to in writing, software
-  distributed under the License is distributed on an "AS IS" BASIS,
-  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  See the License for the specific language governing permissions and
-  limitations under the License. See accompanying LICENSE file.
--->
-
-<!-- Put site-specific property overrides in this file. -->
-
-<configuration>
-    <property>
-        <name>dfs.namenode.name.dir</name>
-        <value>file:/home/tsg/olap/hadoop/dfs/name</value>
-    </property>
-    <property>
-        <name>dfs.datanode.data.dir</name>
-        <value>file:/home/tsg/olap/hadoop/dfs/data</value>
-    </property>
-    <property>
-        <name>dfs.replication</name>
-        <value>2</value>
-    </property>
-    <property>
-	<name>dfs.webhdfs.enabled</name>
-	<value>true</value>
-    </property>
-    <property>
-        <name>dfs.permissions</name>
-        <value>false</value>
-    </property>	
-	<property>  
-        <name>dfs.permissions.enabled</name>  
-        <value>false</value>  
-    </property>  
-    <property>  
-        <name>dfs.nameservices</name>  
-        <value>ns1</value>  
-    </property>  
-    <property>  
-        <name>dfs.blocksize</name>  
-        <value>134217728</value>  
-    </property>  
-    <property>  
-        <name>dfs.ha.namenodes.ns1</name>  
-        <value>nn1,nn2</value>  
-    </property>  
-    <!-- nn1的RPC通信地址，nn1所在地址  -->  
-    <property>  
-        <name>dfs.namenode.rpc-address.ns1.nn1</name>  
-        <value>192.168.40.151:9000</value>  
-    </property>  
-    <!-- nn1的http通信地址，外部访问地址 -->  
-    <property>  
-        <name>dfs.namenode.http-address.ns1.nn1</name>  
-        <value>192.168.40.151:50070</value>  
-    </property>  
-    <!-- nn2的RPC通信地址，nn2所在地址 -->  
-    <property>  
-        <name>dfs.namenode.rpc-address.ns1.nn2</name>  
-        <value>192.168.40.152:9000</value>  
-    </property>  
-    <!-- nn2的http通信地址，外部访问地址 -->  
-    <property>  
-        <name>dfs.namenode.http-address.ns1.nn2</name>  
-        <value>192.168.40.152:50070</value>  
-    </property>  
-    <!-- 指定NameNode的元数据在JournalNode日志上的存放位置(一般和zookeeper部署在一起) -->  
-    <property>  
-        <name>dfs.namenode.shared.edits.dir</name>  
-        <value>qjournal://192.168.40.151:8485;192.168.40.152:8485;192.168.40.203:8485/ns1</value>  
-    </property>  
-    <!-- 指定JournalNode在本地磁盘存放数据的位置 -->  
-    <property>  
-        <name>dfs.journalnode.edits.dir</name>  
-        <value>/home/tsg/olap/hadoop/journal</value>  
-    </property>  
-    <!--客户端通过代理访问namenode，访问文件系统，HDFS 客户端与Active 节点通信的Java 类，使用其确定Active 节点是否活跃  -->  
-    <property>  
-        <name>dfs.client.failover.proxy.provider.ns1</name>  
-        <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>  
-    </property>  
-    <!--这是配置自动切换的方法，有多种使用方法，具体可以看官网，在文末会给地址，这里是远程登录杀死的方法  -->  
-    <property>  
-        <name>dfs.ha.fencing.methods</name>  
-        <value>sshfence</value>    
-        <value>shell(true)</value> 
-    </property>  
-    <!-- 这个是使用sshfence隔离机制时才需要配置ssh免登陆 -->  
-    <property>  
-        <name>dfs.ha.fencing.ssh.private-key-files</name>  
-        <value>/root/.ssh/id_rsa</value>  
-    </property>  
-    <!-- 配置sshfence隔离机制超时时间，这个属性同上，如果你是用脚本的方法切换，这个应该是可以不配置的 -->  
-    <property>  
-        <name>dfs.ha.fencing.ssh.connect-timeout</name>  
-        <value>30000</value>  
-    </property>  
-    <!-- 这个是开启自动故障转移，如果你没有自动故障转移，这个可以先不配 -->  
-    <property>  
-        <name>dfs.ha.automatic-failover.enabled</name>  
-        <value>true</value>  
-    </property>
-    <property>
-        <name>dfs.datanode.max.transfer.threads</name>
-        <value>8192</value>
-    </property>
-    <!-- namenode处理RPC请求线程数，增大该值资源占用不大 -->
-    <property>
-        <name>dfs.namenode.handler.count</name>
-        <value>30</value>
-    </property>
-    <!-- datanode处理RPC请求线程数，增大该值会占用更多内存 -->
-    <property>
-        <name>dfs.datanode.handler.count</name>
-        <value>40</value>
-    </property>
-    <!-- balance时可占用的带宽 -->
-    <property>
-        <name>dfs.balance.bandwidthPerSec</name>
-        <value>104857600</value>
-    </property>
-    <!-- 磁盘预留空间，该空间不会被hdfs占用，单位字节-->
-    <property>
-        <name>dfs.datanode.du.reserved</name>
-        <value>5368709120</value>
-    </property>
-    <!-- datanode与namenode连接超时时间，单位毫秒  2 * heartbeat.recheck.interval + 30000  -->
-    <property>
-	<name>heartbeat.recheck.interval</name>
-	<value>100000</value>
-    </property>
-</configuration>
-
diff --git a/src/main/resources/yarn-site.xml b/src/main/resources/yarn-site.xml
deleted file mode 100644
index 8a4b2fa..0000000
--- a/src/main/resources/yarn-site.xml
+++ /dev/null
@@ -1,196 +0,0 @@
-<?xml version="1.0"?>
-<!--
-            Licensed under the Apache License, Version 2.0 (the "License");
-  you may not use this file except in compliance with the License.
-  You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-  Unless required by applicable law or agreed to in writing, software
-  distributed under the License is distributed on an "AS IS" BASIS,
-  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-  See the License for the specific language governing permissions and
-  limitations under the License. See accompanying LICENSE file.
--->
-<configuration>
-    <property>
-        <name>yarn.nodemanager.aux-services</name>
-        <value>mapreduce_shuffle</value>
-    </property>
-        <property>
-       <name>yarn.resourcemanager.ha.enabled</name>
-       <value>true</value>
-    </property>
-    <!--声明两台resourcemanager的地址-->
-    <property>
-       <name>yarn.resourcemanager.cluster-id</name>
-       <value>rmcluster</value>
-    </property>
-    <property>
-       <name>yarn.resourcemanager.ha.rm-ids</name>
-       <value>rsm1,rsm2</value>
-    </property>
-    <!-- 配置rm1-->
-     <property>
-       <name>yarn.resourcemanager.hostname.rsm1</name>
-       <value>192.168.40.152</value>
-    </property>
-    <property>
-       <!--<name>yarn.resourcemanager.hostname.rm1</name>-->
-       <name>yarn.resourcemanager.address.rsm1</name>
-       <value>192.168.40.152:9916</value>
-    </property>
-    <property>
-           <name>yarn.resourcemanager.scheduler.address.rsm1</name>
-       <value>192.168.40.152:9917</value>
-    </property>
-    <property>
-           <name>yarn.resourcemanager.webapp.address.rsm1</name>
-       <value>192.168.40.152:9918</value>
-    </property>
-     <property>
-       <name>yarn.resourcemanager.admin.address.rsm1</name>
-       <value>192.168.40.152:9919</value>
-    </property>
-    <property>
-       <name>yarn.resourcemanager.resource-tracker.address.rsm1</name>
-       <value>192.168.40.152:9920</value>
-    </property>
-    <property>
-  <name>yarn.resourcemanager.ha.admin.address.rsm1</name>
-  <value>192.168.40.152:23142</value>
-    </property>
-    <!-- 配置rm2-->
-   
-    <property>
-       <name>yarn.resourcemanager.hostname.rsm2</name>
-       <value>192.168.40.203</value>
-    </property>
-    <property>
-       <!--<name>yarn.resourcemanager.hostname.rm1</name>-->
-       <name>yarn.resourcemanager.address.rsm2</name>
-       <value>192.168.40.203:9916</value>
-    </property>
-    <property>
-           <name>yarn.resourcemanager.scheduler.address.rsm2</name>
-       <value>192.168.40.203:9917</value>
-    </property>
-    <property>
-           <name>yarn.resourcemanager.webapp.address.rsm2</name>
-       <value>192.168.40.203:9918</value>
-    </property>
-    <property>
-       <name>yarn.resourcemanager.admin.address.rsm2</name>
-       <value>192.168.40.203:9919</value>
-    </property>
-    <property>
-       <name>yarn.resourcemanager.resource-tracker.address.rsm2</name>
-       <value>192.168.40.203:9920</value>
-    </property>
-    <property>
-  <name>yarn.resourcemanager.ha.admin.address.rsm2</name>
-  <value>192.168.40.203:23142</value>
-</property>
-    <!--指定zookeeper集群的地址-->
-    <property>
-       <name>yarn.resourcemanager.zk-address</name>
-        <value>192.168.40.151:2181,192.168.40.152:2181,192.168.40.203:2181</value>
-    </property>
-    <!--启用自动恢复，当任务进行一半，rm坏掉，就要启动自动恢复，默认是false-->
-    <property>
-       <name>yarn.resourcemanager.recovery.enabled</name>
-       <value>true</value>
-    </property>
-    <!--启动nm自动恢复，当集群重启，container开启自动恢复，保障任务的可靠性，默认为false，启动该配置需开启下面配置 -->
- <property>
-       <name>yarn.nodemanager.recovery.enabled</name>
-       <value>true</value>
-    </property>
-   <!--开启nm故障恢复后，nm元数据存储路径 -->
-   <property>
-       <name>yarn.nodemanager.recovery.dir</name>
-       <value>/home/tsg/olap/hadoop-2.7.1/yarn</value>
-    </property>
-   <!--启用nm恢复时监控功能，开启后不会去尝试清理container，默认false -->  
- <property>
-      <name>yarn.nodemanager.recovery.supervised</name>
-      <value>true</value>
-   </property>
-  <!--配置nm可用的RPC地址，默认${yarn.nodemanager.hostname}:0，为临时端口。集群重启后，nm与rm连接的端口会变化，这里指定端口，保障nm restart功能 -->
-<property>
-      <name>yarn.nodemanager.address</name>
-      <value>${yarn.nodemanager.hostname}:9923</value>
-   </property>
-        <property>
-       <name>yarn.resourcemanager.store.class</name>
-       <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
-    </property>
-    <property>
-        <name>yarn.nodemanager.resource.memory-mb</name>
-        <value>30720</value>
-    </property> 
-    <property>
-        <name>yarn.scheduler.minimum-allocation-mb</name>
-        <value>1024</value>
-    </property>
-
-    <property>
-        <name>yarn.scheduler.maximum-allocation-mb</name>
-        <value>30720</value>
-    </property>
-    <!--开启日志聚合 -->
-    <property>
-        <name>yarn.log-aggregation-enable</name>
-        <value>true</value>
-    </property>
-    <property>
-        <name>yarn.nodemanager.heartbeat-interval-ms</name>
-        <value>3000</value>
-    </property>
-   <!--日志保留7天 -->
-    <property>
-        <name>yarn.log-aggregation.retain-seconds</name>
-        <value>604800</value>
-    </property>
-    <property>
-        <name>yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds</name>
-        <value>3600</value>
-    </property>
-    <property>
-        <name>yarn.nodemanager.remote-app-log-dir</name>
-        <value>/tmp/logs</value>
-    </property>
-   <property>
-        <name>yarn.nodemanager.resource.cpu-vcores</name>
-        <value>14</value>
-    </property>
-    <property>
-        <name>yarn.scheduler.minimum-allocation-vcores</name>
-        <value>1</value>
-    </property>
-    <property>
-        <name>yarn.scheduler.maximum-allocation-vcores</name>
-        <value>14</value>
-    </property>
-    <property>
-        <name>yarn.nodemanager.vmem-check-enabled</name>
-        <value>false</value>
-    </property>
-   <property>
-        <name>yarn.nodemanager.pmem-check-enabled</name>
-        <value>false</value>
-    </property>
-    <property>
-        <name>yarn.nodemanager.disk-health-checker.enable</name>
-        <value>false</value>
-    </property>
-    <!--ApplicationMaster重启次数，配置HA后默认为2-->
-    <property>
-        <name>yarn.resourcemanager.am.max-attempts</name>
-        <value>10000</value>
-    </property>
-    <property>
-        <name>yarn.log.server.url</name>
-        <value>http://bigdata-151:19888/jobhistory/logs</value>
-    </property>
-</configuration>
diff --git a/src/test/java/com/zdjizhi/common/HbaseTest.java b/src/test/java/com/zdjizhi/common/HbaseTest.java
index f5e1961..3467cc3 100644
--- a/src/test/java/com/zdjizhi/common/HbaseTest.java
+++ b/src/test/java/com/zdjizhi/common/HbaseTest.java
@@ -19,12 +19,12 @@ public class HbaseTest {
     public static void main(String[] args) throws IOException {
         org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
 
-        config.set("hbase.zookeeper.quorum", CommonConfig.HBASE_ZOOKEEPER_QUORUM);
+        config.set("hbase.zookeeper.quorum", FlowWriteConfig.HBASE_ZOOKEEPER_QUORUM);
         config.set("hbase.client.retries.number", "3");
         config.set("hbase.bulkload.retries.number", "3");
         config.set("zookeeper.recovery.retry", "3");
-        config.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, CommonConfig.HBASE_CLIENT_OPERATION_TIMEOUT);
-        config.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, CommonConfig.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
+        config.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, FlowWriteConfig.HBASE_CLIENT_OPERATION_TIMEOUT);
+        config.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD, FlowWriteConfig.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);
 
         TableName tableName  = TableName.valueOf("dos_test");
         Connection conn = ConnectionFactory.createConnection(config);
diff --git a/src/test/java/com/zdjizhi/etl/DosDetectionTest.java b/src/test/java/com/zdjizhi/etl/DosDetectionTest.java
index dcd3656..4428dc0 100644
--- a/src/test/java/com/zdjizhi/etl/DosDetectionTest.java
+++ b/src/test/java/com/zdjizhi/etl/DosDetectionTest.java
@@ -1,12 +1,12 @@
 package com.zdjizhi.etl;
 
+import com.geedgenetworks.utils.StringUtil;
 import com.zdjizhi.common.DosDetectionThreshold;
 import com.zdjizhi.common.DosEventLog;
 import com.zdjizhi.common.DosSketchLog;
-import com.zdjizhi.utils.IpUtils;
-import com.zdjizhi.utils.NacosUtils;
+
+import com.zdjizhi.utils.IpLookupUtils;
 import com.zdjizhi.utils.SnowflakeId;
-import com.zdjizhi.utils.StringUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.junit.Test;
 
@@ -34,14 +34,14 @@ public class DosDetectionTest {
         serverIpList.add("192.168.50.1/24");
         serverIpList.add("FC::12:0:0/54");
         serverIpList.add("FC::12:0:0");
-        dosDetectionThreshold.setProfileId(4437);
-        dosDetectionThreshold.setAttackType("DNS Flood");
-        dosDetectionThreshold.setServerIpList(serverIpList);
-        dosDetectionThreshold.setSessionsPerSec(1);
-        dosDetectionThreshold.setPacketsPerSec(1);
-        dosDetectionThreshold.setBitsPerSec(100000);
-        dosDetectionThreshold.setIsValid(1);
-        dosDetectionThreshold.setSuperiorIds(new Integer[]{5,4,12,27});
+        dosDetectionThreshold.setProfile_id(4437);
+        dosDetectionThreshold.setAttack_type("DNS Flood");
+        dosDetectionThreshold.setServer_ip_list(serverIpList);
+        dosDetectionThreshold.setSessions_per_sec(1);
+        dosDetectionThreshold.setPackets_per_sec(1);
+        dosDetectionThreshold.setBits_per_sec(100000);
+        dosDetectionThreshold.setIs_valid(1);
+        dosDetectionThreshold.setSuperior_ids(new Integer[]{5,4,12,27});
 
 
         DosSketchLog dosSketchLog = new DosSketchLog ();
@@ -54,9 +54,9 @@ public class DosDetectionTest {
         dosSketchLog.setSource_ip("45.170.244.25");
         dosSketchLog.setDestination_ip("24.152.57.56");
         //静态阈值获取
-        long sessionBase = dosDetectionThreshold.getSessionsPerSec();
-        long pktBase=dosDetectionThreshold.getPacketsPerSec();
-        long bitBase=dosDetectionThreshold.getBitsPerSec();
+        long sessionBase = dosDetectionThreshold.getSessions_per_sec();
+        long pktBase=dosDetectionThreshold.getPackets_per_sec();
+        long bitBase=dosDetectionThreshold.getBits_per_sec();
         //基于速率进行计算
         long diffSession = dosSketchLog.getSketch_sessions() - sessionBase;
         long diffPkt = dosSketchLog.getSketch_packets() - pktBase;
@@ -69,15 +69,15 @@ public class DosDetectionTest {
         long profileId = 0;
         DosEventLog result =null;
         if (diffSessionPercent >= diffPktPercent && diffSessionPercent >= diffBitPercent){
-            profileId = dosDetectionThreshold.getProfileId();
+            profileId = dosDetectionThreshold.getProfile_id();
             result= getDosEventLog(dosSketchLog, sessionBase, diffSession, profileId, STATIC_CONDITION_TYPE, SESSIONS_TAG);
             System.out.println(result);
         }else if (diffPktPercent >= diffSessionPercent && diffPktPercent >= diffBitPercent){
-            profileId = dosDetectionThreshold.getProfileId();
+            profileId = dosDetectionThreshold.getProfile_id();
             result = getDosEventLog(dosSketchLog, pktBase, diffPkt,profileId, STATIC_CONDITION_TYPE, PACKETS_TAG);
             System.out.println(result);
         }else if (diffBitPercent >= diffPktPercent && diffBitPercent >= diffSessionPercent){
-            profileId = dosDetectionThreshold.getProfileId();
+            profileId = dosDetectionThreshold.getProfile_id();
             result = getDosEventLog(dosSketchLog, bitBase, diffByte, profileId, STATIC_CONDITION_TYPE, BITS_TAG);
             System.out.println(result);
         }
@@ -174,7 +174,7 @@ public class DosDetectionTest {
                 String[] ipArr = sourceIpList.split(",");
                 HashSet<String> countrySet = new HashSet<>();
                 for (String ip : ipArr) {
-                    String country = IpUtils.ipLookup.countryLookup(ip);
+                    String country = IpLookupUtils.getCountryLookup(ip);
                     if (StringUtil.isNotBlank(country)){
                         countrySet.add(country);
                     }