【情报二期】数据域服务 - 预警数据增量查询接口代码

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

@@ -6,6 +6,7 @@ import com.hhwy.qbeqsjy.common.SnowflakeIdGenerator;
 import org.locationtech.jts.geom.Coordinate;
 
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
 
 public class APP {
@@ -36,13 +37,17 @@ public class APP {
 //        coordinates.add(new Coordinate(116.397128, 39.917327));
 //        boolean inArea = CommonUtil.isInArea(coordinates, new Coordinate(117.3975, 39.917));
 //        System.out.println(inArea);
-        SnowflakeIdGenerator generator = new SnowflakeIdGenerator(1);
+        /*SnowflakeIdGenerator generator = new SnowflakeIdGenerator(1);
         String warningMsgId = String.valueOf(generator.nextId());
         String warningMsgId2 = String.valueOf(generator.nextId());
         String warningMsgId3 = String.valueOf(generator.nextId());
         System.out.println(warningMsgId);
         System.out.println(warningMsgId2);
-        System.out.println(warningMsgId3);
+        System.out.println(warningMsgId3);*/
+        Date date = new Date();
+        long time = date.getTime();
+        System.out.println(time);
+        System.out.println(time / 1000);
 
     }
 }

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

@@ -11,6 +11,12 @@ import java.util.Locale;
  */
 public class Constants {
 
+    // 任务标识：单人预警
+    public static final String TASK_DRYJ = "DRYJ";
+    // 任务标识：聚集预警
+    public static final String TASK_JJYJ = "JJYJ";
+    // 默认布控系统编号
+    public static final String DEFAULT_SYSID = "QBEQSJYCTRL";
     // 预警状态：未读
     public static final String YJZT_WD = "未读";
     //网安AUTH协议表数据资源编码

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

@@ -5,7 +5,6 @@ import com.alibaba.fastjson.JSONObject;
 import com.hhwy.qbeqsjy.common.Constants;
 import com.hhwy.qbeqsjy.domain.AjaxResult;
 import com.hhwy.qbeqsjy.dto.CtrlDTO;
-import com.hhwy.qbeqsjy.dto.QueryDTO;
 import com.hhwy.qbeqsjy.service.CtrlService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -38,32 +37,38 @@ public class CtrlController {
      *     "areaIdList":["dddd","eeee"],
      *     "clueList":[
      *         {
-     *             "clue":"",90
+     *             "clue":"13500000000",
      *             "clueType":"D201005",
      *         },
+     *         {
+     *             "clue":"120101000000000000",
+     *             "clueType":"D310111",
+     *         }
      *         ......
-     *     ],
-     *     --- 因为用户域需保存一份来自数据域的预警结果数据，此处参数用作数据回填，避免在用户域出现预警结果值大量关联查询原始线索信息的情况
-     *     "yhyTaskInfo":{   --- 用户域布控任务信息
-     *        "yjry_xm":"",
-     *        "yjry_sfzh":"",
-     *        "yjry_sjh":"",
-     *        "rwmc":"",
-     *        "":"",
-     *     }
+     *     ]
      *
      * }
      */
     @PostMapping("/ctrl")
     public AjaxResult ctrl(@RequestBody CtrlDTO ctrlDTO){
+        String taskFlag = ctrlDTO.getTaskFlag();
         String taskId = ctrlDTO.getTaskId();
         List<Map<String, Object>> clueList = ctrlDTO.getClueList();
         List<String> resourceIdList = ctrlDTO.getResourceIdList();
-        if(StringUtils.isBlank(taskId) || clueList == null || clueList.size() == 0 || resourceIdList == null || resourceIdList.size() == 0){
+        if(StringUtils.isBlank(taskId) || clueList.size() == 0 || resourceIdList.size() == 0){
             return AjaxResult.error(Constants.PARAMS_EMPTY_MSG);
         }
         try {
-            ctrlService.processCtrl(ctrlDTO);
+            if(Constants.TASK_DRYJ.equals(taskFlag)){
+                ctrlService.processDRYJCtrl(ctrlDTO);
+            }
+            if(Constants.TASK_JJYJ.equals(taskFlag)){
+                ctrlService.processJJYJCtrl(ctrlDTO);
+            }
+            if(StringUtils.isBlank(taskFlag)){
+                ctrlService.processDRYJCtrl(ctrlDTO);
+                ctrlService.processJJYJCtrl(ctrlDTO);
+            }
             return AjaxResult.success(Constants.CTRL_ING_MSG);
         }catch (Exception e){
             log.error("任务布控发生异常", e);
@@ -106,19 +111,20 @@ public class CtrlController {
 
 
     /**
-     * 保存用户域提交的布控任务对应的区域信息
+     * 保存用户域提交的布控任务相关信息
      * @param reqBody
      * @return
      */
     @PostMapping("/saveCtrlTaskInfo")
     public AjaxResult saveCtrlTaskInfo(@RequestBody JSONObject reqBody){
-        String taskId = reqBody.getString("taskId");
+        JSONObject taskInfo = reqBody.getJSONObject("taskInfo");
         JSONArray clueInfo = reqBody.getJSONArray("clueInfo");
-        if(StringUtils.isBlank(taskId) || clueInfo.size() == 0){
+        if(taskInfo == null || clueInfo.size() == 0){
             return AjaxResult.error(Constants.PARAMS_EMPTY_MSG);
         }
         try{
-            return ctrlService.saveCtrlTaskInfo(reqBody)? AjaxResult.success() : AjaxResult.error();
+            JSONArray areaInfo = reqBody.getJSONArray("areaInfo");
+            return ctrlService.saveCtrlTaskInfo(taskInfo, areaInfo, clueInfo)? AjaxResult.success() : AjaxResult.error();
         } catch (Exception e){
             log.error("保存用户域任务相关信息发生异常", e);
         }
@@ -127,55 +133,20 @@ public class CtrlController {
 
 
     /**
-     * 条件查询布控任务预警结果数据，支持分页查询，一次查询默认最大返回1000条数据
-     * @param queryDTO
+     * 增量查询布控任务预警结果数据
      * @return
      */
-    @PostMapping("/getWarningInfo")
-    public AjaxResult getWarningInfo(@RequestBody QueryDTO queryDTO){
+    @GetMapping("/getCtrlWarningInfo")
+    public AjaxResult getCtrlWarningInfo(){
         try{
-            return AjaxResult.success(ctrlService.getWarningInfo(queryDTO));
+            return AjaxResult.success(ctrlService.getWarningInfoByInc());
         } catch (Exception e){
-            log.error("查询布控任务预警结果数据发生异常", e);
+            log.error("增量查询布控任务预警结果数据发生异常", e);
         }
         return AjaxResult.error(Constants.SERVER_ERROR_MSG);
     }
 
 
-    /**
-     * 保存用户域提交的布控任务对应的线索人员信息
-     * @param reqBody
-     * @return
-     *
-     * 请求参数：
-     * {
-     *     "taskId":"59275350793392128",
-     *     "clueList":[
-     *            {
-     *             "ryxm":"xxx",
-     *             "zjlxdm":"xxx",
-     *             "zjhm":"xxx",
-     * 				......
-     * 			  }
-     *      ]
-     * }
-     */
-    /*@PostMapping("/saveCtrlClueInfo")
-    public AjaxResult saveCtrlClueInfo(@RequestBody JSONObject reqBody){
-        String taskId = reqBody.getString("taskId");
-        JSONArray clueList = reqBody.getJSONArray("clueList");
-        if(StringUtils.isBlank(taskId) || clueList == null || clueList.size() == 0){
-            return AjaxResult.error(Constants.PARAMS_EMPTY_MSG);
-        }
-        try{
-            return ctrlService.saveCtrlClueInfo(taskId, clueList)? AjaxResult.success() : AjaxResult.error();
-        } catch (Exception e){
-            log.error("保存任务线索人员信息发生异常", e);
-        }
-        return AjaxResult.error(Constants.SERVER_ERROR_MSG);
-    }*/
-
-
 
 
 

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

@@ -8,7 +8,9 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Repository;
 import javax.annotation.Resource;
+import java.sql.Timestamp;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 
@@ -100,7 +102,7 @@ public class CtrlDao {
      * @return
      */
     public boolean batchInsertCtrlTaskInfo(List<Object[]> ctrlTaskInfoList){
-        String sql = "insert into t_ctrl_task_info (rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, sqrxm, sqrsfzh, sqrjh, sqrdwdm, sqrdwmc, bkqyxx, ryxm, zjlxdm, zjhm, hjd, sjh, yzzt, xszt, rylb, rybq, gkjb, gkjbmc, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, zrbmdm, cjsj) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+        String sql = "insert into t_ctrl_task_info (rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, sqr_xm, sqr_sfzh, sqr_jh, sqr_dwbm, sqr_dwmc, bkqyxx, ryxm, zjlxdm, zjhm, hjd, sjh, yzzt, xszt, rylb, rybq, gkjb, gkjbmc, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, zrbmbm, cjsj) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
         try {
             int[] insertRows = gaussDBJdbcTemplate.batchUpdate(sql, ctrlTaskInfoList);
             return insertRows.length == ctrlTaskInfoList.size();
@@ -120,7 +122,7 @@ public class CtrlDao {
      */
     public List<Map<String, Object>> queryCtrlTaskInfo(String taskId, String clue, String clueType){
         List<Map<String, Object>> ctrlTaskInfoList = new ArrayList<>();
-        StringBuilder sql = new StringBuilder("select rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, sqrxm, sqrsfzh, sqrjh, sqrdwdm, sqrdwmc, bkqyxx, ryxm, zjlxdm, zjhm, hjd, sjh, yzzt, xszt, rylb, rybq, gkjb, gkjbmc, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, zrbmdm from t_ctrl_task_info where taskId = '");
+        StringBuilder sql = new StringBuilder("select rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, sqr_xm, sqr_sfzh, sqr_jh, sqr_dwbm, sqr_dwmc, bkqyxx, ryxm, zjlxdm, zjhm, hjd, sjh, yzzt, xszt, rylb, rybq, gkjb, gkjbmc, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, zrbmbm, cjsj from t_ctrl_task_info where rwbh = '");
         sql.append(taskId);
         sql.append("'");
         if(clueCodePhone.equals(clueType)){
@@ -145,7 +147,7 @@ public class CtrlDao {
      * @return
      */
     public boolean batchInsertWarningInfo(List<Object[]> warningInfoList){
-        String sql = "insert into t_ctrl_warning_info (rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, yjztmc, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, zrdwmc, zrbmdm, zrbmmc, sqr_xm, sqr_sfzh, sqr_jh, sqr_dwdm, sqr_dwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, ywly, cjsj, gxsj) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+        String sql = "insert into t_ctrl_warning_info (rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, yjztmc, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, zrdwmc, zrbmbm, zrbmmc, sqr_xm, sqr_sfzh, sqr_jh, sqr_dwbm, sqr_dwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, ywly, cjsj, gxsj) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
         try {
             int[] insertRows = gaussDBJdbcTemplate.batchUpdate(sql, warningInfoList);
             return insertRows.length == warningInfoList.size();
@@ -161,14 +163,14 @@ public class CtrlDao {
      * @param queryDTO
      * @return
      */
-    public List<Map<String, Object>> queryWarningInfo(QueryDTO queryDTO){
+    /*public List<Map<String, Object>> queryWarningInfo(QueryDTO queryDTO){
         String taskId = queryDTO.getTaskId();
         String startTime = queryDTO.getStartTime();
         String endTime = queryDTO.getEndTime();
         Integer pageNum = queryDTO.getPageNum() == null ? Constants.NUM_1 : queryDTO.getPageNum();
         Integer pageSize = queryDTO.getPageSize() == null ? queryMaxCount : queryDTO.getPageSize();
         List<Map<String, Object>> warningInfoList = new ArrayList<>();
-        StringBuilder sql = new StringBuilder("select rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, yjztmc, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, zrdwmc, zrbmdm, zrbmmc, sqr_xm, sqr_sfzh, sqr_jh, sqr_dwdm, sqr_dwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, ywly, cjsj, gxsj from t_ctrl_warning_info where 1 = 1");
+        StringBuilder sql = new StringBuilder("select rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, yjztmc, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, zrdwmc, zrbmbm, zrbmmc, sqr_xm, sqr_sfzh, sqr_jh, sqr_dwbm, sqr_dwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, ywly, cjsj, gxsj from t_ctrl_warning_info where 1 = 1");
         // 1.根据任务ID过滤
         if(StringUtils.isNotBlank(taskId)){
             sql.append(" and rwbh = '");
@@ -197,24 +199,72 @@ public class CtrlDao {
             log.error("【查询布控任务预警结果数据发生异常】 异常信息：", e);
         }
         return warningInfoList;
+    }*/
+
+
+    /**
+     * 根据 系统ID 获取预警数据查询记录表中对应的最新一次查询时间（绝对秒）
+     * @param sysId
+     * @return
+     */
+    public Timestamp getLastQueryTimeBySysId(String sysId){
+        String sql = "select last_query_time from t_ctrl_warning_query_info where sysId = ?";
+        try {
+            return (Timestamp) gaussDBJdbcTemplate.queryForMap(sql, sysId).get("last_query_time");
+        } catch (Exception e){
+            log.error("【根据系统ID获取预警数据查询记录表中对应的最新一次查询时间发生异常】 异常信息：", e);
+        }
+        return null;
     }
 
 
     /**
-     * 根据 任务编号、线索值 查询布控任务对应的线索信息
-     * @param taskId
+     * 基于最新一次查询时间，计算当前预警数据表中的最大入库时间（尽量减少参与计算预警数据中最大入库时间的数据量）
      * @return
      */
-    /*public List<Map<String, Object>> queryClueDetailInfo(String taskId, String clue){
-        List<Map<String, Object>> clueInfoDetailList = new ArrayList<>();
-        String sql = "select * from t_ctrl_clue_info where rwbh = ? and clue = ?";
+    public Timestamp getNewMaxCreateTime(Timestamp lastQueryTime){
+        String sql = "select max(cjsj) max_cjsj from t_ctrl_warning_info where cjsj > ?";
         try {
-            clueInfoDetailList = gaussDBJdbcTemplate.queryForList(sql, taskId, clue);
+            return (Timestamp) gaussDBJdbcTemplate.queryForMap(sql, lastQueryTime).get("max_cjsj");
         } catch (Exception e){
-            log.error("【查询布控任务对应线索详情信息发生异常】 异常信息：", e);
+            log.error("【基于最新一次查询时间，计算当前预警数据表中的最大入库时间发生异常】 异常信息：", e);
         }
-        return clueInfoDetailList;
-    }*/
+        return null;
+    }
+
+
+    /**
+     * 增量查询布控任务预警结果数据
+     * @param lastQueryTime
+     * @return
+     */
+    public List<Map<String, Object>> queryWarningInfoByInc(Timestamp lastQueryTime, Timestamp currentMaxCreateTime){
+        List<Map<String, Object>> warningInfoList = new ArrayList<>();
+        String sql = "select rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, yjztmc, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, zrdwmc, zrbmbm, zrbmmc, sqr_xm, sqr_sfzh, sqr_jh, sqr_dwbm, sqr_dwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, ywly, cjsj, gxsj from t_ctrl_warning_info where yjsj > ? and yjsj <= ? order by yjsj desc";
+        try {
+            warningInfoList = gaussDBJdbcTemplate.queryForList(sql, lastQueryTime, currentMaxCreateTime);
+            log.info("【增量查询布控任务预警结果数据成功】 数据量：{}", warningInfoList.size());
+        } catch (Exception e){
+            log.error("【增量查询布控任务预警结果数据发生异常】 异常信息：", e);
+        }
+        return warningInfoList;
+    }
+
+
+    /**
+     * 根据 系统ID 更新预警数据查询记录表中对应的最新一次查询时间（绝对秒）
+     * @return
+     */
+    public boolean updateLastQueryTimeBySysId(Timestamp currentMaxCreateTime, String sysId){
+        String sql = "update t_ctrl_warning_query_info set last_query_time = ? where where sysId = ?";
+        try {
+            int updateRows = gaussDBJdbcTemplate.update(sql, currentMaxCreateTime, sysId);
+            return updateRows > 0;
+        } catch (Exception e){
+            log.error("【根据系统ID更新预警数据查询记录表中对应的最新一次查询时间发生异常】 异常信息：", e);
+        }
+        return false;
+    }
 
 
     /**

src/main/java/com/hhwy/qbeqsjy/dto/CtrlDTO.java

@@ -17,6 +17,8 @@ import java.util.Map;
 @NoArgsConstructor
 public class CtrlDTO {
 
+    // 任务标识（样例："taskFlag":"DRYJ" || "taskFlag":"JJYJ" || 参数为空则代表当前任务配置中包含单人预警、聚集预警两种模型）
+    private String taskFlag;
     // 任务编号
     private String taskId;
     // 开始布控时间
@@ -40,4 +42,5 @@ public class CtrlDTO {
     //          }
     //       ]
 
+
 }

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

@@ -18,10 +18,7 @@ import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.stereotype.Service;
 import javax.annotation.Resource;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -101,8 +98,11 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                     // 每张布控资源表对应的命中数据JSON对象
                     JSONObject oriField = message.getJSONObject("ORIFIELD");
                     String yjxxsm = oriField.toJSONString();
-                    // 命中时间不采用原始预警数据中的【截获时间】，而采用该服务消费到当前数据时的自然时间作为命中时间
-                    String yjsj = TimeTool.timeStampToDateString(TimeTool.getNowTimeStamp());
+                    Date yjsj = new Date();
+                    // 生成预警信息编号
+                    String yjxxbh = String.valueOf(generator.nextId());
+                    // 预警状态默认为0（未读）
+                    Integer yjzt = 0;
                     // 根据clueId中的任务编号、线索值、线索类型查询对应的布控任务相关信息（包含任务信息、区域信息、线索信息）
                     List<Map<String, Object>> ctrlTaskInfoList = ctrlDao.queryCtrlTaskInfo(rwbh, clue, clueType);
                     if(ctrlTaskInfoList.size() > 0){
@@ -111,17 +111,17 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                         Integer rwlx = (Integer) ctrlTaskInfo.get("rwlx");
                         String rwlxmc = String.valueOf(ctrlTaskInfo.get("rwlxmc"));
                         String rwdxlxmc = String.valueOf(ctrlTaskInfo.get("rwdxlxmc"));
-                        String sqrxm = String.valueOf(ctrlTaskInfo.get("sqrxm"));
-                        String sqrsfzh = String.valueOf(ctrlTaskInfo.get("sqrsfzh"));
-                        String sqrjh = String.valueOf(ctrlTaskInfo.get("sqrjh"));
-                        String sqrdwdm = String.valueOf(ctrlTaskInfo.get("sqrdwdm"));
-                        String sqrdwmc = String.valueOf(ctrlTaskInfo.get("sqrdwmc"));
+                        String sqrXm = String.valueOf(ctrlTaskInfo.get("sqr_xm"));
+                        String sqrSfzh = String.valueOf(ctrlTaskInfo.get("sqr_sfzh"));
+                        String sqrJh = String.valueOf(ctrlTaskInfo.get("sqr_jh"));
+                        String sqrDwbm = String.valueOf(ctrlTaskInfo.get("sqr_dwbm"));
+                        String sqrDwmc = String.valueOf(ctrlTaskInfo.get("sqr_dwmc"));
                         String bkqyxx = String.valueOf(ctrlTaskInfo.get("bkqyxx"));
-                        String yjry_xm = String.valueOf(ctrlTaskInfo.get("ryxm"));
-                        String yjry_sfzh = String.valueOf(ctrlTaskInfo.get("zjhm"));
-                        String yjry_sjh = String.valueOf(ctrlTaskInfo.get("sjh"));
-                        String yjry_lb = String.valueOf(ctrlTaskInfo.get("rylb"));
-                        String yjry_bq = String.valueOf(ctrlTaskInfo.get("rybq"));
+                        String yjryXm = String.valueOf(ctrlTaskInfo.get("ryxm"));
+                        String yjrySfzh = String.valueOf(ctrlTaskInfo.get("zjhm"));
+                        String yjrySjh = String.valueOf(ctrlTaskInfo.get("sjh"));
+                        String yjryLb = String.valueOf(ctrlTaskInfo.get("rylb"));
+                        String yjryBq = String.valueOf(ctrlTaskInfo.get("rybq"));
                         String gkjb = String.valueOf(ctrlTaskInfo.get("gkjb"));
                         String gkjbmc = String.valueOf(ctrlTaskInfo.get("gkjbmc"));
                         String zrjz = String.valueOf(ctrlTaskInfo.get("zrjz"));
@@ -129,11 +129,8 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                         String zrmjxm = String.valueOf(ctrlTaskInfo.get("zrmjxm"));
                         String zrmjjh = String.valueOf(ctrlTaskInfo.get("zrmjjh"));
                         String zrmjlxfs = String.valueOf(ctrlTaskInfo.get("zrmjlxfs"));
-                        String zrdwdm = String.valueOf(ctrlTaskInfo.get("zrdwdm"));
-                        String zrbmdm = String.valueOf(ctrlTaskInfo.get("zrbmdm"));
-                        // 生成预警信息编号
-                        String yjxxbh = String.valueOf(generator.nextId());
-                        Integer yjzt = 0;
+                        String zrdwbm = String.valueOf(ctrlTaskInfo.get("zrdwbm"));
+                        String zrbmbm = String.valueOf(ctrlTaskInfo.get("zrbmbm"));
                         // 如果布控资源为AUTH表
                         if(Constants.TABLE_AUTH_RES_ID.equals(resourceId)){
                             // 1. ========================= 过滤基站号为空的数据
@@ -159,7 +156,7 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                                         String centerLat = loc.getString("lat");
                                         String radius = loc.getString("radius");
                                         if(GeoUtils.isPointInCircle(Double.valueOf(lat), Double.valueOf(lon), Double.valueOf(centerLat), Double.valueOf(centerLon), Double.valueOf(radius))){
-                                            warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, CommonUtil.getConfValueByKey(zrdwdm), zrbmdm, CommonUtil.getConfValueByKey(zrbmdm), sqrxm, sqrsfzh, sqrjh, sqrdwdm, sqrdwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
+                                            warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjryXm, yjrySfzh, yjrySjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, CommonUtil.getConfValueByKey(zrdwbm), zrbmbm, CommonUtil.getConfValueByKey(zrbmbm), sqrXm, sqrSfzh, sqrJh, sqrDwbm, sqrDwmc, yjryLb, yjryBq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
                                             break;
                                         }
                                     } else {   // 布控区域是一般多边形
@@ -171,16 +168,16 @@ public class ConsumeService implements ApplicationListener<ContextRefreshedEvent
                                             coordinates.add(new Coordinate(Double.valueOf(vertexLon), Double.valueOf(vertexLat)));
                                         }
                                         if(GeoUtils.isPointInPolygon(coordinates, new Coordinate(Double.valueOf(lon), Double.valueOf(lat)))){
-                                            warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, CommonUtil.getConfValueByKey(zrdwdm), zrbmdm, CommonUtil.getConfValueByKey(zrbmdm), sqrxm, sqrsfzh, sqrjh, sqrdwdm, sqrdwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
+                                            warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjryXm, yjrySfzh, yjrySjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, CommonUtil.getConfValueByKey(zrdwbm), zrbmbm, CommonUtil.getConfValueByKey(zrbmbm), sqrXm, sqrSfzh, sqrJh, sqrDwbm, sqrDwmc, yjryLb, yjryBq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
                                             break;
                                         }
                                     }
                                 }
                             } else {   // 如果不存在布控区域信息，则默认布控区域范围为全天津市，直接入库
-                                warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, CommonUtil.getConfValueByKey(zrdwdm), zrbmdm, CommonUtil.getConfValueByKey(zrbmdm), sqrxm, sqrsfzh, sqrjh, sqrdwdm, sqrdwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
+                                warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjryXm, yjrySfzh, yjrySjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, CommonUtil.getConfValueByKey(zrdwbm), zrbmbm, CommonUtil.getConfValueByKey(zrbmbm), sqrXm, sqrSfzh, sqrJh, sqrDwbm, sqrDwmc, yjryLb, yjryBq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
                             }
                         } else {   // 如果是其他类布控资源，则直接保存入库
-                            warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjry_xm, yjry_sfzh, yjry_sjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, CommonUtil.getConfValueByKey(zrdwdm), zrbmdm, CommonUtil.getConfValueByKey(zrbmdm), sqrxm, sqrsfzh, sqrjh, sqrdwdm, sqrdwmc, yjry_lb, yjry_bq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
+                            warningInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, yjxxbh, yjryXm, yjrySfzh, yjrySjh, yjxxsm, yjzt, Constants.YJZT_WD, yjsj, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, CommonUtil.getConfValueByKey(zrdwbm), zrbmbm, CommonUtil.getConfValueByKey(zrbmbm), sqrXm, sqrSfzh, sqrJh, sqrDwbm, sqrDwmc, yjryLb, yjryBq, gkjb, gkjbmc, resourceId, yjsj, yjsj});
                         }
                     }
                 }

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

@@ -6,7 +6,6 @@ import com.hhwy.qbeqsjy.common.*;
 import com.hhwy.qbeqsjy.dao.CtrlDao;
 import com.hhwy.qbeqsjy.domain.AjaxResult;
 import com.hhwy.qbeqsjy.dto.CtrlDTO;
-import com.hhwy.qbeqsjy.dto.QueryDTO;
 import com.hhwy.qbeqsjy.task.CtrlTask;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -15,6 +14,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import javax.annotation.Resource;
 import java.io.*;
+import java.sql.Timestamp;
 import java.util.*;
 import java.util.concurrent.*;
 
@@ -56,10 +56,10 @@ public class CtrlService {
 
 
     /**
-     * 处理布控任务
+     * 处理单人预警布控任务
      * @param ctrlDTO
      */
-    public void processCtrl(CtrlDTO ctrlDTO) {
+    public void processDRYJCtrl(CtrlDTO ctrlDTO) {
         String taskId = ctrlDTO.getTaskId();
         String endCtrlTime = ctrlDTO.getEndCtrlTime();
         List<Map<String, Object>> clueList = ctrlDTO.getClueList();
@@ -170,6 +170,59 @@ public class CtrlService {
     }
 
 
+    /**
+     * 处理聚集预警布控任务
+     * @param ctrlDTO
+     */
+    public void processJJYJCtrl(CtrlDTO ctrlDTO) {
+        String taskId = ctrlDTO.getTaskId();
+        List<Map<String, Object>> clueList = ctrlDTO.getClueList();
+        List<String> resourceIdList = ctrlDTO.getResourceIdList();
+        ExecutorService executorService = Executors.newFixedThreadPool(Constants.NUM_1);
+        executorService.execute(() -> {
+            Date createTime = new Date();
+            List<Object[]> ctrlErrorList = new ArrayList<>();
+            int ctrlNum = 0;
+            HttpClient httpClient = new HttpClient();
+            for (Map<String, Object> clueMap : clueList) {
+                if(!clueMap.isEmpty()){
+                    // 重新组装前端传入的参数
+                    String clue = String.valueOf(clueMap.get("clue"));
+                    String clueType = String.valueOf(clueMap.get("clueType"));
+                    clueMap.put("taskId", taskId);
+                    clueMap.put("resourceIdList", resourceIdList);
+                    AjaxResult result = doCtrl(httpClient, clueMap);
+                    int code = (int) result.get("code");
+                    if(code == 500){
+                        // 记录布控失败的线索信息，等待当前任务布控操作执行结束后再批量入库，后续通过设置定时任务按照记录的布控失败线索信息继续实施布控
+                        // 存在操作延迟，但保证数据最终一致性
+                        Object[] errorData = new Object[]{taskId, clue, clueType, resourceIdList, createTime};
+                        ctrlErrorList.add(errorData);
+                    } else {
+                        JSONObject data = (JSONObject) result.get("data");
+                        if(data == null){
+                            String msg = (String) result.get("msg");
+                            log.info("任务ID：{}，线索值：{}，布控服务返回信息：{}", taskId, clue, msg);
+                        } else {
+                            String clueId = data.getString("clueId");
+                            log.info("任务ID：{}，线索值：{}，线索ID：{}", taskId, clue, clueId);
+                        }
+                    }
+                    ctrlNum++;
+                }
+                // 每布控5000条线索，当前线程睡眠30s，减轻【统一布控】服务端压力
+                if(ctrlNum == Constants.NUM_5000){
+                    CommonUtil.threadSleep(Constants.NUM_30000);
+                    ctrlNum = 0;
+                }
+            }
+            log.info("--- > 任务ID：{}，线索集合布控完成", taskId);
+
+        });
+        executorService.shutdown();
+    }
+
+
     /**
      * 处理停控任务
      * @param taskId
@@ -478,27 +531,24 @@ public class CtrlService {
 
     /**
      * 保存用户域布控任务相关信息
-     * @param reqBody
+     * @param taskInfo
      * @return
      */
     @Transactional
-    public boolean saveCtrlTaskInfo(JSONObject reqBody){
+    public boolean saveCtrlTaskInfo(JSONObject taskInfo, JSONArray areaInfo, JSONArray clueInfo){
         List<Object[]> ctrlTaskInfoList = new ArrayList<>();
-        JSONObject taskInfo = reqBody.getJSONObject("taskInfo");
         String rwbh = taskInfo.getString("rwbh");
         String rwmc = taskInfo.getString("rwmc");
         Integer rwlx = taskInfo.getInteger("rwlx");
         String rwlxmc = taskInfo.getString("rwlxmc");
         String rwdxlxmc = taskInfo.getString("rwdxlxmc");
-        String sqrxm = taskInfo.getString("sqrxm");
-        String sqrsfzh = taskInfo.getString("sqrsfzh");
-        String sqrjh = taskInfo.getString("sqrjh");
-        String sqrdwdm = taskInfo.getString("sqrdwdm");
-        String sqrdwmc = taskInfo.getString("sqrdwmc");
-        JSONArray areaList = reqBody.getJSONArray("areaList");
-        JSONArray clueList = reqBody.getJSONArray("clueList");
-        String cjsj = TimeTool.timeStampToDateString(TimeTool.getNowTimeStamp());
-        for (Object item : clueList) {
+        String sqrXm = taskInfo.getString("sqrXm");
+        String sqrSfzh = taskInfo.getString("sqrSfzh");
+        String sqrJh = taskInfo.getString("sqrJh");
+        String sqrDwbm = taskInfo.getString("sqrDwbm");
+        String sqrDwmc = taskInfo.getString("sqrDwmc");
+        Date cjsj = new Date();
+        for (Object item : clueInfo) {
             JSONObject clueObj = (JSONObject) item;
             if(!clueObj.isEmpty()){
                 String ryxm = clueObj.getString("ryxm");
@@ -517,10 +567,9 @@ public class CtrlService {
                 String zrmjxm = clueObj.getString("zrmjxm");
                 String zrmjjh = clueObj.getString("zrmjjh");
                 String zrmjlxfs = clueObj.getString("zrmjlxfs");
-                String zrdwdm = clueObj.getString("zrdwdm");
-                String zrbmdm = clueObj.getString("zrbmdm");
-                Object[] ctrlTaskInfo = new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, sqrxm, sqrsfzh, sqrjh, sqrdwdm, sqrdwmc, areaList.toJSONString(), ryxm, zjlxdm, zjhm, hjd, sjh, yzzt, xszt, rylb, rybq, gkjb, gkjbmc, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwdm, zrbmdm, cjsj};
-                ctrlTaskInfoList.add(ctrlTaskInfo);
+                String zrdwbm = clueObj.getString("zrdwdm");
+                String zrbmbm = clueObj.getString("zrbmdm");
+                ctrlTaskInfoList.add(new Object[]{rwbh, rwmc, rwlx, rwlxmc, rwdxlxmc, sqrXm, sqrSfzh, sqrJh, sqrDwbm, sqrDwmc, areaInfo.toJSONString(), ryxm, zjlxdm, zjhm, hjd, sjh, yzzt, xszt, rylb, rybq, gkjb, gkjbmc, zrjz, zrmjid, zrmjxm, zrmjjh, zrmjlxfs, zrdwbm, zrbmbm, cjsj});
             }
         }
         return ctrlDao.batchInsertCtrlTaskInfo(ctrlTaskInfoList);
@@ -528,47 +577,22 @@ public class CtrlService {
 
 
     /**
-     * 保存布控任务对应线索人员信息
-     * @param taskId
-     * @param clueList
+     * 增量查询布控任务预警结果数据
      * @return
      */
-    /*@Transactional
-    public boolean saveCtrlClueInfo(String taskId, JSONArray clueList){
-        List<Object[]> ctrlClueInfoList = new ArrayList<>();
-        Date createTime = new Date();
-        for (Object item : clueList) {
-            JSONObject clueObj = (JSONObject) item;
-            if(!clueObj.isEmpty()){
-                String ryxm = clueObj.getString("ryxm");
-                String zjlxdm = clueObj.getString("zjlxdm");
-                String zjhm = clueObj.getString("zjhm");
-                String hjd = clueObj.getString("hjd");
-                String sjh = clueObj.getString("sjh");
-                String xswjbh = clueObj.getString("xswjbh");
-                String xswjm = clueObj.getString("xswjm");
-                Integer yzzt = clueObj.getInteger("yzzt");
-                String yzztmc = clueObj.getString("yzztmc");
-                Integer xszt = clueObj.getInteger("xszt");
-                String rylb = clueObj.getString("rylb");
-                String rybq = clueObj.getString("rybq");
-                String gkjb = clueObj.getString("gkjb");
-                String gkjbmc = clueObj.getString("gkjbmc");
-                Object[] ctrlClueInfo = new Object[]{taskId, ryxm, zjlxdm, zjhm, hjd, sjh, xswjbh, xswjm, yzzt, yzztmc, xszt, rylb, rybq, gkjb, gkjbmc, createTime};
-                ctrlClueInfoList.add(ctrlClueInfo);
-            }
+    public List<Map<String, Object>> getWarningInfoByInc(){
+        // 1.获取预警数据查询记录表中最新一次查询时间
+        Timestamp lastQueryTime = ctrlDao.getLastQueryTimeBySysId(Constants.DEFAULT_SYSID);
+        // 2.基于最新一次查询时间，查询此刻预警结果表中最大的创建时间
+        Timestamp currentMaxCreateTime = ctrlDao.getNewMaxCreateTime(lastQueryTime);
+        // 3.查询该时间段内的增量预警数据
+        if(lastQueryTime != null && currentMaxCreateTime != null){
+            List<Map<String, Object>> warningInfoList = ctrlDao.queryWarningInfoByInc(lastQueryTime, currentMaxCreateTime);
+            // 4.更新预警数据查询记录表中最新一次查询时间
+            ctrlDao.updateLastQueryTimeBySysId(currentMaxCreateTime, Constants.DEFAULT_SYSID);
+            return warningInfoList;
         }
-        return ctrlDao.batchInsertCtrlClueInfo(ctrlClueInfoList);
-    }*/
-
-
-    /**
-     * 条件查询布控任务预警结果数据
-     * @param queryDTO
-     * @return
-     */
-    public List<Map<String, Object>> getWarningInfo(QueryDTO queryDTO){
-        return ctrlDao.queryWarningInfo(queryDTO);
+        return new ArrayList<>(0);
     }