增加启动时重新发布通道抽稀图层

src/main/java/com/genersoft/iot/vmp/gb28181/bean/VectorTileSource.java

@@ -2,15 +2,18 @@ package com.genersoft.iot.vmp.gb28181.bean;
 
 import lombok.Getter;
 import lombok.Setter;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.ArrayList;
-import java.util.Map;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
 
 @Getter
 @Setter
-public class VectorTileSource {
+public class VectorTileSource implements Delayed {
 
     /**
      * 抽稀的图层数据
@@ -22,6 +25,8 @@ public class VectorTileSource {
      */
     private List<CommonGBChannel> channelList = new ArrayList<>();
 
+    private String id;
+
     /**
      * 创建时间， 大于6小时后删除
      */
@@ -30,4 +35,14 @@ public class VectorTileSource {
     public VectorTileSource() {
         this.time = System.currentTimeMillis();
     }
+
+    @Override
+    public long getDelay(@NotNull TimeUnit unit) {
+        return unit.convert(time + 6 * 60 * 60 * 1000 - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(@NotNull Delayed o) {
+        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
+    }
 }

src/main/java/com/genersoft/iot/vmp/gb28181/controller/ChannelController.java

@@ -8,7 +8,7 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.controller.bean.*;
 import com.genersoft.iot.vmp.gb28181.service.IGbChannelPlayService;
 import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
-import com.genersoft.iot.vmp.gb28181.utils.VectorTileUtils;
+import com.genersoft.iot.vmp.gb28181.utils.VectorTileCatch;
 import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.utils.DateUtil;
@@ -59,6 +59,9 @@ public class ChannelController {
     @Autowired
     private UserSetting userSetting;
 
+    @Autowired
+    private VectorTileCatch vectorTileCatch;
+
 
     @Operation(summary = "查询通道信息", security = @SecurityRequirement(name = JwtUtils.HEADER))
     @Parameter(name = "id", description = "通道的数据库自增Id", required = true)
@@ -525,7 +528,7 @@ public class ChannelController {
     @Parameter(name = "id", description = "抽稀ID", required = true)
     @GetMapping("/map/thin/clear")
     public void clearThin(String id){
-        VectorTileUtils.INSTANCE.remove(id);
+        vectorTileCatch.remove(id);
     }
 
     @Operation(summary = "保存的抽稀结果", security = @SecurityRequirement(name = JwtUtils.HEADER))
@@ -574,7 +577,7 @@ public class ChannelController {
             thinId = "DEFAULT";
         }
         String catchKey = z + "_" + x + "_" + y + "_" + geoCoordSys.toUpperCase();
-        byte[] mvt = VectorTileUtils.INSTANCE.getVectorTile(thinId, catchKey);
+        byte[] mvt = vectorTileCatch.getVectorTile(thinId, catchKey);
         HttpHeaders headers = new HttpHeaders();
         headers.setContentType(MediaType.parseMediaType("application/x-protobuf"));
         if (mvt == null) {

src/main/java/com/genersoft/iot/vmp/gb28181/dao/provider/ChannelProvider.java

@@ -925,7 +925,8 @@ public class ChannelProvider {
     public String queryAllWithPosition(Map<String, Object> params ){
         return BASE_SQL + " where channel_type = 0 " +
                 " AND coalesce(gb_longitude, longitude) > 0" +
-                " AND coalesce(gb_latitude,  latitude) > 0";
+                " AND coalesce(gb_latitude,  latitude) > 0 " +
+                " ORDER BY map_level";
     }
 
     public String queryListInExtent(Map<String, Object> params ){

src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/GbChannelServiceImpl.java

@@ -17,7 +17,7 @@ import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
 import com.genersoft.iot.vmp.gb28181.event.channel.ChannelEvent;
 import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
 import com.genersoft.iot.vmp.gb28181.service.IPlatformChannelService;
-import com.genersoft.iot.vmp.gb28181.utils.VectorTileUtils;
+import com.genersoft.iot.vmp.gb28181.utils.VectorTileCatch;
 import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
 import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
 import com.genersoft.iot.vmp.utils.Coordtransform;
@@ -34,6 +34,7 @@ import org.locationtech.jts.geom.GeometryFactory;
 import org.locationtech.jts.geom.Point;
 import org.springframework.beans.BeanWrapperImpl;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -42,11 +43,12 @@ import org.springframework.util.ObjectUtils;
 
 import java.time.Duration;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 
 @Slf4j
 @Service
-public class GbChannelServiceImpl implements IGbChannelService {
+public class GbChannelServiceImpl implements IGbChannelService, CommandLineRunner {
 
     @Autowired
     private EventPublisher eventPublisher;
@@ -72,8 +74,38 @@ public class GbChannelServiceImpl implements IGbChannelService {
     @Autowired
     private RedisTemplate<Object, Object> redisTemplate;
 
+    @Autowired
+    private VectorTileCatch vectorTileCatch;
+
     private final GeometryFactory geometryFactory = new GeometryFactory();
 
+
+    @Override
+    public void run(String... args) throws Exception {
+        // 启动时重新发布抽稀图层
+        List<CommonGBChannel> channelList = commonGBChannelMapper.queryAllWithPosition();
+        Map<Integer, List<CommonGBChannel>> zoomCameraMap = new ConcurrentHashMap<>();
+
+        channelList.stream().forEach(commonGBChannel ->  {
+            if (commonGBChannel.getMapLevel() == null) {
+                return;
+            }
+            List<CommonGBChannel> channelListForZoom = zoomCameraMap.computeIfAbsent(commonGBChannel.getMapLevel(), k -> new ArrayList<>());
+            channelListForZoom.add(commonGBChannel);
+        });
+
+        String id = "DEFAULT";
+        List<CommonGBChannel> beforeData = new ArrayList<>();
+        for (Integer zoom : zoomCameraMap.keySet()) {
+            beforeData.addAll(zoomCameraMap.get(zoom));
+            log.info("[抽稀-发布mvt矢量瓦片] ID：{}，当前层级： {}, ", id, zoom);
+            // 按照 z/x/y 数据组织数据， 矢量数据暂时保存在内存中
+            // 按照范围生成 x y范围，
+            saveTile(id, zoom, "WGS84", beforeData);
+            saveTile(id, zoom, "GCJ02", beforeData);
+        }
+    }
+
     @Override
     public CommonGBChannel queryByDeviceId(String gbDeviceId) {
         List<CommonGBChannel> commonGBChannels = commonGBChannelMapper.queryByDeviceId(gbDeviceId);
@@ -1069,25 +1101,25 @@ public class GbChannelServiceImpl implements IGbChannelService {
                     saveProcess(id, process.get(), "抽稀图层： " + zoom);
                 }
 
-                // 抽稀完成, 对数据生成mvt矢量瓦片
+                // 抽稀完成, 对数据发布mvt矢量瓦片
                 List<CommonGBChannel> beforeData = new ArrayList<>();
-                for (Integer key : zoomCameraMap.keySet()) {
+                for (Integer zoom : zoomCameraMap.keySet()) {
-                    beforeData.addAll(zoomCameraMap.get(key));
+                    beforeData.addAll(zoomCameraMap.get(zoom));
-                    log.info("[抽稀-生成mvt矢量瓦片] ID：{}，当前层级： {}", id, key);
+                    log.info("[抽稀-发布mvt矢量瓦片] ID：{}，当前层级： {}", id, zoom);
                     // 按照 z/x/y 数据组织数据， 矢量数据暂时保存在内存中
                     // 按照范围生成 x y范围，
-                    saveTile(id, key, "WGS84", beforeData);
+                    saveTile(id, zoom, "WGS84", beforeData);
-                    saveTile(id, key, "GCJ02", beforeData);
+                    saveTile(id, zoom, "GCJ02", beforeData);
                     process.updateAndGet(v -> (v + 0.5 / zoomParam.size()));
-                    saveProcess(id, process.get(), "发布矢量瓦片： " + key);
+                    saveProcess(id, process.get(), "发布矢量瓦片： " + zoom);
                 }
                 // 记录原始数据，未保存做准备
-                VectorTileUtils.INSTANCE.addSource(id, new ArrayList<>(useCameraMap.values()));
+                vectorTileCatch.addSource(id, new ArrayList<>(useCameraMap.values()));
 
                 log.info("[抽稀完成] ID：{}, 耗时： {}ms", id, (System.currentTimeMillis() - time));
                 saveProcess(id, 1, "抽稀完成");
             } catch (Exception e) {
-
+                log.info("[抽稀] 失败 ID：{}", id, e);
             }
 
         }, 1);
@@ -1121,7 +1153,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
             encoder.addFeature("points", beanMap, pointGeom);
         });
         encoderMap.forEach((key, encoder) -> {
-            VectorTileUtils.INSTANCE.addVectorTile(id, key, encoder.encode());
+            vectorTileCatch.addVectorTile(id, key, encoder.encode());
         });
     }
 
@@ -1141,7 +1173,7 @@ public class GbChannelServiceImpl implements IGbChannelService {
     @Transactional
     public void saveThin(String id) {
         commonGBChannelMapper.resetLevel();
-        List<CommonGBChannel> channelList = VectorTileUtils.INSTANCE.getChannelList(id);
+        List<CommonGBChannel> channelList = vectorTileCatch.getChannelList(id);
         if (channelList != null && !channelList.isEmpty()) {
             int limitCount = 1000;
             if (channelList.size() > limitCount) {
@@ -1156,6 +1188,6 @@ public class GbChannelServiceImpl implements IGbChannelService {
                 commonGBChannelMapper.saveLevel(channelList);
             }
         }
-        VectorTileUtils.INSTANCE.save(id);
+        vectorTileCatch.save(id);
     }
 }

src/main/java/com/genersoft/iot/vmp/gb28181/utils/VectorTileCatch.java

@@ -2,21 +2,30 @@ package com.genersoft.iot.vmp.gb28181.utils;
 
 import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
 import com.genersoft.iot.vmp.gb28181.bean.VectorTileSource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
 import org.springframework.util.ConcurrentReferenceHashMap;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.TimeUnit;
 
-public enum VectorTileUtils {
+@Slf4j
-    INSTANCE;
+@Component
+public class VectorTileCatch {
 
-    private Map<String, VectorTileSource> vectorTileMap = new ConcurrentReferenceHashMap<>();
+    private final Map<String, VectorTileSource> vectorTileMap = new ConcurrentReferenceHashMap<>();
+    private final DelayQueue<VectorTileSource> delayQueue = new DelayQueue<>();
 
     public void addVectorTile(String id, String key, byte[] content) {
         VectorTileSource vectorTileSource = vectorTileMap.get(id);
         if (vectorTileSource == null) {
             vectorTileSource = new VectorTileSource();
+            vectorTileSource.setId(id);
             vectorTileMap.put(id, vectorTileSource);
+            delayQueue.offer(vectorTileSource);
         }
 
         vectorTileSource.getVectorTileMap().put(key, content);
@@ -33,13 +42,19 @@ public enum VectorTileUtils {
         VectorTileSource vectorTileSource = vectorTileMap.get(id);
         if (vectorTileSource == null) {
             vectorTileSource = new VectorTileSource();
+            vectorTileSource.setId(id);
             vectorTileMap.put(id, vectorTileSource);
+            delayQueue.offer(vectorTileSource);
         }
         vectorTileMap.get(id).getChannelList().addAll(channelList);
     }
 
 
     public void remove(String id) {
+        VectorTileSource vectorTileSource = vectorTileMap.get(id);
+        if (vectorTileSource != null) {
+            delayQueue.remove(vectorTileSource);
+        }
         vectorTileMap.remove(id);
     }
 
@@ -59,8 +74,22 @@ public enum VectorTileUtils {
             return;
         }
         vectorTileMap.remove(id);
+        delayQueue.remove(vectorTileSource);
         vectorTileMap.put("DEFAULT", vectorTileSource);
     }
 
 
+    // 缓存数据过期检查
+    @Scheduled(fixedDelay = 30, timeUnit = TimeUnit.MINUTES)
+    public void expirationCheck(){
+        while (!delayQueue.isEmpty()) {
+            try {
+                VectorTileSource vectorTileSource = delayQueue.take();
+                vectorTileMap.remove(vectorTileSource.getId());
+            } catch (InterruptedException e) {
+                log.error("[清理过期的抽稀数据] ", e);
+            }
+        }
+    }
+
 }