多线程

src/main/java/com/shanghaichengdi/controller/PollutionWarningController.java

@@ -2,6 +2,7 @@ package com.shanghaichengdi.controller;
 
 import com.shanghaichengdi.common.result.Result;
 import com.shanghaichengdi.dao.entity.*;
+import com.shanghaichengdi.service.IBoatPollutionWarningHandleService;
 import com.shanghaichengdi.service.IBoatPollutionWarningService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.cache.annotation.Cacheable;
@@ -22,17 +23,17 @@ import java.util.List;
 @RequestMapping("/pollution-warning")
 public class PollutionWarningController {
     @Resource
-    private IBoatPollutionWarningService boatPollutionWarningService;
+    private IBoatPollutionWarningHandleService boatPollutionWarningHandleService;
 
     @PostMapping( "shortBargeHandle")
     public Result<Void> handleShortBarge() {
-        boatPollutionWarningService.handleShortBarge();
+        boatPollutionWarningHandleService.handleShortBarge();
         return Result.success();
     }
 
     @PostMapping("OrdinaryShipHandle")
     public Result<Void> handleOrdinaryShip() {
-        boatPollutionWarningService.handleOrdinaryShip();
+        boatPollutionWarningHandleService.handleOrdinaryShip();
         return Result.success();
     }
 }

src/main/java/com/shanghaichengdi/schedule/PollutionWarningSchedule.java

@@ -1,6 +1,6 @@
 package com.shanghaichengdi.schedule;
 
-import com.shanghaichengdi.service.IBoatPollutionWarningService;
+import com.shanghaichengdi.service.IBoatPollutionWarningHandleService;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -16,15 +16,15 @@ import javax.annotation.Resource;
 public class PollutionWarningSchedule {
 
     @Resource
-    private IBoatPollutionWarningService boatPollutionWarningService;
+    private IBoatPollutionWarningHandleService boatPollutionWarningHaneldService;
 
-    @Scheduled(fixedRate = 10 * 60 * 1000)
+//    @Scheduled(fixedRate = 10 * 60 * 1000)
     public void ShortBarge() {
-        boatPollutionWarningService.handleShortBarge();
+        boatPollutionWarningHaneldService.handleShortBarge();
     }
 
-    @Scheduled(fixedRate = 10 * 60 * 1000)
+//    @Scheduled(fixedRate = 10 * 60 * 1000)
     public void OrdinaryShip() {
-        boatPollutionWarningService.handleOrdinaryShip();
+        boatPollutionWarningHaneldService.handleOrdinaryShip();
     }
 }

src/main/java/com/shanghaichengdi/service/IBoatPollutionWarningHandleService.java

@@ -0,0 +1,7 @@
+package com.shanghaichengdi.service;
+
+public interface IBoatPollutionWarningHandleService {
+    void handleShortBarge();
+
+    void handleOrdinaryShip();
+}

src/main/java/com/shanghaichengdi/service/IBoatPollutionWarningService.java

@@ -1,10 +1,16 @@
 package com.shanghaichengdi.service;
 
 import com.baomidou.mybatisplus.extension.service.IService;
+import com.shanghaichengdi.dao.entity.BoatPollutionSupervision;
 import com.shanghaichengdi.dao.entity.BoatPollutionWarning;
+import com.shanghaichengdi.dao.entity.DictDistrictPort;
+import com.shanghaichengdi.model.vo.BoatAISVO;
+
+import java.util.List;
+import java.util.Map;
 
 public interface IBoatPollutionWarningService extends IService<BoatPollutionWarning> {
-    void handleShortBarge();
+    void handleShortBarge(Map<String, List<BoatAISVO>> subBoatAISVOMap, List<BoatPollutionSupervision> boatPollutionSupervisionList, List<DictDistrictPort> dictDistrictPortList);
 
-    void handleOrdinaryShip();
+    void handleOrdinaryShip(Map<String, List<BoatAISVO>> subBoatAISVOMap, List<BoatPollutionSupervision> boatPollutionSupervisionList, List<DictDistrictPort> dictDistrictPortList);
 }

src/main/java/com/shanghaichengdi/service/impl/BoatPollutionWarningHandleServiceImpl.java

