+ * current_timestamp
+ * snowflake_id
+ * geo_ip_detail
+ * geo_asn
+ * radius_match
+ * geo_ip_country
+ * geo_asn
+ * sub_domain
+ * sub_domain
+ */
+ public static String dealCommonMessage(String message) {
+
+ Object object = JSONObject.parseObject(message, mapObject.getClass());
+// System.out.println("补全之前 ===》 "+JSON.toJSONString(object));
+ try {
+ for (String[] strings : jobList) {
+
+ if (strings[2].equals("current_timestamp")) {
+ JsonParseUtil.setValue(object, strings[1], getCurrentTime());
+ } else if (strings[2].equals("snowflake_id")) {
+ JsonParseUtil.setValue(object, strings[1], getSnowflakeId());
+ } else if (strings[2].equals("geo_ip_detail")) {
+ JsonParseUtil.setValue(object, strings[1], getGeoIpDetail(JsonParseUtil.getValue(object, strings[0]).toString()));
+ } else if (strings[2].equals("geo_asn")) {
+ JsonParseUtil.setValue(object, strings[1], getGeoIpDetail(JsonParseUtil.getValue(object, strings[0]).toString()));
+ } else if (strings[2].equals("radius_match")) {
+ JsonParseUtil.setValue(object,strings[1],HBaseUtils.getAccount(JsonParseUtil.getValue(object,strings[0]).toString()));
+// JsonParseUtil.setValue(object, strings[1], "aaaaaaaaa");
+ } else if (strings[2].equals("geo_ip_country")) {
+ JsonParseUtil.setValue(object, strings[1], getGeoIpCountry(JsonParseUtil.getValue(object, strings[0]).toString()));
+ } else if (strings[0].equals("http_host") && strings[2].equals("sub_domain")) {
+ JsonParseUtil.setValue(object,strings[1],getTopDomain(null,JsonParseUtil.getValue(object,strings[0]).toString()));
+ } else if (strings[0].equals("ssl_sni") && strings[2].equals("sub_domain")) {
+ if (StringUtil.isBlank(JsonParseUtil.getValue(object, strings[1]).toString())) {
+ JsonParseUtil.setValue(object,strings[1],getTopDomain(JsonParseUtil.getValue(object,strings[0]).toString(),null));
+ }
+
+ }
+ }
+
+
+ return JSONObject.toJSONString(object);
+// System.out.println("补全之后 ===》 "+JSON.toJSONString(object));
+
+ } catch (Exception e) {
+ logger.error(FlowWriteConfig.KAFKA_TOPIC + "日志解析过程出现异常");
+ e.printStackTrace();
+ return "";
+ }
+ }
+
+
+ /**
+ * 有sni通过sni获取域名，有host根据host获取域名
+ *
+ * @param sni sni
+ * @param host host
+ * @return 顶级域名
+ */
+ private static String getTopDomain(String sni, String host) {
+ if (StringUtil.isNotBlank(host)) {
+ return getDomainName(host);
+ } else if (StringUtil.isNotBlank(sni)) {
+ return getDomainName(sni);
+ } else {
+ return "";
+ }
+ }
+
+
+ /**
+ * 根据url截取顶级域名
+ *
+ * @param host 网站url
+ * @return 顶级域名
+ */
+ private static String getDomainName(String host) {
+ String domain = "";
+ try {
+ domain = InternetDomainName.from(host).topPrivateDomain().toString();
+ } catch (Exception e) {
+ logger.error("host解析顶级域名异常: " + e.getMessage());
+ }
+ return domain;
+ }
+
+ /**
+ * 生成当前时间戳的操作
+ */
+ private static long getCurrentTime() {
+ return (System.currentTimeMillis() / 1000);
+ }
+
+ /**
+ * 雪花模型生成id
+ *
+ * @return
+ */
+ private static long getSnowflakeId() {
+
+ return SnowflakeId.generateId();
+ }
+
+ /**
+ * 根据clientIp获取location信息
+ *
+ * @param ip
+ * @return
+ */
+ private static String getGeoIpDetail(String ip) {
+
+ return ipLookup.cityLookupDetail(ip);
+ }
+
+ /**
+ * 根据ip获取asn信息
+ *
+ * @param ip
+ * @return
+ */
+ private static String getGeoAsn(String ip) {
+
+ return ipLookup.asnLookup(ip, true);
+ }
+
+ /**
+ * 根据ip获取country信息
+ *
+ * @param ip
+ * @return
+ */
+ private static String getGeoIpCountry(String ip) {
+
+ return ipLookup.countryLookup(ip);
+ }
+
+ /**
+ * radius借助hbase补齐
+ *
+ * @param ip
+ * @return
+ */
+ private static String radiusMatch(String ip) {
+ return HBaseUtils.getAccount(ip);
+ }
+
+ /**
+ * switch 匹配合适的方法
+ * current_timestamp
+ * snowflake_id
+ * geo_ip_detail
+ * geo_asn
+ * radius_match
+ * geo_ip_country
+ * geo_asn
+ * sub_domain
+ * sub_domain
+ * @param func
+ */
+//TODO 行不通的原因是无法统一一个确定的返回值类型
+ /* private static String switchFunc(String func){
+ switch (func){
+ case "current_timestamp":
+ return String.valueOf(getCurrentTime());
+ case "snowflake_id":
+ case "geo_ip_detail":
+ case "geo_asn":
+ case "radius_match":
+ case "geo_ip_country":
+ case "sub_domain":
+ }
+ return func;
+ }*/
+}
diff --git a/src/main/java/cn/ac/iie/utils/hbase/HBaseUtils.java b/src/main/java/cn/ac/iie/utils/hbase/HBaseUtils.java
index dfcf3b8..f386003 100644
--- a/src/main/java/cn/ac/iie/utils/hbase/HBaseUtils.java
+++ b/src/main/java/cn/ac/iie/utils/hbase/HBaseUtils.java
@@ -17,6 +17,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
+
/**
* HBase 工具类
*
diff --git a/src/main/java/cn/ac/iie/utils/http/HttpClientUtil.java b/src/main/java/cn/ac/iie/utils/http/HttpClientUtil.java
new file mode 100644
index 0000000..2aa8885
--- /dev/null
+++ b/src/main/java/cn/ac/iie/utils/http/HttpClientUtil.java
@@ -0,0 +1,51 @@
+package cn.ac.iie.utils.http;
+
+import org.apache.http.HttpEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+
+/**
+ * 获取网关schema的工具类
+ */
+public class HttpClientUtil {
+ public static String requestByGetMethod(String s) {
+ CloseableHttpClient httpClient = HttpClients.createDefault();
+ StringBuilder entityStringBuilder = null;
+ try {
+ HttpGet get = new HttpGet(s);
+ CloseableHttpResponse httpResponse = null;
+ httpResponse = httpClient.execute(get);
+ try {
+ HttpEntity entity = httpResponse.getEntity();
+ entityStringBuilder = new StringBuilder();
+ if (null != entity) {
+ BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(httpResponse.getEntity().getContent(), "UTF-8"), 8 * 1024);
+ String line = null;
+ while ((line = bufferedReader.readLine()) != null) {
+ entityStringBuilder.append(line);
+ }
+ }
+ } finally {
+ httpResponse.close();
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ try {
+ if (httpClient != null) {
+ httpClient.close();
+ }
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ return entityStringBuilder.toString();
+ }
+
+}
diff --git a/src/main/java/cn/ac/iie/utils/json/JsonParseUtil.java b/src/main/java/cn/ac/iie/utils/json/JsonParseUtil.java
new file mode 100644
index 0000000..ec8b7f5
--- /dev/null
+++ b/src/main/java/cn/ac/iie/utils/json/JsonParseUtil.java
@@ -0,0 +1,180 @@
+package cn.ac.iie.utils.json;
+
+import cn.ac.iie.utils.http.HttpClientUtil;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import net.sf.cglib.beans.BeanGenerator;
+import net.sf.cglib.beans.BeanMap;
+
+import java.util.*;
+
+/**
+ * 使用fastjson解析json的工具类
+ */
+public class JsonParseUtil {
+
+ /**
+ * 模式匹配，给定一个类型字符串返回一个类类型
+ *
+ * @param type
+ * @return
+ */
+
+ public static Class getClassName(String type) {
+ Class clazz = int.class;
+
+ switch (type) {
+ case "int":
+ clazz = int.class;
+ break;
+ case "double":
+ clazz = double.class;
+ break;
+ case "float":
+ clazz = float.class;
+ break;
+ case "long":
+ clazz = long.class;
+ break;
+ case "char":
+ clazz = char.class;
+ break;
+ case "byte":
+ clazz = byte.class;
+ break;
+ case "boolean":
+ clazz = boolean.class;
+ break;
+ case "short":
+ clazz = short.class;
+ break;
+ default:
+ clazz = String.class;
+ }
+ return clazz;
+ }
+
+ /**
+ * 根据反射生成对象的方法
+ *
+ * @param properties
+ * @return 生成的Object类型的对象
+ */
+ public static Object generateObject(Map properties) {
+ BeanGenerator generator = new BeanGenerator();
+ Set keySet = properties.keySet();
+ for (Iterator i = keySet.iterator(); i.hasNext(); ) {
+ String key = (String) i.next();
+ generator.addProperty(key, (Class) properties.get(key));
+ }
+ return generator.create();
+ }
+
+ /**
+ * 获取属性值的方法
+ *
+ * @param obj
+ * @param property
+ * @return 属性的值
+ */
+ public static Object getValue(Object obj, String property) {
+ BeanMap beanMap = BeanMap.create(obj);
+ return beanMap.get(property);
+ }
+
+ /**
+ * 更新属性值的方法
+ *
+ * @param obj
+ * @param property
+ * @param value
+ */
+ public static void setValue(Object obj, String property, Object value) {
+ BeanMap beanMap = BeanMap.create(obj);
+ beanMap.put(property, value);
+ }
+
+ /**
+ * 通过获取String类型的网关schema链接来获取map，用于生成一个Object类型的对象
+ *
+ * @param http
+ * @return 用于反射生成schema类型的对象的一个map集合
+ */
+ public static HashMap