Merge remote-tracking branch 'origin/dev-pano'

src/main/java/com/fdkankan/mqcontroller/entity/MqCameraLevel.java

@@ -53,5 +53,6 @@ public class MqCameraLevel implements Serializable {
     @TableField("update_time")
     private Date updateTime;
 
-
+    @TableField("type")
+    private String type;
 }

src/main/java/com/fdkankan/mqcontroller/entity/MqEcs.java

@@ -55,4 +55,8 @@ public class MqEcs implements Serializable {
     @TableField("rec_status")
     @TableLogic(value = "A",delval = "I")
     private String recStatus;
+
+    // 默认 4dkk，pano 全景看看
+    @TableField("type")
+    private String type;
 }

src/main/java/com/fdkankan/mqcontroller/entity/MqNumLevel.java

@@ -52,6 +52,7 @@ public class MqNumLevel implements Serializable {
 
     @TableField("update_time")
     private Date updateTime;
-
+    @TableField("type")
+    private String type;
 
 }

src/main/java/com/fdkankan/mqcontroller/entity/MqQueueConfig.java

@@ -86,4 +86,6 @@ public class MqQueueConfig implements Serializable {
 
     @TableField("is_special")
     private Integer isSpecial;
+    @TableField("type")
+    private String type;
 }

src/main/java/com/fdkankan/mqcontroller/entity/MqScalingConfig.java

@@ -59,5 +59,6 @@ public class MqScalingConfig implements Serializable {
     @TableField("endpoint")
     private String endpoint;
 
-
+    @TableField("type")
+    private String type;
 }

src/main/java/com/fdkankan/mqcontroller/entity/MqSendLog.java

@@ -63,4 +63,8 @@ public class MqSendLog implements Serializable {
 
     @TableField(exist = false)
     private Integer configId;
+
+    // 默认 4dkk pano 全景看看
+    @TableField("type")
+    private String type;
 }

src/main/java/com/fdkankan/mqcontroller/service/IMqCameraLevelService.java

@@ -13,5 +13,5 @@ import com.baomidou.mybatisplus.extension.service.IService;
  */
 public interface IMqCameraLevelService extends IService<MqCameraLevel> {
 
-    MqCameraLevel getByCameraId(Long cameraId);
+    MqCameraLevel getByCameraId(Long cameraId,String type);
 }

src/main/java/com/fdkankan/mqcontroller/service/IMqEcsService.java

