位置数据插入逻辑重构

beidou-track-geomesa/src/main/java/cn/com/taiji/track/dto/BeidouShipLocationDTO.java

@@ -50,8 +50,6 @@ public class BeidouShipLocationDTO extends IGeomesaTrackDTO implements Serializa
     @Override
     public boolean checkPoint() {
         if (location != null) {
-//            longitude = location.substring(location.indexOf("(")+1,location.indexOf(" ",location.indexOf("(")));
-//            latitude = location.substring(location.indexOf(" ",location.indexOf("("))+1,location.indexOf(")"));
             return true;
         }
         return false;

beidou-track-geomesa/src/main/java/cn/com/taiji/track/entity/BeidouLocationEntity.java

@@ -21,12 +21,12 @@ public class BeidouLocationEntity {
 
     @TableId
     private String deviceId;
+    private String locationTime;
     private String location;
     private String shipType;
     private String workType;
     private String workWay;
     private String sendTime;
-    private String locationTime;
     private String online;
     private String shipLength;
     private String shipWidth;

beidou-track-geomesa/src/main/java/cn/com/taiji/track/mapper/BeidouLocationMapper.java

@@ -3,7 +3,9 @@ package cn.com.taiji.track.mapper;
 import cn.com.taiji.track.dto.BeidouShipLocationDTO;
 import cn.com.taiji.track.entity.BeidouLocationEntity;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.apache.ibatis.annotations.Insert;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
 import org.apache.ibatis.annotations.Select;
 
 import java.util.List;
@@ -35,5 +37,16 @@ public interface BeidouLocationMapper extends BaseMapper<BeidouLocationEntity> {
             "\tdevice_id,\n" +
             "\tlocation_time")
     List<BeidouShipLocationDTO> listLocation();
+
+    @Insert("<script>" +
+            "INSERT INTO ax_beidou_ship_location VALUES" +
+            "<foreach collection ='locationList' item='location' separator =','>" +
+            "(#{location.deviceId}, #{location.locationTime}, #{location.location}, #{location.shipType},#{location.workType}" +
+            ",#{location.workWay},#{location.sendTime},#{location.online},#{location.texture}" +
+            ",#{location.longitude},#{location.latitude},#{location.direction},#{location.speed}" +
+            ",#{location.kwh},#{location.shipName},#{location.shipLength},#{location.shipWidth})" +
+            "</foreach>" +
+            "</script>")
+    public void insertLocationList(@Param("locationList") List<BeidouLocationEntity> locationList);
 }
 

beidou-track-geomesa/src/main/java/cn/com/taiji/track/service/impl/BeidouLocationServiceImpl.java

@@ -51,8 +51,8 @@ public class BeidouLocationServiceImpl extends ServiceImpl<BeidouLocationMapper,
     @Autowired
     protected IShipStatusService shipStatusService;
 
-    private Map<String,String> locationMaps= new HashMap<>();
-    private Vector<BeidouLocationEntity> locationEntities= new Vector<>();
+    private Map<String,String> locationMap= new HashMap<>();
+    private Map<String,BeidouLocationEntity> entityMap= new HashMap<>();
 
     @Autowired
     private SimpleFeatureType sftLocation;
@@ -75,12 +75,15 @@ public class BeidouLocationServiceImpl extends ServiceImpl<BeidouLocationMapper,
                     BeidouLocationEntity locationEntity = JSONObject.parseObject(msg.toString(), BeidouLocationEntity.class);
                     String location = StrUtil.format("POINT ({} {})", locationEntity.getLongitude(),locationEntity.getLatitude());
                     locationEntity.setLocation(location);
-//                    BigDecimal longitudeDecimal = new BigDecimal(locationEntity.getLongitude());
-//                    locationEntity.setLongitude(LatLngUtil.latLng2Dfm(longitudeDecimal.doubleValue()));
-//                    BigDecimal latitudeDecimal = new BigDecimal(locationEntity.getLatitude());
-//                    locationEntity.setLatitude(LatLngUtil.latLng2Dfm(latitudeDecimal.doubleValue()));
-                    locationEntities.add(locationEntity);
-                    shipStatusService.pushOnLineDeviceId(locationEntity.getDeviceId(),"true");
+                    String locationTime = locationEntity.getLocationTime();
+                    if (null == entityMap.get(locationEntity.getDeviceId())) {
+                        shipStatusService.pushOnLineDeviceId(locationEntity.getDeviceId(),"true");
+                        entityMap.put(locationEntity.getDeviceId(),locationEntity);
+                    }else if (StringUtils.hasText(locationTime)&& !locationTime.equals(entityMap.get(
+                                    locationEntity.getDeviceId()).getLocationTime())) {
+                        shipStatusService.pushOnLineDeviceId(locationEntity.getDeviceId(),"true");
+                        entityMap.put(locationEntity.getDeviceId(),locationEntity);
+                    }
                 }
             }
         } catch (Exception e) {
@@ -91,9 +94,9 @@ public class BeidouLocationServiceImpl extends ServiceImpl<BeidouLocationMapper,
     @Override
     public void cacheBeidouToMySql() {
         Date begin = new Date();
-        if(locationEntities.size() > 0){
-            List<BeidouLocationEntity> saveList = new ArrayList<>(locationEntities);
-            locationEntities.clear();
+        if(entityMap.size() > 0){
+            List<BeidouLocationEntity> saveList = new ArrayList<>(entityMap.values());
+            entityMap.clear();
             for(BeidouLocationEntity e : saveList){
                 if(!StringUtils.hasText(e.getShipName())){
                     QueryWrapper<BeidouShipArchives> wrapper = new QueryWrapper();
@@ -104,20 +107,17 @@ public class BeidouLocationServiceImpl extends ServiceImpl<BeidouLocationMapper,
                     }
                 }
                 String location = e.getLocation();
-                if (StringUtils.hasText(location) && !location.equals(locationMaps.get(e.getDeviceId()))) {
-//                if (StringUtils.hasText(location)) {
-//                    String longitude = location.substring(location.indexOf("(")+1,location.indexOf(" ",location.indexOf("(")));
-//                    String latitude = location.substring(location.indexOf(" ",location.indexOf("("))+1,location.indexOf(")"));
+                if (StringUtils.hasText(location) && !location.equals(locationMap.get(e.getDeviceId()))) {
                     List<ContainsDTO> result = warningAreaMapper.containsPoint(e.getLongitude(),e.getLatitude());
                     if(result.size()>0){
                         shipStatusService.pushinportDeviceId(e.getDeviceId(), WarningCodeConstants.INPORT_WARNING);
                     }else{
                         shipStatusService.pushinportDeviceId(e.getDeviceId(),WarningCodeConstants.OUTPROT_WARNING);
                     }
-                    locationMaps.put(e.getDeviceId(),e.getLocation());
+                    locationMap.put(e.getDeviceId(),location);
                 }
             }
-            saveBatch(saveList);
+            baseMapper.insertLocationList(saveList);
             log.info("位置数据写入数据库完成！耗时 {} 毫秒, 合计保存 {} 条记录", (new Date()).getTime() - begin.getTime(), saveList.size());
         }
     }

beidou-track-geomesa/src/main/resources/application-prod.yml

@@ -44,7 +44,7 @@ spring:
     # 配置检测连接是否有效
     validationQuery: SELECT 1
   kafka:
-    bootstrap-servers: 10.112.89.239:9092
+    bootstrap-servers: 10.112.89.101:9092,10.112.89.105:9092,10.112.89.239:9092
     producer:
       # 发生错误后，消息重发的次数。
       retries: 0
@@ -71,6 +71,8 @@ spring:
       enable-auto-commit: false
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      max-poll-interval-ms: 600000
+      max-poll-records: 10
     listener:
       # 在侦听器容器中运行的线程数。
       concurrency: 5