【情报二期】数据域服务 - kafka数据消费业务代码

pom.xml

@@ -42,13 +42,19 @@
       <version>3.5.1</version>
     </dependency>-->
 
+    <!-- JTS Topology Suite 处理二维几何图形的 Java 类库 -->
+    <dependency>
+      <groupId>org.locationtech.jts</groupId>
+      <artifactId>jts-core</artifactId>
+      <version>1.19.0</version>
+    </dependency>
+
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-validation</artifactId>
       <version>2.5.15</version>
     </dependency>
 
-
     <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter</artifactId>

src/main/java/com/hhwy/qbeqsjy/APP.java

@@ -2,6 +2,10 @@ package com.hhwy.qbeqsjy;
 
 import com.alibaba.fastjson.JSONArray;
 import com.hhwy.qbeqsjy.common.CommonUtil;
+import org.locationtech.jts.geom.Coordinate;
+
+import java.util.ArrayList;
+import java.util.List;
 
 public class APP {
 
@@ -23,6 +27,14 @@ public class APP {
 //        System.out.println(endIndex);
         // 0,1   1,2   2,3   3,6
 
+        // 定义多边形的经纬度坐标
+        List<Coordinate> coordinates = new ArrayList<>();
+        coordinates.add(new Coordinate(116.397128, 39.916527));
+        coordinates.add(new Coordinate(116.397928, 39.916527));
+        coordinates.add(new Coordinate(116.397928, 39.917327));
+        coordinates.add(new Coordinate(116.397128, 39.917327));
+        boolean inArea = CommonUtil.isInArea(coordinates, new Coordinate(117.3975, 39.917));
+        System.out.println(inArea);
 
     }
 }

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

@@ -1,5 +1,9 @@
 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;
@@ -16,6 +20,7 @@ import java.util.*;
 public class CommonUtil {
 
   private static ResourceBundle bundle;
+  private static GeometryFactory geometryFactory = new GeometryFactory();
 
   static {
     bundle = ResourceBundle.getBundle("application");
@@ -223,4 +228,25 @@ public class CommonUtil {
     }
   }
 
+
+  /**
+   * 判断某个点的经纬度坐标是否在某个多边形区域内
+   * @param coordinates   多边形区域的顶点经纬度坐标集合
+   * @param coordinate    待判断的点经纬度坐标
+   * @return
+   */
+  public static boolean isInArea(List<Coordinate> coordinates, Coordinate coordinate){
+    if(coordinates != null && coordinates.size() >= 3){
+      // 连接第一个点，封闭多边形
+      coordinates.add(coordinates.get(0));
+      Coordinate[] coordinateArray = coordinates.toArray(new Coordinate[0]);
+      Polygon polygon = geometryFactory.createPolygon(coordinateArray);
+      // 待判断的点
+      Point point = geometryFactory.createPoint(coordinate);
+      // 判断点是否在多边形内
+      return polygon.contains(point);
+    }
+    return false;
+  }
+
 }

src/main/java/com/hhwy/qbeqsjy/controller/CtrlController.java

