新增数据透传开关

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/MqttReceiveServiceImpl.java

@@ -82,7 +82,8 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
     private static Map<String, TemperatureAlarmDto> temperatureMap = new HashMap<>();
 
     static IEquipmentSpecificIndexService equipmentSpecificIndexService;
-
+    @Value("${iot.async.flag}")
+    private  boolean iotAsyncExecutorFlag;
     @Autowired
     public void setEquipmentSpecificIndexService(IEquipmentSpecificIndexService equipmentSpecificIndexService) {
         MqttReceiveServiceImpl.equipmentSpecificIndexService = equipmentSpecificIndexService;
@@ -2548,29 +2549,32 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
 
     @PostConstruct
     public void iotAsyncExecutor() {
-        ThreadPoolTaskExecutor workExecutor = new ThreadPoolTaskExecutor();
-        // 设置核心线程数
-        int length = Runtime.getRuntime().availableProcessors();
-        int size = Math.max(length, 80);
-        workExecutor.setCorePoolSize(size * 2);
-        log.info("装备服务初始化，系统线程数：{},运行线程数:{}", length, size);
-        // 设置最大线程数
-        workExecutor.setMaxPoolSize(workExecutor.getCorePoolSize());
-        //配置队列大小
-        workExecutor.setQueueCapacity(Integer.MAX_VALUE);
-        // 设置线程活跃时间（秒）
-        workExecutor.setKeepAliveSeconds(60);
-        // 设置默认线程名称
-        workExecutor.setThreadNamePrefix("装备服务-Iot透传消息消费线程池" + "-");
-        // 等待所有任务结束后再关闭线程池
-        //当调度器shutdown被调用时，等待当前被调度的任务完成
-        workExecutor.setWaitForTasksToCompleteOnShutdown(true);
-        //执行初始化
-        workExecutor.initialize();
-        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
-        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
-        workExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
-        this.dataExecutor = workExecutor;
+        if(iotAsyncExecutorFlag) {
+            System.out.println("-----------------iotAsyncExecutorFlagiotAsyncExecutorFlagiotAsyncExecutorFlagiotAsyncExecutorFlagiotAsyncExecutorFlagiotAsyncExecutorFlagiotAsyncExecutorFlag");
+            ThreadPoolTaskExecutor workExecutor = new ThreadPoolTaskExecutor();
+            // 设置核心线程数
+            int length = Runtime.getRuntime().availableProcessors();
+            int size = Math.max(length, 80);
+            workExecutor.setCorePoolSize(size * 2);
+            log.info("装备服务初始化，系统线程数：{},运行线程数:{}", length, size);
+            // 设置最大线程数
+            workExecutor.setMaxPoolSize(workExecutor.getCorePoolSize());
+            //配置队列大小
+            workExecutor.setQueueCapacity(Integer.MAX_VALUE);
+            // 设置线程活跃时间（秒）
+            workExecutor.setKeepAliveSeconds(60);
+            // 设置默认线程名称
+            workExecutor.setThreadNamePrefix("装备服务-Iot透传消息消费线程池" + "-");
+            // 等待所有任务结束后再关闭线程池
+            //当调度器shutdown被调用时，等待当前被调度的任务完成
+            workExecutor.setWaitForTasksToCompleteOnShutdown(true);
+            //执行初始化
+            workExecutor.initialize();
+            // rejection-policy：当pool已经达到max size的时候，如何处理新任务
+            // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
+            workExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+            this.dataExecutor = workExecutor;
+        }
     }
 
 }

amos-boot-system-equip/src/main/resources/application-dev.properties

@@ -149,4 +149,5 @@ spring.influx.database=iot_platform
 spring.influx.retention_policy=default
 spring.influx.retention_policy_time=30d
 spring.influx.actions=10000
-spring.influx.bufferLimit=20000
\ No newline at end of file
+spring.influx.bufferLimit=20000
+iot.async.flag = false
\ No newline at end of file