@@ -0,0 +1,180 @@
+package com.shanghaichengdi.service.impl;
+
+import com.shanghaichengdi.common.enums.SupervisionTypeEnum;
+import com.shanghaichengdi.common.exception.BizException;
+import com.shanghaichengdi.common.properties.GpsStartTimeProperties;
+import com.shanghaichengdi.common.properties.WarningProperties;
+import com.shanghaichengdi.common.util.LocalDateTimeUtil;
+import com.shanghaichengdi.dao.entity.BoatPollutionSupervision;
+import com.shanghaichengdi.dao.entity.DictDistrictPort;
+import com.shanghaichengdi.model.vo.BoatAISVO;
+import com.shanghaichengdi.service.*;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+
+/**
+ * @ClassName BoatPollutionWarningHandleServiceImpl
+ * @Author Tevin
+ * @DATE 2025/1/10 13:40
+ * @Description TODO
+ **/
+@Slf4j
+@Service
+public class BoatPollutionWarningHandleServiceImpl implements IBoatPollutionWarningHandleService {
+    @Resource
+    private GpsStartTimeProperties gpsStartTimeProperties;
+    @Resource
+    private WarningProperties warningProperties;
+
+    @Resource
+    private IGpsinfoHistoryService gpsinfoHistoryService;
+
+    @Resource
+    private IGpsinfoRealtimeService gpsinfoRealtimeService;
+
+    @Resource
+    private IDictDistrictPortService dictDistrictPortService;
+
+    @Resource
+    private IBoatPollutionSupervisionService boatPollutionSupervisionService;
+
+    @Resource
+    private IBoatPollutionWarningService boatPollutionWarningService;
+
+    public void handleShortBarge() {
+        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge start==========");
+        LocalDateTime handleStartTime = LocalDateTime.now();
+        Date maxGpsTime = boatPollutionSupervisionService.getMaxGpsTime(SupervisionTypeEnum.SHORT_BARGE.getStatusCode());
+        LocalDateTime startdefaultLocalDateTime = LocalDateTimeUtil.parse(gpsStartTimeProperties.getGpsstartsime());
+        Date startTime = Optional.ofNullable(maxGpsTime).orElse(Date.from(startdefaultLocalDateTime.atZone(ZoneId.systemDefault()).toInstant()));
+        LocalDateTime startLocalDateTime = startTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
+        long diffMillis = Duration.between(startLocalDateTime, LocalDateTime.now()).toMillis();
+        if (diffMillis <= 0) {
+            return;
+        }
+        LocalDateTime endLocalDateTime = this.calSuperviseEndTime(startLocalDateTime, diffMillis);
+        Date endTime = Date.from(endLocalDateTime.atZone(ZoneId.systemDefault()).toInstant());
+        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.getShotBargeAis==========");
+        //查询开始结束时间段内的市内短驳船AIS历史数据进行分析
+//        List<BoatAISVO> boatAISList = gpsinfoHistoryService.getShotBargeAisHistory(startTime, endTime);
+        List<BoatAISVO> boatAISList = gpsinfoRealtimeService.getShotBargeAisRealtime(startTime, endTime);
+        if (boatAISList.isEmpty()) {
+            throw new BizException("未查到船舶AIS数据！");
+        }
+        // 查询全部有效的监管船舶
+        List<BoatPollutionSupervision> boatPollutionSupervisionList = boatPollutionSupervisionService.list();
+        if (boatPollutionSupervisionList.isEmpty()) {
+            throw new BizException("未查到船舶监管记录！");
+        }
+        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.dictDistrictPortList==========");
+        List<DictDistrictPort> dictDistrictPortList = dictDistrictPortService.getDictDistrictPortList();
+        Map<String, List<BoatAISVO>> boatAISVOMap = boatAISList.stream().collect(Collectors.groupingBy(BoatAISVO::getDeviceId));
+        List<Map<String, List<BoatAISVO>>> boatAISVOMaplist = splitMap(boatAISVOMap, 50);
+        boatAISVOMaplist.forEach(subBoatAISVOMap -> CompletableFuture.runAsync(() -> {
+            {
+                boatPollutionWarningService.handleShortBarge(subBoatAISVOMap, boatPollutionSupervisionList, dictDistrictPortList);
+            }
+        }));
+
+        LocalDateTime handleEndTime = LocalDateTime.now();
+        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge crossTime:[" + Duration.between(handleStartTime, handleEndTime).toMillis() + " milliseconds]==========");
+        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge end==========");
+    }
+
+    public void handleOrdinaryShip() {
+        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip start==========");
+        LocalDateTime handleStartTime = LocalDateTime.now();
+        Date maxGpsTime = boatPollutionSupervisionService.getMaxGpsTime(SupervisionTypeEnum.ORDINARY_SHIP_TYPE.getStatusCode());
+        LocalDateTime startdefaultLocalDateTime = LocalDateTimeUtil.parse(gpsStartTimeProperties.getGpsstartsime());
+        Date startTime = Optional.ofNullable(maxGpsTime).orElse(Date.from(startdefaultLocalDateTime.atZone(ZoneId.systemDefault()).toInstant()));
+        LocalDateTime startLocalDateTime = startTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
+        SimpleDateFormat sdf = new SimpleDateFormat(" yyyy-MM-dd HH:mm:ss ");
+        long diffMillis = Duration.between(startLocalDateTime, LocalDateTime.now()).toMillis();
+        if (diffMillis <= 0) {
+            return;
+        }
+        LocalDateTime endLocalDateTime = this.calSuperviseEndTime(startLocalDateTime, diffMillis);
+        Date endTime = Date.from(endLocalDateTime.atZone(ZoneId.systemDefault()).toInstant());
+        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.getShotBargeAis==========");
+        //查询开始结束时间段内的市内短驳船AIS历史数据进行分析
+        List<BoatAISVO> boatAISList = gpsinfoHistoryService.getOrdinaryShipAisHistory(startTime, endTime);
+        if (boatAISList.isEmpty()) {
+            throw new BizException("未查到船舶AIS数据！");
+        }
+        // 查询全部有效的监管船舶
+        List<BoatPollutionSupervision> boatPollutionSupervisionList = boatPollutionSupervisionService.list();
+        if (boatPollutionSupervisionList.isEmpty()) {
+            throw new BizException("未查到船舶监管记录！");
+        }
+        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.dictDistrictPortList==========");
+        List<DictDistrictPort> dictDistrictPortList = dictDistrictPortService.getDictDistrictPortList();
+        Map<String, List<BoatAISVO>> boatAISVOMap = boatAISList.stream().collect(Collectors.groupingBy(BoatAISVO::getDeviceId));
+        List<Map<String, List<BoatAISVO>>> boatAISVOMaplist = splitMap(boatAISVOMap, 50);
+        boatAISVOMaplist.forEach(subBoatAISVOMap -> CompletableFuture.runAsync(() -> {
+            {
+                boatPollutionWarningService.handleOrdinaryShip(subBoatAISVOMap, boatPollutionSupervisionList, dictDistrictPortList);
+            }
+        }));
+        LocalDateTime handleEndTime = LocalDateTime.now();
+        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge crossTime:[" + Duration.between(handleStartTime, handleEndTime).toMillis() + " milliseconds]==========");
+        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip end==========");
+    }
+
+    /**
+     * 计算监控时间
+     *
+     * @param startTime
+     * @param diffMillis
+     * @return
+     */
+    private LocalDateTime calSuperviseEndTime(LocalDateTime startTime, long diffMillis) {
+        int millsAdd = 0;
+        if (diffMillis <= 600000) {
+            millsAdd = 600;
+        } else if (diffMillis <= 1800000) {
+            millsAdd = 1800;
+        } else if (diffMillis <= 3600000) {
+            millsAdd = 3600;
+        } else if (diffMillis <= 5400000) {
+            millsAdd = 5400;
+        } else if (diffMillis <= 7200000) {
+            millsAdd = 7200;
+        } else {
+            millsAdd = 7200;
+        }
+        return startTime.plusSeconds(millsAdd);
+    }
+
+
+    private List<Map<String, List<BoatAISVO>>> splitMap(Map<String, List<BoatAISVO>> map, int splitSize) {
+        List<Map<String, List<BoatAISVO>>> result = new ArrayList<>();
+        int count = 0;
+        Map<String, List<BoatAISVO>> subMap = new HashMap<>();
+
+        for (Map.Entry<String, List<BoatAISVO>> entry : map.entrySet()) {
+            subMap.put(entry.getKey(), entry.getValue());
+            count++;
+
+            if (count == splitSize) {
+                result.add(subMap);
+                subMap = new HashMap<>();
+                count = 0;
+            }
+        }
+
+        if (!subMap.isEmpty()) {
+            result.add(subMap);
+        }
+
+        return result;
+    }
+}