@@ -24,7 +24,7 @@ public class CtrlController {
 
 
     /**
-     * 下发布控任务，不支持批量操作
+     * 下发布控任务，不支持批量操作，并记录布控任务对应的区域信息
      * @param ctrlDTO
      * @return
      *
@@ -95,7 +95,7 @@ public class CtrlController {
 
 
     /**
-     * 保存布控任务对应区域信息
+     * 保存用户域提交的的布控任务对应区域信息
      * @param reqBody
      * @return
      *
@@ -116,7 +116,7 @@ public class CtrlController {
      * 					"lat":"xxx"
      * 				}
      * 				......
-     * 			]
+     * 			  ]
      *         }
      *     ]
      * }

src/main/java/com/hhwy/qbeqsjy/dao/CtrlDao.java

@@ -49,7 +49,7 @@ public class CtrlDao {
 
 
     /**
-     * 批量插入布控任务区域信息
+     * 批量插入布控任务对应区域信息
      * @param ctrlAreaInfoList
      * @return
      */
@@ -71,7 +71,7 @@ public class CtrlDao {
      * @return
      */
     public boolean batchInsertWarningInfo(List<Object[]> warningInfoList){
-        String sql = "insert into t_ctrl_area_info (task_id, clue, clue_type, warning_msg, create_time) values (?, ?, ?, ?)";
+        String sql = "insert into t_ctrl_warning_info (task_id, clue, clue_type, warning_msg, create_time) values (?, ?, ?, ?, ?)";
         try {
             int[] insertRows = gaussDBJdbcTemplate.batchUpdate(sql, warningInfoList);
             return insertRows.length == warningInfoList.size();

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

@@ -4,8 +4,10 @@ import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.hhwy.qbeqsjy.common.CommonUtil;
 import com.hhwy.qbeqsjy.common.Constants;
+import com.hhwy.qbeqsjy.common.TimeTool;
 import com.hhwy.qbeqsjy.dao.CtrlDao;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -87,39 +89,37 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                 for (ConsumerRecord<Integer, String> record : records) {
                     JSONObject message = JSONObject.parseObject(record.value());
 //                    log.info(message.toJSONString());
-                    // 解析[UNICTRL]节点
+                    // 解析[ORIFIELD]节点，该节点数据格式与布控资源表有关，资源表不同，数据格式不同
+                    JSONObject oriField = message.getJSONObject("ORIFIELD");
+                    String baseStationId = oriField.getString("BASE_STATION_ID");
+                    // 过滤基站ID为空的数据
+                    if (StringUtils.isBlank(baseStationId)) {
+                        continue;
+                    }
+
+                    // 查询当前任务对应的布控区域信息，进行范围比对
+
+
+                    String lon = oriField.getString("LONGITUDE");
+                    String lat = oriField.getString("LATITUDE");
+
+
+
+                    // 解析[UNICTRL]节点，该节点数据格式保持一致，与布控资源表无关
                     JSONObject uniCtrl = message.getJSONObject("UNICTRL");
-                    String clueId = uniCtrl.getString("U_CLUEID");   // 样例：120000_59275350793392128_D201005_13502082832
+                    // 样例：clueId = 120000_59275350793392128_D201005_13502082832
+                    String clueId = uniCtrl.getString("U_CLUEID");
                     String[] clueIdStrArr = clueId.split("_");
                     String taskId = clueIdStrArr[Constants.NUM_1];
                     String clueType = clueIdStrArr[Constants.NUM_2];
                     String clue = clueIdStrArr[Constants.NUM_3];
                     String warningInfoJSONStr = message.toJSONString();
                     String resourceId = uniCtrl.getString("U_RESID");
+                    String captureTime = TimeTool.timeStampToDateString(Long.valueOf(oriField.getString("CAPTURE_TIME")));
                     Object[] warningInfo = new Object[]{taskId, clue, clueType, warningInfoJSONStr, createTime};
                     warningInfoList.add(warningInfo);
 
 
-                    // 解析[ORIFIELD]节点
-                    JSONObject oriField = message.getJSONObject("ORIFIELD");
-                    // 获取配置文件中数据资源编码对应的字段信息（以JSON字符串形式存储）
-                    String fieldJSONStr = CommonUtil.getConfValueByKey(resourceId);
-
-
-                    //====================== 1.过滤基站号为空的数据 ======================
-//                    if (StringUtils.isNotBlank(baseStationId)) {
-//                        JSONObject msg = new JSONObject();
-//                        msg.put("clue", clue);
-//                        msg.put("longitude", longitude);
-//                        msg.put("latitude", latitude);
-//                        msg.put("baseStationId", baseStationId);
-//                        msg.put("captureTime", captureTime);
-//                        consumeCount++;
-//                        hitMsgList.add(message);
-//                        //====================== 2.过滤掉经纬度不在指定区域范围的数据 - to do ======================
-
-
-//                    }
                 }
                 //====================== 3.符合条件的预警数据批量入库 - to do ======================
                 ctrlDao.batchInsertWarningInfo(warningInfoList);

src/main/java/com/hhwy/qbeqsjy/service/CtrlService.java

@@ -15,7 +15,6 @@ import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
-
 import javax.annotation.Resource;
 import java.io.*;
 import java.util.*;
@@ -480,7 +479,7 @@ public class CtrlService {
 
 
     /**
-     * 保存布控任务区域信息
+     * 保存布控任务对应区域信息
      * @param taskId
      * @param areaList
      * @return

src/main/java/com/hhwy/qbeqsjy/strategy/impl/AuthStrategy.java

@@ -17,14 +17,14 @@ public class AuthStrategy implements Strategy {
         // 原始字段
         JSONObject oriField = warningInfo.getJSONObject("ORIFIELD");
         // 线索值
-//        String clue = oriField.getString("AUTH_ACCOUNT");
+//        String clue = oriField.getString("SRC_AUTH_ACCOUNT");
         // 命中时间
         String captureTime = TimeTool.timeStampToDateString(Long.valueOf(oriField.getString("CAPTURE_TIME")));
         // 经度
         String lon = oriField.getString("LONGITUDE");
         // 维度
         String lat = oriField.getString("LATITUDE");
-        // 基站ID（4G）
+        // 基站ID
         String baseStationId = oriField.getString("BASE_STATION_ID");
 
         return null;