@@ -15,7 +15,7 @@ import java.util.List;
  */
 public interface IMqEcsService extends IService<MqEcs> {
 
-    void add(String queueName);
+    void add(String queueName,String type);
 
     List<MqEcs> getScalingNotStopList();
 

src/main/java/com/fdkankan/mqcontroller/service/IMqNumLevelService.java

@@ -13,5 +13,5 @@ import com.baomidou.mybatisplus.extension.service.IService;
  */
 public interface IMqNumLevelService extends IService<MqNumLevel> {
 
-    MqNumLevel getByNum(String num);
+    MqNumLevel getByNum(String num,String type);
 }

src/main/java/com/fdkankan/mqcontroller/service/IMqQueueConfigService.java

@@ -4,6 +4,7 @@ import com.fdkankan.mqcontroller.entity.MqQueueConfig;
 import com.baomidou.mybatisplus.extension.service.IService;
 
 import java.util.HashMap;
+import java.util.List;
 
 /**
  * <p>
@@ -17,7 +18,9 @@ public interface IMqQueueConfigService extends IService<MqQueueConfig> {
 
     HashMap<String, MqQueueConfig> getQueueMap();
 
-    MqQueueConfig getDfConfig();
+    MqQueueConfig getDfConfig(String type);
 
     MqQueueConfig getDfConfig128();
+
+    List<MqQueueConfig> getByType(String type);
 }

src/main/java/com/fdkankan/mqcontroller/service/impl/MqCameraLevelServiceImpl.java

@@ -19,9 +19,10 @@ import org.springframework.stereotype.Service;
 public class MqCameraLevelServiceImpl extends ServiceImpl<IMqCameraLevelMapper, MqCameraLevel> implements IMqCameraLevelService {
 
     @Override
-    public MqCameraLevel getByCameraId(Long cameraId) {
+    public MqCameraLevel getByCameraId(Long cameraId,String type) {
         LambdaQueryWrapper<MqCameraLevel> wrapper = new LambdaQueryWrapper<>();
         wrapper.eq(MqCameraLevel::getCameraId,cameraId);
+        wrapper.eq(MqCameraLevel::getType,type);
         return this.getOne(wrapper);
     }
 }

src/main/java/com/fdkankan/mqcontroller/service/impl/MqEcsServiceImpl.java

@@ -34,9 +34,10 @@ public class MqEcsServiceImpl extends ServiceImpl<IMqEcsMapper, MqEcs> implement
     @Autowired
     RedisUtil redisUtil;
     @Override
-    public void add(String queueName) {
+    public void add(String queueName,String type) {
         MqEcs mqEcs = new MqEcs();
         mqEcs.setQueueName(queueName);
+        mqEcs.setType(type);
         this.save(mqEcs);
     }
 

src/main/java/com/fdkankan/mqcontroller/service/impl/MqNumLevelServiceImpl.java

@@ -19,9 +19,10 @@ import org.springframework.stereotype.Service;
 public class MqNumLevelServiceImpl extends ServiceImpl<IMqNumLevelMapper, MqNumLevel> implements IMqNumLevelService {
 
     @Override
-    public MqNumLevel getByNum(String num) {
+    public MqNumLevel getByNum(String num,String type) {
         LambdaQueryWrapper<MqNumLevel> wrapper = new LambdaQueryWrapper<>();
         wrapper.eq(MqNumLevel::getNum,num);
+        wrapper.eq(MqNumLevel::getType,type);
         return this.getOne(wrapper);
     }
 }

src/main/java/com/fdkankan/mqcontroller/service/impl/MqQueueConfigServiceImpl.java

@@ -30,9 +30,10 @@ public class MqQueueConfigServiceImpl extends ServiceImpl<IMqQueueConfigMapper,
     }
 
     @Override
-    public MqQueueConfig getDfConfig() {
+    public MqQueueConfig getDfConfig(String type) {
         LambdaQueryWrapper<MqQueueConfig> wrapper = new LambdaQueryWrapper<>();
         wrapper.eq(MqQueueConfig::getIsDefault,1);
+        wrapper.eq(MqQueueConfig::getType,type);
         return this.getOne(wrapper);
     }
 
@@ -42,4 +43,11 @@ public class MqQueueConfigServiceImpl extends ServiceImpl<IMqQueueConfigMapper,
         wrapper.eq(MqQueueConfig::getIsSpecial,1);
         return this.getOne(wrapper);
     }
+
+    @Override
+    public List<MqQueueConfig> getByType(String type) {
+        LambdaQueryWrapper<MqQueueConfig> wrapper = new LambdaQueryWrapper<>();
+        wrapper.eq(MqQueueConfig::getType,type);
+        return this.list(wrapper);
+    }
 }

src/main/java/com/fdkankan/mqcontroller/service/impl/ScenePlusServiceImpl.java

@@ -37,32 +37,22 @@ public class ScenePlusServiceImpl extends ServiceImpl<IScenePlusMapper, ScenePlu
             if(scenePro == null){
                 return null;
             }
-            MqQueueConfig dfConfig = queueConfigService.getDfConfig();
+            MqQueueConfig dfConfig = queueConfigService.getDfConfig(mqSendLog.getType());
             if(dfConfig != null){
                 return dfConfig.getId();
             }
             return null;
         }
-        JSONObject jsonObject = JSONObject.parseObject(mqSendLog.getContent());
-        if(jsonObject.containsKey("ext")){
-            JSONObject ext  = jsonObject.getJSONObject("ext");
-            if(ext != null && ext.containsKey("128G") && ext.getInteger("128G") == 1){
-                MqQueueConfig dfConfig = queueConfigService.getDfConfig128();
-                if(dfConfig != null){
-                    return dfConfig.getId();
-                }
-            }
-        }
 
-        MqNumLevel mqNumLevel = mqNumLevelService.getByNum(num);
+        MqNumLevel mqNumLevel = mqNumLevelService.getByNum(num,mqSendLog.getType());
         if(mqNumLevel != null){
             return mqNumLevel.getQueueConfigId();
         }
-        MqCameraLevel mqCameraLevel = mqCameraLevelService.getByCameraId(scenePlus.getCameraId());
+        MqCameraLevel mqCameraLevel = mqCameraLevelService.getByCameraId(scenePlus.getCameraId(),mqSendLog.getType());
         if(mqCameraLevel != null){
             return mqCameraLevel.getQueueConfigId();
         }
-        MqQueueConfig dfConfig = queueConfigService.getDfConfig();
+        MqQueueConfig dfConfig = queueConfigService.getDfConfig(mqSendLog.getType());
         if(dfConfig != null){
             return dfConfig.getId();
         }

src/main/java/com/fdkankan/mqcontroller/task/TaskController.java

@@ -13,18 +13,25 @@ public class TaskController {
 
     @Autowired
     TaskService taskService;
-
-
     @Scheduled(cron = "*/10 * * * * ?") // 每10秒执行一次（使用cron表达式）
+    public void run(){
+        sendMq();
+        openEcs();
+        checkDelEcs();
+        delEcs();
+    }
+
+   // @Scheduled(cron = "*/10 * * * * ?") // 每10秒执行一次（使用cron表达式）
     public void sendMq(){
         try {
             taskService.sendMq();
         }catch (Exception e){
             log.info("runTask：",e);
         }
+
     }
 
-    @Scheduled(cron = "*/30 * * * * ?") // 每30秒执行一次（使用cron表达式）
+    //@Scheduled(cron = "*/30 * * * * ?") // 每30秒执行一次（使用cron表达式）
     public void openEcs(){
         try {
             taskService.openEcsList();
@@ -33,7 +40,7 @@ public class TaskController {
         }
     }
 
-    @Scheduled(cron = "*/20 * * * * ?") // 每20秒执行一次（使用cron表达式）
+   // @Scheduled(cron = "*/20 * * * * ?") // 每20秒执行一次（使用cron表达式）
     public void checkDelEcs() {
         try {
             taskService.checkDelEcs();
@@ -41,7 +48,7 @@ public class TaskController {
             log.info("checkDelEcs：", e);
         }
     }
-    @Scheduled(cron = "*/5 * * * * ?") // 每5秒执行一次（使用cron表达式）
+    //@Scheduled(cron = "*/5 * * * * ?") // 每5秒执行一次（使用cron表达式）
     public void delEcs(){
         try {
             taskService.delEcsList();

src/main/java/com/fdkankan/mqcontroller/task/TaskService.java

@@ -69,8 +69,22 @@ public class TaskService {
             configLogsMap.clear();
             return;
         }
-        log.info("未分配的mq队列数：{}",mqSendLogs.size());
 
+        List<MqSendLog> fdkkList = mqSendLogs.stream().filter(e -> "4dkk".equals(e.getType())).collect(Collectors.toList());
+        List<MqSendLog> panoList = mqSendLogs.stream().filter(e -> "pano".equals(e.getType())).collect(Collectors.toList());
+        List<MqQueueConfig> queueConfigList = queueConfigService.list();
+        rabbitMqService.getMqMsgMap(queueConfigList);
+
+        if(!fdkkList.isEmpty()){
+            sendMqByType(fdkkList,"4dkk");
+        }
+        if(!panoList.isEmpty()){
+            sendMqByType(panoList,"pano");
+        }
+        log.info("未分配的mq队列数4dkk：{},pano:{}",fdkkList.size(),panoList.size());
+    }
+
+    private void sendMqByType(List<MqSendLog> mqSendLogs,String type){
         //设置分配队列
         HashSet<String> contentList  = new HashSet<>();
         for (MqSendLog mqSendLog : mqSendLogs) {
@@ -90,8 +104,8 @@ public class TaskService {
             mqSendLog.setConfigId(configId);
             contentList.add(mqSendLog.getContent());
         }
-        List<MqQueueConfig> queueConfigList = queueConfigService.list();
-        rabbitMqService.getMqMsgMap(queueConfigList);
+
+        List<MqQueueConfig> queueConfigList = queueConfigService.getByType(type);
 
         for (MqQueueConfig config : queueConfigList) {
             List<MqSendLog> msgList = mqSendLogs.stream().filter(e -> e.getConfigId().equals(config.getId())).collect(Collectors.toList());
@@ -268,7 +282,7 @@ public class TaskService {
                 }
                 Boolean flag = createEcs(take.getMqScalingConfig());
                 if(flag){
-                    mqEcsService.add(take.getQueueName());
+                    mqEcsService.add(take.getQueueName(),take.getMqScalingConfig().getType());
                     sendRabbitMq(msgList,1,mqQueueConfig.getQueueName());
                 }else {
                     openList.offer(take);
@@ -311,8 +325,9 @@ public class TaskService {
                 if(mqMsg == null){
                     continue;
                 }
+                log.info("mq消息信息：{},{}",mqEcs.getQueueName(),mqMsg);
                 Integer notModelingCount = mqMsg.getConsumers() - mqMsg.getMessages_unacknowledged() - mqMsg.getMessages_ready();
-                if(mqSize - (notModelingCount + mqQueueConfig.getScalingThreshold()) >0 ){
+                if( mqSize - (notModelingCount + mqQueueConfig.getScalingThreshold()) >0 ){
                     log.info("待计算数量:{}大于空闲服务数量:{},阈值{}，不关闭",mqSize,notModelingCount,mqQueueConfig.getScalingThreshold());
                     continue;
                 }