Merge remote-tracking branch 'origin/master'

device-api-service/src/main/java/com/xy/config/DeviceThreadPoolConfig.java

@@ -17,11 +17,13 @@ public class DeviceThreadPoolConfig {
     @DynamicTp
     @Bean(DEVICE_COMMON_POLL)
     public ThreadPoolTaskExecutor deviceCommonPoll() {
+        int coreSize = 15;
         return ThreadPoolUtils.newPoll()
                 .name(DEVICE_COMMON_POLL)
-                .coreSize(10)
-                .maxSize(200)
+                .coreSize(coreSize)
+                .maxSize(coreSize * 2)
                 .keepAlive(60)
+                .queueSize(coreSize * 10)
                 .builder();
     }
 

device-api-service/src/main/java/com/xy/entity/MqttCommand.java

@@ -16,6 +16,11 @@ public class MqttCommand {
     @TableId
     private Long sn;
 
+    /**
+     * 业务sn
+     */
+    private Long wkSn;
+
     /**
      * 是否调试
      */

device-api-service/src/main/java/com/xy/service/DeviceInfoServiceImpl.java

@@ -83,23 +83,41 @@ public class DeviceInfoServiceImpl extends ServiceImpl<DeviceInfoMapper, DeviceI
             return R.ok();
         }
         DeviceInfoDto.Vo deviceInfo = copy(DeviceInfoDto.Vo.class, list.get(0));
-        ThreadPoolUtils.excPoll(DeviceThreadPoolConfig.DEVICE_COMMON_POLL, 3)
-                .execute(() -> {
+        int num = 0;
+        if (obj.getIsSysinfo()) {
+            num++;
+        }
+        if (obj.getIsStatus()) {
+            num++;
+        }
+        if (obj.getIsRegister()) {
+            num++;
+        }
+        if (num > 0) {
+            ThreadPoolUtils.Execute execute = ThreadPoolUtils.excPoll(DeviceThreadPoolConfig.DEVICE_COMMON_POLL, num);
+            if (obj.getIsSysinfo()) {
+                execute.execute(() -> {
                     //系统信息
                     DeviceSysinfoDto.Vo deviceSysinfo = deviceSysinfoService.get(new DeviceSysinfoDto.Vo().setDeviceId(deviceInfo.getDeviceId())).getData();
                     deviceInfo.setDeviceSysinfo(deviceSysinfo);
-                })
-                .execute(() -> {
+                });
+            }
+            if (obj.getIsStatus()) {
+                execute.execute(() -> {
                     //状态信息
                     DeviceStatusDto.Vo deviceStatus = deviceStatusService.obj(new DeviceStatusDto.Vo().setDeviceId(deviceInfo.getDeviceId())).getData();
                     deviceInfo.setDeviceStatus(deviceStatus);
-                })
-                .execute(() -> {
+                });
+            }
+            if (obj.getIsRegister()) {
+                execute.execute(() -> {
                     //注册信息
                     DeviceRegisterDto.Vo deviceRegister = deviceRegisterService.obj(new DeviceRegisterDto.Vo().setDeviceId(deviceInfo.getDeviceId())).getData();
                     deviceInfo.setDeviceRegister(deviceRegister);
-                })
-                .end();
+                });
+            }
+            execute.end();
+        }
         return R.ok(deviceInfo);
     }
 

device-api-service/src/main/java/com/xy/service/MqttServiceImpl.java

@@ -15,6 +15,7 @@ import com.xy.dto.CommandMqtt;
 import com.xy.dto.MqttDto;
 import com.xy.dto.PaterDto;
 import com.xy.entity.MqttCommand;
+import com.xy.error.CommRuntimeException;
 import com.xy.mapper.MqttCommandMapper;
 import com.xy.producer.CommProducer;
 import com.xy.utils.*;
@@ -50,11 +51,19 @@ public class MqttServiceImpl extends ServiceImpl<MqttCommandMapper, MqttCommand>
 
     private String keyStr = "mqtt:cmd:result:%d";
 
+    private boolean snInit;
+
     private long getSn() {
+        if (!snInit) {
+            throw new CommRuntimeException("sn号未初始化");
+        }
         return redisService.atomicIncrease(CommConsts.CMD_ATOM_SN);
     }
 
     private long getWkSn() {
+        if (!snInit) {
+            throw new CommRuntimeException("sn号未初始化");
+        }
         return redisService.atomicIncrease(CommConsts.CMD_ATOM_WK_SN);
     }
 
@@ -101,6 +110,7 @@ public class MqttServiceImpl extends ServiceImpl<MqttCommandMapper, MqttCommand>
     @PostMapping("sysnAtoms")
     @ApiOperation("同步sn号redis原子变量")
     public void sysnAtoms() {
+        //todo 集群有问题，暂不处理
         String sn = LambdaUtils.getProperty(MqttCommand::getSn);
         QueryWrapper<MqttCommand> wrapper = new QueryWrapper<MqttCommand>()
                 .select(String.format("max(%s) as %s", sn, sn));
@@ -109,7 +119,9 @@ public class MqttServiceImpl extends ServiceImpl<MqttCommandMapper, MqttCommand>
             return;
         }
         redisService.setAtomic(CommConsts.CMD_ATOM_SN, mqttCommand.getSn() + 1);
-        redisService.setAtomic(CommConsts.CMD_ATOM_WK_SN, JSONUtil.parseObj(mqttCommand.getValue()).getLong("wkSn") + 1);
+        MqttCommand byId = getById(mqttCommand.getSn());
+        redisService.setAtomic(CommConsts.CMD_ATOM_WK_SN, byId.getWkSn() + 1);
+        snInit = true;
     }
 
     @Timer
@@ -247,6 +259,7 @@ public class MqttServiceImpl extends ServiceImpl<MqttCommandMapper, MqttCommand>
         PaterDto pater = mqttDto.getPater();
         MqttCommand mqttCommand = Beans.copy(Beans.copy(MqttCommand.class, mqttDto), pater)
                 .setSn(pater.getSn())
+                .setWkSn(pater.getWkSn())
                 .setStatus(Integer.parseInt(status.getKey()))
                 .setValue(value.toString())
                 .setSendTime(pater.getTime())

device-api/src/main/java/com/xy/dto/DeviceInfoDto.java

@@ -44,12 +44,21 @@ public class DeviceInfoDto {
 
         @ApiModelProperty("门id 0=1号门 1=2号门")
         private Integer door;
+
+        @ApiModelProperty("是否需要系统信息")
+        private Boolean isSysinfo = false;
+
+        @ApiModelProperty("是否需要状态信息")
+        private Boolean isStatus = false;
+
+        @ApiModelProperty("是否需要注册信息")
+        private Boolean isRegister = false;
     }
 
     @Data
     @Accessors(chain = true)
     public static class ListDto {
-        @ApiModelProperty(value = "设备id批量查", required = false)
+        @ApiModelProperty(value = "设备id批量查")
         private List<Long> deviceIds;
     }