临时提交

src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java

@@ -6,7 +6,6 @@ import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.stereotype.Component;
 
-import javax.annotation.PostConstruct;
 import java.time.Instant;
 import java.util.Date;
 import java.util.Map;
@@ -23,17 +22,18 @@ import java.util.concurrent.TimeUnit;
 @Component
 public class DynamicTask {
 
-    private ThreadPoolTaskScheduler threadPoolTaskScheduler;
+    private final ThreadPoolTaskScheduler threadPoolTaskScheduler;
 
     private final Map<String, ScheduledFuture<?>> futureMap = new ConcurrentHashMap<>();
     private final Map<String, Runnable> runnableMap = new ConcurrentHashMap<>();
 
-    @PostConstruct
-    public void DynamicTask() {
+
+    public DynamicTask() {
         threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
         threadPoolTaskScheduler.setPoolSize(300);
         threadPoolTaskScheduler.setWaitForTasksToCompleteOnShutdown(true);
         threadPoolTaskScheduler.setAwaitTerminationSeconds(10);
+        threadPoolTaskScheduler.setThreadNamePrefix("dynamicTask-");
         threadPoolTaskScheduler.initialize();
     }
 
@@ -42,7 +42,6 @@ public class DynamicTask {
      * @param key 任务ID
      * @param task 任务
      * @param cycleForCatalog 间隔 毫秒
-     * @return
      */
     public void startCron(String key, Runnable task, int cycleForCatalog) {
         if(ObjectUtils.isEmpty(key)) {
@@ -74,7 +73,6 @@ public class DynamicTask {
      * @param key 任务ID
      * @param task 任务
      * @param delay 延时 /毫秒
-     * @return
      */
     public void startDelay(String key, Runnable task, int delay) {
         if(ObjectUtils.isEmpty(key)) {

src/main/java/com/genersoft/iot/vmp/conf/ScheduleConfig.java

@@ -1,30 +1,24 @@
 package com.genersoft.iot.vmp.conf;
 
-import org.apache.commons.lang3.concurrent.BasicThreadFactory;
-import org.springframework.context.annotation.Configuration;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.scheduling.annotation.SchedulingConfigurer;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.scheduling.config.ScheduledTaskRegistrar;
-
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadPoolExecutor;
+import org.springframework.stereotype.Component;
 
 /**
  * "@Scheduled"是Spring框架提供的一种定时任务执行机制，默认情况下它是单线程的，在同时执行多个定时任务时可能会出现阻塞和性能问题。
  * 为了解决这种单线程瓶颈问题，可以将定时任务的执行机制改为支持多线程
  */
-@Configuration
+@Component
 public class ScheduleConfig implements SchedulingConfigurer {
 
-	public static final int cpuNum = Runtime.getRuntime().availableProcessors();
+    @Qualifier("taskExecutor")
+	private ThreadPoolTaskExecutor taskExecutor;
 
-	private static final int corePoolSize = cpuNum;
-
-	private static final String threadNamePrefix = "scheduled-task-pool-%d";
 
 	@Override
 	public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
-		taskRegistrar.setScheduler(new ScheduledThreadPoolExecutor(corePoolSize,
-				new BasicThreadFactory.Builder().namingPattern(threadNamePrefix).daemon(true).build(),
-				new ThreadPoolExecutor.CallerRunsPolicy()));
+		taskRegistrar.setScheduler(taskExecutor);
 	}
 }

src/main/java/com/genersoft/iot/vmp/conf/ThreadPoolTaskConfig.java

@@ -45,12 +45,9 @@ public class ThreadPoolTaskConfig {
     /**
      * 线程池名前缀
      */
-    private static final String threadNamePrefix = "wvp-";
+    private static final String threadNamePrefix = "async-";
+
 
-    /**
-     *
-     * @return
-     */
     @Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名
     public ThreadPoolTaskExecutor taskExecutor() {
         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

src/main/java/com/genersoft/iot/vmp/conf/WVPTimerTask.java

@@ -2,12 +2,14 @@ package com.genersoft.iot.vmp.conf;
 
 import com.alibaba.fastjson2.JSONObject;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 @Component
+@Slf4j
 public class WVPTimerTask {
 
     @Autowired
@@ -21,6 +23,7 @@ public class WVPTimerTask {
 
     @Scheduled(fixedDelay = 2 * 1000)   //每3秒执行一次
     public void execute(){
+        log.info("[更新服务信息]");
         JSONObject jsonObject = new JSONObject();
         jsonObject.put("ip", sipConfig.getShowIp());
         jsonObject.put("port", serverPort);

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

@@ -167,7 +167,7 @@ public class DeviceServiceImpl implements IDeviceService {
                 deviceMapper.update(device);
                 redisCatchStorage.updateDevice(device);
             }
-            if (deviceChannelMapper.queryChannelsByDeviceDbId(device.getId()).isEmpty()) {
+            if (deviceChannelMapper.queryChannelsByDeviceDbId(device.getId()).isEmpty() && !catalogResponseMessageHandler.isSyncRunning(device.getDeviceId())) {
                 log.info("[设备上线]: {}，通道数为0,查询通道信息", device.getDeviceId());
                 sync(device);
             }
@@ -221,9 +221,14 @@ public class DeviceServiceImpl implements IDeviceService {
                 sessionManager.removeByCallId(ssrcTransaction.getCallId());
             }
         }
-        // 移除订阅
-        removeCatalogSubscribe(device, null);
-        removeMobilePositionSubscribe(device, null);
+        if(device.getSubscribeCycleForCatalog() > 0) {
+            // 移除订阅
+            removeCatalogSubscribe(device, null);
+        }
+        if(device.getSubscribeCycleForMobilePosition() > 0) {
+            // 移除订阅
+            removeMobilePositionSubscribe(device, null);
+        }
 
         List<AudioBroadcastCatch> audioBroadcastCatches = audioBroadcastManager.getByDeviceId(deviceId);
         if (!audioBroadcastCatches.isEmpty()) {

src/main/java/com/genersoft/iot/vmp/gb28181/task/SipRunner.java

@@ -66,9 +66,13 @@ public class SipRunner implements CommandLineRunner {
 
         for (Device device : deviceList) {
             if (deviceService.expire(device)){
-                deviceService.offline(device.getDeviceId(), "注册已过期");
+                if (device.isOnLine()) {
+                    deviceService.offline(device.getDeviceId(), "注册已过期");
+                }
             }else {
-                deviceService.online(device, null);
+                if (!device.isOnLine()) {
+                    deviceService.online(device, null);
+                }
             }
         }
         // 重置cseq计数