【情报二期】数据域服务 - kafka数据消费业务代码、同行类数据查询代码开发

src/main/java/com/hhwy/qbeqsjy/common/CommonUtil.java

@@ -1,9 +1,5 @@
 package com.hhwy.qbeqsjy.common;
 
-import org.locationtech.jts.geom.Coordinate;
-import org.locationtech.jts.geom.GeometryFactory;
-import org.locationtech.jts.geom.Point;
-import org.locationtech.jts.geom.Polygon;
 import java.io.*;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;

src/main/java/com/hhwy/qbeqsjy/common/HttpClient.java

@@ -418,14 +418,12 @@ public class HttpClient {
     public String postResourceFWKFPT(String url, List<NameValuePair> nameValuePairs, Map<String, Object> headerParamsMap) {
         HttpPost postMethod = new HttpPost(url);
         postMethod.setConfig(this.requestConfig);
-
         if(headerParamsMap != null && headerParamsMap.size() > 0){
             Set<String> keys = headerParamsMap.keySet();
             for (String key : keys) {
                 postMethod.setHeader(key, headerParamsMap.get(key).toString());
             }
         }
-
         CloseableHttpResponse response = null;
         String content = null;
         try {

src/main/java/com/hhwy/qbeqsjy/kafka/service/ConsumeService.java

@@ -104,13 +104,14 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                     String warningInfoJSONStr = oriField.toJSONString();
                     // 命中时间不采用原始预警数据中的【截获时间】，而采用该服务消费到当前数据时的自然时间作为命中时间
                     String captureTime = TimeTool.timeStampToDateString(TimeTool.getNowTimeStamp());
+                    // 如果布控资源为网安AUTH协议表
                     if(Constants.TABLE_AUTH_RES_ID.equals(resourceId)){
                         // 1. ========================= 过滤基站号为空的数据
                         String baseStationId = oriField.getString("BASE_STATION_ID");
                         if(StringUtils.isBlank(baseStationId)){
                             continue;
                         }
-                        // 2. ========================= 过滤不在布控区域范围的数据
+                        // 2. ========================= 如果存在布控区域，则过滤不在布控区域范围的数据
                         List<Map<String, Object>> ctrlAreaInfoList = ctrlDao.queryCtrlAreaInfo(taskId);
                         if(ctrlAreaInfoList.size() > 0){
                             String lon = oriField.getString("LONGITUDE");
@@ -118,8 +119,8 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                             for (Map<String, Object> ctrlAreaInfo : ctrlAreaInfoList) {
                                 Integer areaGeoType = (Integer) ctrlAreaInfo.get("area_geo_type");
                                 JSONArray location = JSON.parseArray((String) ctrlAreaInfo.get("location"));
-                                // ****** 说明：如果出现布控区域范围存在重叠，并且命中点位也恰巧出现在重叠区域的极端情况，目前则只保存一份数据
-                                // 布控区域是圆形时
+                                // ****** 说明：如果出现多个布控区域范围存在重叠，并且命中点位也恰巧出现在重叠区域的极端情况，目前则只保存命中第一个布控区域时的一份数据
+                                // 布控区域是圆形
                                 if(Constants.NUM_0 == areaGeoType){
                                     JSONObject loc = location.getJSONObject(0);
                                     String centerLon = loc.getString("lon");
@@ -129,7 +130,7 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                                         warningInfoList.add(new Object[]{taskId, resourceId, clue, clueType, warningInfoJSONStr, captureTime});
                                         break;
                                     }
-                                } else {   // 布控区域是一般多边形时
+                                } else {   // 布控区域是一般多边形
                                     List<Coordinate> coordinates = new ArrayList<>();
                                     for (Object obj : location) {
                                         JSONObject loc = (JSONObject) obj;
@@ -143,17 +144,19 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                                     }
                                 }
                             }
+                        } else {   // 如果不存在布控区域（默认布控区域范围为全天津市），则直接保存入库
+                            warningInfoList.add(new Object[]{taskId, resourceId, clue, clueType, warningInfoJSONStr, captureTime});
                         }
-                    } else {
+                    } else {   // 如果是其他类布控资源，则直接保存入库
                         warningInfoList.add(new Object[]{taskId, resourceId, clue, clueType, warningInfoJSONStr, captureTime});
                     }
                 }
                 // ========================= 批量入库符合条件的预警数据
                 if(warningInfoList.size() > 0){
                     ctrlDao.batchInsertWarningInfo(warningInfoList);
+                    warningInfoList.clear();
                 }
-                log.info("【QBEQSJYCTRL - 消费者服务】 当次消费的数据量：{}，符合条件的数据量：{}", records.count(), warningInfoList.size());
-                warningInfoList.clear();
+                log.info("【QBEQSJYCTRL - 消费者服务】 当次消费的数据量：{}，符合条件保存入库的数据量：{}", records.count(), warningInfoList.size());
             }
         }
     }

