getJobList() {
+ return jobList;
}
+
/**
* 通过获取String类型的网关schema链接来获取map，用于生成一个Object类型的对象
+ *
+ * // * @param http 网关schema地址
*
- * @param http 网关schema地址
* @return 用于反射生成schema类型的对象的一个map集合
*/
- public static HashMap getMapFromHttp(String http) {
+ public static HashMap getMapFromHttp(String schema) {
HashMap map = new HashMap<>(16);
- String schema = HttpClientUtil.requestByGetMethod(http);
- Object data = JSON.parseObject(schema).get("data");
-
//获取fields，并转化为数组，数组的每个元素都是一个name doc type
- JSONObject schemaJson = JSON.parseObject(data.toString());
+ JSONObject schemaJson = JSON.parseObject(schema);
JSONArray fields = (JSONArray) schemaJson.get("fields");
for (Object field : fields) {
@@ -239,6 +288,7 @@ public class JsonParseUtil {
/**
* 删除schema内指定的无效字段（jackson）
+ *
* @param jsonMap
*/
public static void dropJsonField(Map jsonMap) {
@@ -247,31 +297,17 @@ public class JsonParseUtil {
}
}
- /**
- * 删除schema内指定的无效字段（fastjson）
- * @param jsonMap
- */
- public static void dropJsonField(JSONObject jsonMap) {
- for (String field : dropList) {
- jsonMap.remove(field);
- }
- }
-
/**
* 根据http链接获取schema，解析之后返回一个任务列表 (useList toList funcList paramlist)
*
- * @param http 网关url
+ * @param schema 网关url
* @return 任务列表
*/
- public static ArrayList getJobListFromHttp(String http) {
+ public static ArrayList getJobListFromHttp(String schema) {
ArrayList list = new ArrayList<>();
- String schema = HttpClientUtil.requestByGetMethod(http);
- //解析data
- Object data = JSON.parseObject(schema).get("data");
-
//获取fields，并转化为数组，数组的每个元素都是一个name doc type
- JSONObject schemaJson = JSON.parseObject(data.toString());
+ JSONObject schemaJson = JSON.parseObject(schema);
JSONArray fields = (JSONArray) schemaJson.get("fields");
for (Object field : fields) {
diff --git a/src/main/java/com/zdjizhi/utils/json/JsonTypeUtils.java b/src/main/java/com/zdjizhi/utils/json/JsonTypeUtils.java
index 0b6bc1e..63af9d5 100644
--- a/src/main/java/com/zdjizhi/utils/json/JsonTypeUtils.java
+++ b/src/main/java/com/zdjizhi/utils/json/JsonTypeUtils.java
@@ -2,6 +2,11 @@ package com.zdjizhi.utils.json;
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.alibaba.nacos.api.exception.NacosException;
import com.zdjizhi.common.FlowWriteConfig;
import com.zdjizhi.utils.JsonMapper;
import com.zdjizhi.utils.exception.FlowWriteException;
@@ -9,6 +14,11 @@ import com.zdjizhi.utils.exception.FlowWriteException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+import static com.zdjizhi.utils.json.JsonParseUtil.getJobListFromHttp;
+import static com.zdjizhi.utils.json.JsonParseUtil.getMapFromHttp;
/**
* @author qidaijie
@@ -17,70 +27,26 @@ import java.util.Map;
* @date 2021/7/1217:34
*/
public class JsonTypeUtils {
- private static final Log logger = LogFactory.get();
- /**
- * 在内存中加载反射类用的map
- */
- private static HashMap map = JsonParseUtil.getMapFromHttp(FlowWriteConfig.SCHEMA_HTTP);
-
- /**
- * 类型转换
- *
- * @param jsonMap 原始日志map
- */
- public static Map typeTransform(Map jsonMap) throws RuntimeException {
- JsonParseUtil.dropJsonField(jsonMap);
- HashMap tmpMap = new HashMap<>(192);
- for (String key : jsonMap.keySet()) {
- if (map.containsKey(key)) {
- String simpleName = map.get(key).getSimpleName();
- switch (simpleName) {
- case "String":
- tmpMap.put(key, checkString(jsonMap.get(key)));
- break;
- case "Integer":
- tmpMap.put(key, getIntValue(jsonMap.get(key)));
- break;
- case "long":
- tmpMap.put(key, checkLongValue(jsonMap.get(key)));
- break;
- case "List":
- tmpMap.put(key, checkArray(jsonMap.get(key)));
- break;
- case "Map":
- tmpMap.put(key, checkObject(jsonMap.get(key)));
- break;
- case "double":
- tmpMap.put(key, checkDouble(jsonMap.get(key)));
- break;
- default:
- tmpMap.put(key, checkString(jsonMap.get(key)));
- }
- }
- }
- return tmpMap;
- }
-
/**
* String 类型检验转换方法
*
* @param value json value
* @return String value
*/
- private static String checkString(Object value) {
+ static String checkString(Object value) {
if (value == null) {
return null;
}
- if (value instanceof Map){
+ if (value instanceof Map) {
return JsonMapper.toJsonString(value);
}
- if (value instanceof List){
+ if (value instanceof List) {
return JsonMapper.toJsonString(value);
}
- return value.toString();
+ return value.toString();
}
/**
@@ -89,7 +55,7 @@ public class JsonTypeUtils {
* @param value json value
* @return List value
*/
- private static Map checkObject(Object value) {
+ static Map checkObject(Object value) {
if (value == null) {
return null;
}
@@ -107,7 +73,7 @@ public class JsonTypeUtils {
* @param value json value
* @return List value
*/
- private static List checkArray(Object value) {
+ static List checkArray(Object value) {
if (value == null) {
return null;
}
@@ -119,27 +85,19 @@ public class JsonTypeUtils {
throw new FlowWriteException("can not cast to List, value : " + value);
}
- private static Long checkLong(Object value) {
- if (value == null) {
- return null;
- }
-
- return TypeUtils.castToLong(value);
- }
-
/**
* long 类型检验转换方法,若为空返回基础值
*
* @param value json value
* @return Long value
*/
- private static long checkLongValue(Object value) {
+ static long checkLongValue(Object value) {
Long longVal = TypeUtils.castToLong(value);
+
if (longVal == null) {
return 0L;
}
-// return longVal.longValue();
return longVal;
}
@@ -149,7 +107,7 @@ public class JsonTypeUtils {
* @param value json value
* @return Double value
*/
- private static Double checkDouble(Object value) {
+ static Double checkDouble(Object value) {
if (value == null) {
return null;
}
@@ -158,29 +116,18 @@ public class JsonTypeUtils {
}
- private static Integer checkInt(Object value) {
- if (value == null) {
- return null;
- }
-
- return TypeUtils.castToInt(value);
- }
-
-
/**
* int 类型检验转换方法,若为空返回基础值
*
* @param value json value
* @return int value
*/
- private static int getIntValue(Object value) {
+ static int getIntValue(Object value) {
Integer intVal = TypeUtils.castToInt(value);
if (intVal == null) {
return 0;
}
-
-// return intVal.intValue();
return intVal;
}
diff --git a/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java b/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java
index 078c2fe..f3d979b 100644
--- a/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java
+++ b/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java
@@ -2,8 +2,12 @@ package com.zdjizhi.utils.kafka;
import com.zdjizhi.common.FlowWriteConfig;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
@@ -25,9 +29,7 @@ public class KafkaConsumer {
properties.put("session.timeout.ms", FlowWriteConfig.SESSION_TIMEOUT_MS);
properties.put("max.poll.records", FlowWriteConfig.MAX_POLL_RECORDS);
properties.put("max.partition.fetch.bytes", FlowWriteConfig.MAX_PARTITION_FETCH_BYTES);
- properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
- properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "10");
+ properties.put("partition.discovery.interval.ms", "10000");
CertUtils.chooseCert(FlowWriteConfig.SOURCE_KAFKA_SERVERS, properties);
return properties;
@@ -42,7 +44,10 @@ public class KafkaConsumer {
FlinkKafkaConsumer> kafkaConsumer = new FlinkKafkaConsumer<>(FlowWriteConfig.SOURCE_KAFKA_TOPIC,
new TimestampDeserializationSchema(), createConsumerConfig());
- kafkaConsumer.setCommitOffsetsOnCheckpoints(false);
+ //随着checkpoint提交，将offset提交到kafka
+ kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
+
+ //从消费组当前的offset开始消费
kafkaConsumer.setStartFromGroupOffsets();
return kafkaConsumer;
@@ -57,7 +62,7 @@ public class KafkaConsumer {
FlinkKafkaConsumer kafkaConsumer = new FlinkKafkaConsumer<>(FlowWriteConfig.SOURCE_KAFKA_TOPIC,
new SimpleStringSchema(), createConsumerConfig());
- kafkaConsumer.setCommitOffsetsOnCheckpoints(false);
+ kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
kafkaConsumer.setStartFromGroupOffsets();
return kafkaConsumer;
diff --git a/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java b/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
index f2f399d..843028b 100644
--- a/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
+++ b/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
@@ -3,8 +3,6 @@ package com.zdjizhi.utils.kafka;
import com.zdjizhi.common.FlowWriteConfig;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.config.SslConfigs;
import java.util.Optional;
import java.util.Properties;
@@ -39,12 +37,14 @@ public class KafkaProducer {
FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer(
FlowWriteConfig.SINK_KAFKA_TOPIC,
new SimpleStringSchema(),
- createProducerConfig(), Optional.empty());
+ //sink与所有分区建立连接，轮询写入；
+ createProducerConfig(),
+ Optional.empty());
- kafkaProducer.setLogFailuresOnly(false);
-
-// kafkaProducer.setWriteTimestampToKafka(true);
+ //允许producer记录失败日志而不是捕获和抛出它们
+ kafkaProducer.setLogFailuresOnly(true);
return kafkaProducer;
}
+
}
diff --git a/src/main/java/com/zdjizhi/utils/kafka/TimestampDeserializationSchema.java b/src/main/java/com/zdjizhi/utils/kafka/TimestampDeserializationSchema.java
index e978369..920ffab 100644
--- a/src/main/java/com/zdjizhi/utils/kafka/TimestampDeserializationSchema.java
+++ b/src/main/java/com/zdjizhi/utils/kafka/TimestampDeserializationSchema.java
@@ -2,6 +2,7 @@ package com.zdjizhi.utils.kafka;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
+import com.zdjizhi.common.FlowWriteConfig;
import com.zdjizhi.utils.JsonMapper;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
@@ -17,7 +18,11 @@ import java.util.Map;
*/
public class TimestampDeserializationSchema implements KafkaDeserializationSchema {
private static final Log logger = LogFactory.get();
- private final String ENCODING = "UTF8";
+
+ @Override
+ public TypeInformation getProducedType() {
+ return TypeInformation.of(Map.class);
+ }
@Override
public boolean isEndOfStream(Object nextElement) {
@@ -25,11 +30,12 @@ public class TimestampDeserializationSchema implements KafkaDeserializationSchem
}
@Override
+ @SuppressWarnings("unchecked")
public Map deserialize(ConsumerRecord record) throws Exception {
if (record != null) {
try {
long timestamp = record.timestamp() / 1000;
- String value = new String((byte[]) record.value(), ENCODING);
+ String value = new String((byte[]) record.value(), FlowWriteConfig.ENCODING);
Map json = (Map) JsonMapper.fromJsonString(value, Map.class);
json.put("common_ingestion_time", timestamp);
return json;
@@ -39,9 +45,4 @@ public class TimestampDeserializationSchema implements KafkaDeserializationSchem
}
return null;
}
-
- @Override
- public TypeInformation getProducedType() {
- return TypeInformation.of(Map.class);
- }
}
diff --git a/src/main/java/com/zdjizhi/utils/system/FlowWriteConfigurations.java b/src/main/java/com/zdjizhi/utils/system/FlowWriteConfigurations.java
index 08fa29b..d793628 100644
--- a/src/main/java/com/zdjizhi/utils/system/FlowWriteConfigurations.java
+++ b/src/main/java/com/zdjizhi/utils/system/FlowWriteConfigurations.java
@@ -1,8 +1,14 @@
package com.zdjizhi.utils.system;
+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 com.zdjizhi.utils.StringUtil;
import java.io.IOException;
+import java.io.StringReader;
import java.util.Locale;
import java.util.Properties;
@@ -25,7 +31,6 @@ public final class FlowWriteConfigurations {
} else {
return null;
}
-
}
public static Integer getIntProperty(Integer type, String key) {
diff --git a/src/test/java/com/zdjizhi/EncryptorTest.java b/src/test/java/com/zdjizhi/EncryptorTest.java
new file mode 100644
index 0000000..170086c
--- /dev/null
+++ b/src/test/java/com/zdjizhi/EncryptorTest.java
@@ -0,0 +1,35 @@
+package com.zdjizhi;
+
+import org.jasypt.encryption.pbe.StandardPBEStringEncryptor;
+import org.junit.Test;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2022/3/1610:55
+ */
+public class EncryptorTest {
+
+
+ @Test
+ public void passwordTest(){
+ StandardPBEStringEncryptor encryptor = new StandardPBEStringEncryptor();
+ // 配置加密解密的密码/salt值
+ encryptor.setPassword("galaxy");
+ // 对"raw_password"进行加密：S5kR+Y7CI8k7MaecZpde25yK8NKUnd6p
+ String pin = "galaxy2019";
+ String encPin = encryptor.encrypt(pin);
+ String user = "admin";
+ String encUser = encryptor.encrypt(user);
+ System.out.println(encPin);
+ System.out.println(encUser);
+ // 再进行解密：raw_password
+ String rawPwd = encryptor.decrypt("6MleDyA3Z73HSaXiKsDJ2k7Ys8YWLhEJ");
+ String rawUser = encryptor.decrypt("nsyGpHKGFA4KW0zro9MDdw==");
+
+ System.out.println("The username is: "+rawPwd);
+ System.out.println("The pin is: "+rawUser);
+ }
+
+}
diff --git a/src/test/java/com/zdjizhi/FunctionTest.java b/src/test/java/com/zdjizhi/FunctionTest.java
new file mode 100644
index 0000000..2dd5837
--- /dev/null
+++ b/src/test/java/com/zdjizhi/FunctionTest.java
@@ -0,0 +1,52 @@
+package com.zdjizhi;
+
+import com.zdjizhi.common.FlowWriteConfig;
+import com.zdjizhi.utils.IpLookupV2;
+import com.zdjizhi.utils.general.CityHash;
+import org.junit.Test;
+
+import java.math.BigInteger;
+import java.util.Calendar;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2021/11/611:38
+ */
+public class FunctionTest {
+
+ private static IpLookupV2 ipLookup = new IpLookupV2.Builder(false)
+ .loadDataFileV4(FlowWriteConfig.TOOLS_LIBRARY + "ip_v4_built_in.mmdb")
+ .loadDataFileV6(FlowWriteConfig.TOOLS_LIBRARY + "ip_v6_built_in.mmdb")
+ .loadDataFilePrivateV4(FlowWriteConfig.TOOLS_LIBRARY + "ip_v4_user_defined.mmdb")
+ .loadDataFilePrivateV6(FlowWriteConfig.TOOLS_LIBRARY + "ip_v6_user_defined.mmdb")
+ .loadAsnDataFile(FlowWriteConfig.TOOLS_LIBRARY + "asn_v4.mmdb")
+ .loadAsnDataFileV6(FlowWriteConfig.TOOLS_LIBRARY + "asn_v6.mmdb")
+ .build();
+
+ @Test
+ public void CityHashTest() {
+
+ byte[] dataBytes = String.valueOf(613970406986188816L).getBytes();
+ long hashValue = CityHash.CityHash64(dataBytes, 0, dataBytes.length);
+ String decimalValue = Long.toUnsignedString(hashValue, 10);
+ BigInteger result = new BigInteger(decimalValue);
+ System.out.println(result);
+ }
+
+ @Test
+ public void ipLookupTest() {
+ String ip = "61.144.36.144";
+ System.out.println(ipLookup.cityLookupDetail(ip));
+ System.out.println(ipLookup.countryLookup(ip));
+ }
+
+ @Test
+ public void timestampTest(){
+ Calendar cal = Calendar.getInstance();
+ Long utcTime=cal.getTimeInMillis();
+ System.out.println(utcTime);
+ System.out.println(System.currentTimeMillis());
+ }
+}
diff --git a/src/test/java/com/zdjizhi/HBaseTest.java b/src/test/java/com/zdjizhi/HBaseTest.java
new file mode 100644
index 0000000..5f94e32
--- /dev/null
+++ b/src/test/java/com/zdjizhi/HBaseTest.java
@@ -0,0 +1,54 @@
+package com.zdjizhi;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2021/12/310:42
+ */
+public class HBaseTest {
+
+ @Test
+ public void getColumn() {
+ // 管理Hbase的配置信息
+ Configuration configuration = HBaseConfiguration.create();
+ // 设置zookeeper节点
+ configuration.set("hbase.zookeeper.quorum", "192.168.44.11:2181");
+ configuration.set("hbase.client.retries.number", "3");
+ configuration.set("hbase.bulkload.retries.number", "3");
+ configuration.set("zookeeper.recovery.retry", "3");
+ try {
+ Connection connection = ConnectionFactory.createConnection(configuration);
+ Table table = connection.getTable(TableName.valueOf("tsg_galaxy:relation_framedip_account"));
+ Scan scan2 = new Scan();
+ ResultScanner scanner = table.getScanner(scan2);
+ for (Result result : scanner) {
+ int acctStatusType;
+ boolean hasType = result.containsColumn(Bytes.toBytes("radius"), Bytes.toBytes("acct_status_type"));
+ if (hasType) {
+ acctStatusType = Bytes.toInt(result.getValue(Bytes.toBytes("radius"), Bytes.toBytes("acct_status_type")));
+ } else {
+ acctStatusType = 3;
+ }
+ String framedIp = Bytes.toString(result.getValue(Bytes.toBytes("radius"), Bytes.toBytes("framed_ip")));
+ String account = Bytes.toString(result.getValue(Bytes.toBytes("radius"), Bytes.toBytes("account")));
+ System.out.println("status" + acctStatusType + "key:" + framedIp + "value:" + account);
+// System.out.println(Arrays.toString(result.getValue(Bytes.toBytes("radius"), Bytes.toBytes("acct_status_type"))));
+ }
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+}
diff --git a/src/test/java/com/zdjizhi/KafkaTest.java b/src/test/java/com/zdjizhi/KafkaTest.java
deleted file mode 100644
index 4b034a3..0000000
--- a/src/test/java/com/zdjizhi/KafkaTest.java
+++ /dev/null
@@ -1,55 +0,0 @@
-package com.zdjizhi;
-
-import cn.hutool.log.Log;
-import cn.hutool.log.LogFactory;
-import org.apache.kafka.clients.producer.*;
-import org.apache.kafka.common.config.SslConfigs;
-
-import java.util.Properties;
-
-/**
- * @author qidaijie
- * @Package com.zdjizhi
- * @Description:
- * @date 2021/8/217:39
- */
-public class KafkaTest {
- private static final Log logger = LogFactory.get();
-
- public static void main(String[] args) {
- Properties properties = new Properties();
- properties.put("bootstrap.servers", "192.168.44.12:9091");
- properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- properties.put("acks", "1");
-// properties.put("retries", DefaultProConfig.RETRIES);
-// properties.put("linger.ms", DefaultProConfig.LINGER_MS);
-// properties.put("request.timeout.ms", DefaultProConfig.REQUEST_TIMEOUT_MS);
-// properties.put("batch.size", DefaultProConfig.BATCH_SIZE);
-// properties.put("buffer.memory", DefaultProConfig.BUFFER_MEMORY);
-// properties.put("max.request.size", DefaultProConfig.MAX_REQUEST_SIZE);
-
- properties.put("security.protocol", "SSL");
-// properties.put("ssl.keystore.location", "D:\\K18-Phase2\\tsgSpace\\dat\\kafka\\client.keystore.jks");
- properties.put("ssl.keystore.location", "D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\keystore.jks");
- properties.put("ssl.keystore.password", "galaxy2019");
-// properties.put("ssl.truststore.location", "D:\\K18-Phase2\\tsgSpace\\dat\\kafka\\client.truststore.jks");
- properties.put("ssl.truststore.location", "D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\truststore.jks");
- properties.put("ssl.truststore.password", "galaxy2019");
- properties.put("ssl.key.password", "galaxy2019");
- properties.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
-
- Producer producer = new KafkaProducer(properties);
-
- producer.send(new ProducerRecord<>("test", "hello!"), new Callback() {
- @Override
- public void onCompletion(RecordMetadata metadata, Exception exception) {
- if (exception != null) {
- logger.error("写入test出现异常", exception);
- }
- }
- });
-
- producer.close();
- }
-}
diff --git a/src/test/java/com/zdjizhi/LocationTest.java b/src/test/java/com/zdjizhi/LocationTest.java
deleted file mode 100644
index e7b2d15..0000000
--- a/src/test/java/com/zdjizhi/LocationTest.java
+++ /dev/null
@@ -1,28 +0,0 @@
-package com.zdjizhi;
-
-import com.zdjizhi.common.FlowWriteConfig;
-import com.zdjizhi.utils.IpLookup;
-import org.junit.Test;
-
-/**
- * @author qidaijie
- * @Package com.zdjizhi
- * @Description:
- * @date 2021/8/1811:34
- */
-public class LocationTest {
- private static IpLookup ipLookup = new IpLookup.Builder(false)
- .loadDataFileV4("D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\ip_v4.mmdb")
- .loadDataFileV6("D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\ip_v6.mmdb")
- .loadDataFilePrivateV4("D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\ip_private_v4.mmdb")
- .loadDataFilePrivateV6("D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\ip_private_v6.mmdb")
- .build();
-
- @Test
- public void IpLocationTest() {
- System.out.println(ipLookup.cityLookupDetail("24.241.112.0"));
- System.out.println(ipLookup.cityLookupDetail("1.1.1.1"));
- System.out.println(ipLookup.cityLookupDetail("192.168.50.58"));
- System.out.println(ipLookup.cityLookupDetail("2600:1700:9010::"));
- }
-}
diff --git a/src/test/java/com/zdjizhi/nacos/NacosTest.java b/src/test/java/com/zdjizhi/nacos/NacosTest.java
new file mode 100644
index 0000000..c2b6267
--- /dev/null
+++ b/src/test/java/com/zdjizhi/nacos/NacosTest.java
@@ -0,0 +1,86 @@
+package com.zdjizhi.nacos;
+
+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.zdjizhi.common.FlowWriteConfig;
+import com.zdjizhi.utils.system.FlowWriteConfigurations;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi
+ * @Description:
+ * @date 2022/3/1016:58
+ */
+public class NacosTest {
+ private static Properties properties = new Properties();
+
+ @Test
+ public void getProperties() {
+ properties.setProperty(PropertyKeyConst.SERVER_ADDR, "192.168.44.12:8848");
+ properties.setProperty(PropertyKeyConst.NAMESPACE, "flink");
+ properties.setProperty(PropertyKeyConst.USERNAME, "nacos");
+ properties.setProperty(PropertyKeyConst.PASSWORD, "nacos");
+ }
+
+
+ @Test
+ public void GetConfigurationTest() {
+ try {
+ getProperties();
+ ConfigService configService = NacosFactory.createConfigService(properties);
+ String content = configService.getConfig("topology_common_config.properties", "Galaxy", 5000);
+ Properties nacosConfigMap = new Properties();
+ nacosConfigMap.load(new StringReader(content));
+ System.out.println(nacosConfigMap.getProperty("source.kafka.servers"));
+ System.out.println(nacosConfigMap.getProperty("schema.http"));
+ } catch (NacosException | IOException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+
+ }
+
+ @Test
+ public void ListenerConfigurationTest() {
+ getProperties();
+ ConfigService configService = null;
+ try {
+ configService = NacosFactory.createConfigService(properties);
+ String content = configService.getConfig("ETL-SESSION-RECORD-COMPLETED", "etl", 5000);
+
+ Properties nacosConfigMap = new Properties();
+ nacosConfigMap.load(new StringReader(content));
+ System.out.println(nacosConfigMap.getProperty("source.kafka.servers"));
+
+ configService.addListener("ETL-SESSION-RECORD-COMPLETED", "etl", new Listener() {
+ @Override
+ public Executor getExecutor() {
+ return null;
+ }
+
+ @Override
+ public void receiveConfigInfo(String configMsg) {
+ try {
+ Properties nacosConfigMap = new Properties();
+ nacosConfigMap.load(new StringReader(configMsg));
+ System.out.println(nacosConfigMap.getProperty("source.kafka.servers"));
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ });
+ } catch (NacosException | IOException e) {
+ e.printStackTrace();
+ }
+ }
+}
diff --git a/src/test/java/com/zdjizhi/nacos/SchemaListener.java b/src/test/java/com/zdjizhi/nacos/SchemaListener.java
new file mode 100644
index 0000000..b64d9eb
--- /dev/null
+++ b/src/test/java/com/zdjizhi/nacos/SchemaListener.java
@@ -0,0 +1,138 @@
+package com.zdjizhi.nacos;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.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.zdjizhi.common.FlowWriteConfig;
+import com.zdjizhi.utils.JsonMapper;
+import com.zdjizhi.utils.StringUtil;
+import com.zdjizhi.utils.json.JsonParseUtil;
+import com.zdjizhi.utils.json.JsonTypeUtils;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+/**
+ * @author qidaijie
+ * @Package com.zdjizhi.nacos
+ * @Description:
+ * @date 2022/3/1714:57
+ */
+public class SchemaListener {
+
+ private static Properties properties = new Properties();
+ private static ArrayList jobList;
+
+
+ static {
+ properties.setProperty(PropertyKeyConst.SERVER_ADDR, "192.168.44.12:8848");
+ properties.setProperty(PropertyKeyConst.NAMESPACE, "flink");
+ properties.setProperty(PropertyKeyConst.USERNAME, "nacos");
+ properties.setProperty(PropertyKeyConst.PASSWORD, "nacos");
+
+ try {
+ ConfigService configService = NacosFactory.createConfigService(properties);
+ String dataId = "session_record.json";
+ String group = "Galaxy";
+ jobList = getJobListFromHttp(configService.getConfig(dataId, group, 5000));
+ configService.addListener(dataId, group, new Listener() {
+ @Override
+ public Executor getExecutor() {
+ return null;
+ }
+
+ @Override
+ public void receiveConfigInfo(String configMsg) {
+ jobList = getJobListFromHttp(configMsg);
+ }
+ });
+ } catch (NacosException e) {
+ e.printStackTrace();
+ }
+ }
+
+
+ @SuppressWarnings("unchecked")
+ public static void dealCommonMessage() {
+
+ System.out.println(Arrays.toString(jobList.get(0)));
+
+ }
+
+ /**
+ * 根据http链接获取schema，解析之后返回一个任务列表 (useList toList funcList paramlist)
+ *
+ * @return 任务列表
+ */
+ public static ArrayList getJobListFromHttp(String schema) {
+ ArrayList list = new ArrayList<>();
+
+ //解析data
+// Object data = JSON.parseObject(schema).get("data");
+
+ //获取fields，并转化为数组，数组的每个元素都是一个name doc type
+ JSONObject schemaJson = JSON.parseObject(schema);
+ JSONArray fields = (JSONArray) schemaJson.get("fields");
+
+ for (Object field : fields) {
+
+ if (JSON.parseObject(field.toString()).containsKey("doc")) {
+ Object doc = JSON.parseObject(field.toString()).get("doc");
+
+ if (JSON.parseObject(doc.toString()).containsKey("format")) {
+ String name = JSON.parseObject(field.toString()).get("name").toString();
+ Object format = JSON.parseObject(doc.toString()).get("format");
+ JSONObject formatObject = JSON.parseObject(format.toString());
+
+ String functions = formatObject.get("functions").toString();
+ String appendTo = null;
+ String params = null;
+
+ if (formatObject.containsKey("appendTo")) {
+ appendTo = formatObject.get("appendTo").toString();
+ }
+
+ if (formatObject.containsKey("param")) {
+ params = formatObject.get("param").toString();
+ }
+
+
+ if (StringUtil.isNotBlank(appendTo) && StringUtil.isBlank(params)) {
+ String[] functionArray = functions.split(FlowWriteConfig.FORMAT_SPLITTER);
+ String[] appendToArray = appendTo.split(FlowWriteConfig.FORMAT_SPLITTER);
+
+ for (int i = 0; i < functionArray.length; i++) {
+ list.add(new String[]{name, appendToArray[i], functionArray[i], null});
+ }
+
+ } else if (StringUtil.isNotBlank(appendTo) && StringUtil.isNotBlank(params)) {
+ String[] functionArray = functions.split(FlowWriteConfig.FORMAT_SPLITTER);
+ String[] appendToArray = appendTo.split(FlowWriteConfig.FORMAT_SPLITTER);
+ String[] paramArray = params.split(FlowWriteConfig.FORMAT_SPLITTER);
+
+ for (int i = 0; i < functionArray.length; i++) {
+ list.add(new String[]{name, appendToArray[i], functionArray[i], paramArray[i]});
+
+ }
+ } else {
+ list.add(new String[]{name, name, functions, params});
+ }
+
+ }
+ }
+
+ }
+ return list;
+ }
+
+}