src/main/java/com/hhwy/qbeqsjy/kafka/service/SameTravelService.java

@@ -0,0 +1,169 @@
+package com.hhwy.qbeqsjy.kafka.service;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.hhwy.qbeqsjy.common.HttpClient;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.NameValuePair;
+import org.apache.http.message.BasicNameValuePair;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * 同行类数据查询业务逻辑
+ */
+@Slf4j
+@Service
+public class SameTravelService {
+
+    @Value("${fwkfpt.token.url}")
+    private String fwkfptTokenUrl;
+    @Value("${fwkfpt.uniQuery.url}")
+    private String fwkfptUniQueryUrl;
+
+    public void callService_3(){
+        Map<String, Object> tokenHeaderParamsMap = new HashMap<>();
+        tokenHeaderParamsMap.put("Content-Type", "application/x-www-form-urlencoded");
+        Map<String, Object> userTokenMap = new HashMap<>();
+        userTokenMap.put("type", "user");
+        userTokenMap.put("policeNo", "000000");
+        userTokenMap.put("userId", "-1");
+        Map<String, Object> appTokenMap = new HashMap<>();
+        appTokenMap.put("appId", "A-17865251449995264");
+        appTokenMap.put("userToken", JSON.toJSONString(userTokenMap));
+        List<NameValuePair> nameValuePairs = new ArrayList<>();
+        nameValuePairs.add(new BasicNameValuePair("userToken", JSON.toJSONString(userTokenMap)));
+        nameValuePairs.add(new BasicNameValuePair("appToken", JSON.toJSONString(appTokenMap)));
+        nameValuePairs.add(new BasicNameValuePair("serviceid", "7251188999258112"));
+        HttpClient httpClient = new HttpClient();
+        String content = httpClient.postResourceFWKFPT(fwkfptTokenUrl, nameValuePairs, tokenHeaderParamsMap);
+        JSONObject resultJSONobj = JSON.parseObject(content);
+        String appTokenId = "";
+        String userTokenId = "";
+        if(resultJSONobj != null){
+            String appTokenJSONStr = resultJSONobj.getString("appTokenId");
+            String userTokenJSONStr = resultJSONobj.getString("userTokenId");
+            if(StringUtils.isBlank(appTokenJSONStr) && StringUtils.isBlank(userTokenJSONStr)){
+                String message = resultJSONobj.getString("message");
+                log.error("获取【服务开放平台 # 权限认证令牌】 令牌创建失败，异常信息：{}", message);
+            }
+            appTokenId = JSON.parseObject(appTokenJSONStr).getString("appToken");
+            userTokenId = JSON.parseObject(userTokenJSONStr).getString("userToken");
+            log.info("获取【服务开放平台 # 权限认证令牌】 令牌创建成功，响应消息：{}", content);
+        }
+        long start = System.currentTimeMillis();
+        Map<String, Object> headerParamsMap = new HashMap<>();
+        headerParamsMap.put("Content-Type", "application/json");
+        headerParamsMap.put("apptoken", appTokenId);
+        headerParamsMap.put("usertoken", userTokenId);
+        headerParamsMap.put("userid", "-1");
+        headerParamsMap.put("subid", "A-17865251449995264");
+        headerParamsMap.put("serviceid", "7251188999258112");
+        headerParamsMap.put("police-no", "000000");
+        headerParamsMap.put("app-id", "1");
+        headerParamsMap.put("role-id", "1");
+        httpClient.sendGetRequst(fwkfptUniQueryUrl, headerParamsMap);
+        log.info("调用【服务开放平台 # 天津市手机号最后一次位置查询接口】 服务调用成功，耗时：{}ms", System.currentTimeMillis() - start);
+    }
+
+
+    public void callService_14(){
+        Map<String, Object> tokenHeaderParamsMap = new HashMap<>();
+        tokenHeaderParamsMap.put("Content-Type", "application/x-www-form-urlencoded");
+        Map<String, Object> userTokenMap = new HashMap<>();
+        userTokenMap.put("type", "user");
+        userTokenMap.put("policeNo", "000000");
+        userTokenMap.put("userId", "-1");
+        Map<String, Object> appTokenMap = new HashMap<>();
+        appTokenMap.put("appId", "A-17865251449995264");
+        appTokenMap.put("userToken", JSON.toJSONString(userTokenMap));
+        List<NameValuePair> nameValuePairs = new ArrayList<>();
+        nameValuePairs.add(new BasicNameValuePair("userToken", JSON.toJSONString(userTokenMap)));
+        nameValuePairs.add(new BasicNameValuePair("appToken", JSON.toJSONString(appTokenMap)));
+        nameValuePairs.add(new BasicNameValuePair("serviceid", "10579974096420864"));
+        HttpClient httpClient = new HttpClient();
+        String content = httpClient.postResourceFWKFPT(fwkfptTokenUrl, nameValuePairs, tokenHeaderParamsMap);
+        JSONObject resultJSONobj = JSON.parseObject(content);
+        String appTokenId = "";
+        String userTokenId = "";
+        if(resultJSONobj != null){
+            String appTokenJSONStr = resultJSONobj.getString("appTokenId");
+            String userTokenJSONStr = resultJSONobj.getString("userTokenId");
+            if(StringUtils.isBlank(appTokenJSONStr) && StringUtils.isBlank(userTokenJSONStr)){
+                String message = resultJSONobj.getString("message");
+                log.error("获取【服务开放平台 # 权限认证令牌】 令牌创建失败，异常信息：{}", message);
+            }
+            appTokenId = JSON.parseObject(appTokenJSONStr).getString("appToken");
+            userTokenId = JSON.parseObject(userTokenJSONStr).getString("userToken");
+            log.info("获取【服务开放平台 # 权限认证令牌】 令牌创建成功，响应消息：{}", content);
+        }
+        long start = System.currentTimeMillis();
+        Map<String, Object> headerParamsMap = new HashMap<>();
+        headerParamsMap.put("Content-Type", "application/json");
+        headerParamsMap.put("apptoken", appTokenId);
+        headerParamsMap.put("usertoken", userTokenId);
+        headerParamsMap.put("userid", "-1");
+        headerParamsMap.put("subid", "A-17865251449995264");
+        headerParamsMap.put("serviceid", "10579974096420864");
+        headerParamsMap.put("police-no", "000000");
+        headerParamsMap.put("app-id", "1");
+        headerParamsMap.put("role-id", "1");
+        String reqBody = "{ \"From\":\"320100\", \"To\":\"320100\", \"MessageSequence\":\"2020030314141200001\", \"RequestParam\":{ \"Condition\":\"VEH_PLATE_NUM='津RVG959'\", \"OtherCondition\":{ \"AsyncBoolean\":\"0\", \"AsyncIdentity\":\"0\", \"AsyncOnceReturnNum\":\"100\", \"AsyncQuery\":\"\", \"CallbackID\":\"\", \"CodeMode\":\"0\", \"MaxReturnNum\":\"10\", \"SortResults\":\"ILLE_TIME+\" }, \"ResourceInfos\":[ { \"DataItems\":[ { \"Name\":\"VEH_PLATE_NUM\", \"Fmt\": \"\" }, { \"Name\":\"ILLE_TIME\", \"Fmt\": \"\" } ], \"ResourceName\":\"R-120000000001-00000170\" } ] } }";
+        httpClient.postResource(fwkfptUniQueryUrl, reqBody, headerParamsMap);
+        log.info("调用【服务开放平台 # 天津市通查服务查询接口】 服务调用成功，耗时：{}ms", System.currentTimeMillis() - start);
+    }
+
+
+    public void callService_15(){
+        Map<String, Object> tokenHeaderParamsMap = new HashMap<>();
+        tokenHeaderParamsMap.put("Content-Type", "application/x-www-form-urlencoded");
+        Map<String, Object> userTokenMap = new HashMap<>();
+        userTokenMap.put("type", "user");
+        userTokenMap.put("policeNo", "000000");
+        userTokenMap.put("userId", "-1");
+        Map<String, Object> appTokenMap = new HashMap<>();
+        appTokenMap.put("appId", "A-17865251449995264");
+        appTokenMap.put("userToken", JSON.toJSONString(userTokenMap));
+        List<NameValuePair> nameValuePairs = new ArrayList<>();
+        nameValuePairs.add(new BasicNameValuePair("userToken", JSON.toJSONString(userTokenMap)));
+        nameValuePairs.add(new BasicNameValuePair("appToken", JSON.toJSONString(appTokenMap)));
+        nameValuePairs.add(new BasicNameValuePair("serviceid", "32655244525568"));
+        HttpClient httpClient = new HttpClient();
+        String content = httpClient.postResourceFWKFPT(fwkfptTokenUrl, nameValuePairs, tokenHeaderParamsMap);
+        JSONObject resultJSONobj = JSON.parseObject(content);
+        String appTokenId = "";
+        String userTokenId = "";
+        if(resultJSONobj != null){
+            String appTokenJSONStr = resultJSONobj.getString("appTokenId");
+            String userTokenJSONStr = resultJSONobj.getString("userTokenId");
+            if(StringUtils.isBlank(appTokenJSONStr) && StringUtils.isBlank(userTokenJSONStr)){
+                String message = resultJSONobj.getString("message");
+                log.error("获取【服务开放平台 # 权限认证令牌】 令牌创建失败，异常信息：{}", message);
+            }
+            appTokenId = JSON.parseObject(appTokenJSONStr).getString("appToken");
+            userTokenId = JSON.parseObject(userTokenJSONStr).getString("userToken");
+            log.info("获取【服务开放平台 # 权限认证令牌】 令牌创建成功，响应消息：{}", content);
+        }
+        long start = System.currentTimeMillis();
+        Map<String, Object> headerParamsMap = new HashMap<>();
+        headerParamsMap.put("Content-Type", "application/json");
+        headerParamsMap.put("apptoken", appTokenId);
+        headerParamsMap.put("usertoken", userTokenId);
+        headerParamsMap.put("userid", "-1");
+        headerParamsMap.put("subid", "A-17865251449995264");
+        headerParamsMap.put("serviceid", "32655244525568");
+        headerParamsMap.put("police-no", "000000");
+        headerParamsMap.put("app-id", "1");
+        headerParamsMap.put("role-id", "1");
+        String reqBody = "{ \"From\":\"120000\", \"To\":\"120000\", \"MessageSequence\":\"2020030314141200001\", \"RequestParam\":{ \"Condition\":\"DE00000618='370881198904288273'\", \"ResourceInfos\":[ { \"DataItems\":[ { \"Name\": \"HBH\", \"Fmt\": \"\" }, { \"Name\": \"LKZWXM\", \"Fmt\": \"\" }, { \"Name\": \"ZJHM\", \"Fmt\": \"\" }, { \"Name\": \"ZJLX\", \"Fmt\": \"\" } , { \"Name\": \"HBRQ\", \"Fmt\": \"\" } ], \"ResourceName\":\"R-120000000000-00002228\" }], \"OtherCondition\":{ \"AsyncBoolean\":\"1\", \"AsyncIdentity\":\"1\", \"AsyncOnceReturnNum\":\"100\", \"AsyncQuery\":\"1\", \"CallbackID\":\"1\", \"CodeMode\":\"0\", \"MaxReturnNum\":\"2\", \"SortResults\":\"\" } } }";
+        httpClient.postResource(fwkfptUniQueryUrl, reqBody, headerParamsMap);
+        log.info("调用【服务开放平台 # 天津市通用拓展查询服务查询接口】 服务调用成功，耗时：{}ms", System.currentTimeMillis() - start);
+    }
+
+}

src/main/resources/application.properties

@@ -40,6 +40,13 @@ uniCtrl.del.url=http://130.0.46.141:8090/lokiRest/uniCtrl/monitor/del
 # \u3010\u7EDF\u4E00\u5E03\u63A7\u3011\u5E03\u63A7\u7ED3\u679C\u67E5\u8BE2\u63A5\u53E3url
 uniCtrl.query.url=http://130.0.46.141:8090/lokiRest/unictrl/query/status
 
+# \u70FD\u706B - \u3010\u670D\u52A1\u5F00\u653E\u5E73\u53F0\u3011\u914D\u7F6E
+# \u3010\u670D\u52A1\u5F00\u653E\u5E73\u53F0\u3011\u83B7\u53D6token\u63A5\u53E3url
+fwkfpt.token.url=http://130.0.56.75:8080/restcloud/api/v1/token/create
+# \u3010\u670D\u52A1\u5F00\u653E\u5E73\u53F0\u3011\u540C\u884C\u7C7B\u6570\u636E\u901A\u7528\u67E5\u8BE2\u63A5\u53E3url - \u540C\u822A\u73ED\u3001\u540C\u706B\u8F66\u3001\u540C\u5BBE\u9986\u3001\u540C\u7F51\u5427
+fwkfpt.uniQuery.url=http://130.0.56.75:8080/restcloud/fhopenapiplatform/
+
+
 # \u534E\u4E3A - kafka\u914D\u7F6E
 # \u4E1A\u52A1topic
 topic.name=UNICTRL_DATA_QBEQSJYCTRL_V1.0