设备升级任务

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

@@ -17,6 +17,8 @@ public class DeviceThreadPoolConfig {
 
     public static final String ALI_DEVICE_ACTIVE = "aliDeviceActive";
 
+    public static final String DEVICE_STATUS_UP = "deviceStatusUp";
+
     /**
      * 公用线程池
      */
@@ -80,4 +82,20 @@ public class DeviceThreadPoolConfig {
                 .queueSize(coreSize)
                 .builder();
     }
+
+    /**
+     * 设备状态上报线程池
+     */
+    @DynamicTp
+    @Bean(DEVICE_STATUS_UP)
+    public ThreadPoolTaskExecutor deviceStatusUp() {
+        int coreSize = 8;
+        return ThreadPoolUtils.newPoll()
+                .name(DEVICE_STATUS_UP)
+                .coreSize(coreSize)
+                .maxSize(coreSize * 2)
+                .keepAlive(300)
+                .queueSize(coreSize)
+                .builder();
+    }
 }

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

@@ -1,5 +1,6 @@
 package com.xy.service;
 
+import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
@@ -7,14 +8,9 @@ import com.xy.config.DeviceThreadPoolConfig;
 import com.xy.dbmapper.SyncUpdate;
 import com.xy.device.EnumDevcieDoorStatus;
 import com.xy.device.EnumDeviceTempConfig;
-import com.xy.dto.ActivityInfoDto;
-import com.xy.dto.DeviceEventMsgDto;
-import com.xy.dto.DeviceStatusDto;
-import com.xy.dto.DeviceTempRecordsDto;
-import com.xy.entity.DeviceConfig;
-import com.xy.entity.DeviceStatus;
-import com.xy.entity.DeviceSysinfo;
-import com.xy.entity.SysDictRedis;
+import com.xy.device.EnumMqttCmdTempletTask;
+import com.xy.dto.*;
+import com.xy.entity.*;
 import com.xy.mapper.DeviceStatusMapper;
 import com.xy.mapper.entity.DeviceStatusCount;
 import com.xy.utils.*;
@@ -23,12 +19,14 @@ import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 
 import java.math.BigDecimal;
 import java.time.LocalDateTime;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
@@ -42,9 +40,9 @@ import static com.xy.utils.Beans.copy;
  * @author lijin
  * @since 2022-12-23
  */
-@Service
-@AllArgsConstructor
 @Slf4j
+@Service
+@AllArgsConstructor(onConstructor_ = @Lazy)
 @Api(tags = "设备-状态")
 public class DeviceStatusServiceImpl extends ServiceImpl<DeviceStatusMapper, DeviceStatus> implements DeviceStatusService {
 
@@ -60,91 +58,124 @@ public class DeviceStatusServiceImpl extends ServiceImpl<DeviceStatusMapper, Dev
 
     private DeviceSysinfoServiceImpl deviceSysinfoService;
 
+    private DeviceVersionUpServiceImpl deviceVersionUpService;
+
+    private RedisService<DeviceVersionUpServiceImpl.UpInfo> upInfoRedisService;
+
+    private MqttServiceImpl mqttService;
+
     @Override
     @ApiOperation("上报信息")
     public R up(DeviceStatusDto.Up up) {
         log.info("设备信息上报：{}", JSONUtil.toJsonStr(up));
-        DeviceStatus deviceStatusInfo = getById(up.getDeviceId());
-        if (deviceStatusInfo == null) {
+        DeviceStatus deviceStatus = getById(up.getDeviceId());
+        if (deviceStatus == null) {
             return R.ok();
         }
-        DeviceStatus deviceStatus = copy(DeviceStatus.class, up).setUpdateTime(LocalDateTime.now());
-        updateById(deviceStatus);
-        Integer tempState = up.getTempState();
-        //故障
-        boolean isRisk = false;
-        //温控仪异常
-        if (tempState != null && tempState.intValue() != 0) {
-            isRisk = true;
-        }
-        //温度上报
-        if (Emptys.check(deviceStatus.getTempValue())) {
-            DeviceSysinfo deviceSysinfo = deviceSysinfoService.getById(up.getDeviceId());
-            //添加设备温度日志
-            if (deviceSysinfo.getIsHaveTemp()) {
-                DeviceTempRecordsDto.Save save = new DeviceTempRecordsDto.Save()
-                        .setDeviceId(up.getDeviceId());
-                save.setTempValue(isRisk ? null : deviceStatus.getTempValue());
-                deviceTempRecordsService.save(save);
-                //查询设备配置
-                DeviceConfig deviceConfig = deviceConfigService.getById(deviceStatus.getDeviceId());
-                int tempMax, tempMin;
-                if (deviceConfig == null) {
-                    Map<String, SysDictRedis> stringSysDictRedisMap = SysDictUtils.get(EnumDeviceTempConfig.Code.CODE.getCode());
-                    tempMax = Integer.valueOf(stringSysDictRedisMap.get(EnumDeviceTempConfig.MAX.getCode()).getValue());
-                    tempMin = Integer.valueOf(stringSysDictRedisMap.get(EnumDeviceTempConfig.MIN.getCode()).getValue());
-                } else {
-                    tempMax = deviceConfig.getTempMax();
-                    tempMin = deviceConfig.getTempMin();
-                }
-                //温度有异常
-                String key = String.format("device_temp_error:%s:%s", deviceStatus.getDeviceId(), deviceStatus.getTempValue() >= tempMax ? "max" : "min");
-                if (deviceStatus.getTempValue() > tempMax || deviceStatus.getTempValue() < tempMin) {
-                    //校验异常次数
-                    Integer errorSize = SysDictUtils.getValue(EnumDeviceTempConfig.Code.CODE.getCode(), EnumDeviceTempConfig.ERROR_SIZE.getCode(), Integer.class);
-                    boolean fal = false;
-                    Integer maxSize = redisService.get(key);
-                    if (maxSize == null) {
-                        redisService.set(key, 1, 60 * 60);
+        ThreadPoolUtils.excPoll(DeviceThreadPoolConfig.DEVICE_STATUS_UP, 4)
+                .execute(() -> {
+                    //修改状态信息
+                    updateById(copy(DeviceStatus.class, up).setUpdateTime(LocalDateTime.now()));
+                })
+                .execute(() -> {
+                    //上报关门
+                    if (!Emptys.check(deviceStatus.getDoorStateL())) {
+                        return;
+                    }
+                    Integer value = SysDictUtils.getValue(EnumDevcieDoorStatus.Code.CODE.getCode(), EnumDevcieDoorStatus.LOCKED.getCode(), Integer.class);
+                    if (!deviceStatus.getDoorStateL().equals(value)) {
+                        return;
+                    }
+                    activityInfoService.abort(new ActivityInfoDto.Abort()
+                            .setDeviceId(deviceStatus.getDeviceId())
+                            .setWorkType(2)
+                            .setTimeout(60 * 1)
+                    );
+                })
+                .execute(() -> {
+                    //温度上报
+                    if (!Emptys.check(up.getTempValue())) {
+                        return;
+                    }
+                    DeviceSysinfo deviceSysinfo = deviceSysinfoService.getById(deviceStatus.getDeviceId());
+                    //添加设备温度日志
+                    if (!deviceSysinfo.getIsHaveTemp()) {
+                        return;
+                    }
+                    DeviceTempRecordsDto.Save save = new DeviceTempRecordsDto.Save()
+                            .setDeviceId(deviceStatus.getDeviceId());
+                    save.setTempValue(up.getTempValue());
+                    deviceTempRecordsService.save(save);
+                    //查询设备配置
+                    DeviceConfig deviceConfig = deviceConfigService.getById(deviceStatus.getDeviceId());
+                    int tempMax, tempMin;
+                    if (deviceConfig == null) {
+                        Map<String, SysDictRedis> stringSysDictRedisMap = SysDictUtils.get(EnumDeviceTempConfig.Code.CODE.getCode());
+                        tempMax = Integer.valueOf(stringSysDictRedisMap.get(EnumDeviceTempConfig.MAX.getCode()).getValue());
+                        tempMin = Integer.valueOf(stringSysDictRedisMap.get(EnumDeviceTempConfig.MIN.getCode()).getValue());
                     } else {
+                        tempMax = deviceConfig.getTempMax();
+                        tempMin = deviceConfig.getTempMin();
+                    }
+                    //温度有异常
+                    String key = String.format("device_temp_error:%d", deviceStatus.getDeviceId());
+                    if (up.getTempValue() > tempMax || up.getTempValue() < tempMin) {
+                        //校验异常次数
+                        Integer errorSize = SysDictUtils.getValue(EnumDeviceTempConfig.Code.CODE.getCode(), EnumDeviceTempConfig.ERROR_SIZE.getCode(), Integer.class);
+                        Integer maxSize = redisService.get(key);
+                        maxSize = maxSize == null ? 1 : maxSize;
                         if (maxSize >= errorSize) {
-                            fal = true;
+                            //添加事件
+                            String msg = "温度异常-温度阈值%d - %d,当前温度%d";
+                            DeviceEventMsgDto.Save deviceEventMsg = new DeviceEventMsgDto.Save()
+                                    .setDeviceId(deviceStatus.getDeviceId());
+                            deviceEventMsg.setCode(DeviceErrorRecordTypesEnum.T.getCode());
+                            deviceEventMsg.setMsg(String.format(msg, tempMin, tempMax, up.getTempValue()));
+                            deviceEventMsgService.save(deviceEventMsg);
                             redisService.remove(key);
-                        } else {
-                            redisService.set(key, maxSize + 1, 60 * 60);
-                        }
-                    }
-                    //添加事件
-                    if (fal) {
-                        DeviceEventMsgDto.Save deviceEventMsg = new DeviceEventMsgDto.Save()
-                                .setDeviceId(deviceStatus.getDeviceId());
-                        deviceEventMsg.setCode(DeviceErrorRecordTypesEnum.T.getCode());
-                        String msg = "温度异常-温度阈值%d,当前温度%d";
-                        if (deviceStatus.getTempValue() >= tempMax) {
-                            deviceEventMsg.setMsg(String.format(msg, tempMax, deviceStatus.getTempValue()));
-                        }
-                        if (deviceStatus.getTempValue() <= tempMin) {
-                            deviceEventMsg.setMsg(String.format(msg, tempMin, deviceStatus.getTempValue()));
+                            return;
                         }
-                        deviceEventMsgService.save(deviceEventMsg);
+                        redisService.set(key, maxSize + 1, 60 * 60);
+                        return;
                     }
-                } else {
                     redisService.remove(key);
-                }
-            }
-        }
-        //上报关门
-        if (Emptys.check(deviceStatus.getDoorStateL())) {
-            Integer value = SysDictUtils.getValue(EnumDevcieDoorStatus.Code.CODE.getCode(), EnumDevcieDoorStatus.LOCKED.getCode(), Integer.class);
-            if (deviceStatus.getDoorStateL().equals(value)) {
-                ThreadPoolUtils.excPoll(DeviceThreadPoolConfig.DEVICE_COMMON_POLL, 1)
-                        .execute(() -> activityInfoService.abort(new ActivityInfoDto.Abort()
-                                .setDeviceId(up.getDeviceId())
-                                .setWorkType(2)
-                                .setTimeout(60 * 1)
-                        ));
-            }
-        }
+                })
+                .execute(() -> {
+                    //检查是否可升级
+                    if (!Emptys.check(up.getIsUpVersion()) || !up.getIsUpVersion()) {
+                        return;
+                    }
+                    DeviceVersionUpServiceImpl.UpInfo upInfo = upInfoRedisService.get(DeviceVersionUpServiceImpl.upInfoKey);
+                    if (upInfo == null || !upInfo.getEnable()) {
+                        return;
+                    }
+                    LambdaQueryWrapper<DeviceVersionUp> lambdaQueryWrapper = new LambdaQueryWrapper<DeviceVersionUp>()
+                            .eq(DeviceVersionUp::getDeviceId, deviceStatus.getDeviceId())
+                            .ge(DeviceVersionUp::getDeviceVersion, upInfo.getVersion())
+                            .eq(DeviceVersionUp::getStatus, 4)
+                            .eq(DeviceVersionUp::getProgress, 100);
+                    long count = deviceVersionUpService.count(lambdaQueryWrapper);
+                    if (count > 0) {
+                        return;
+                    }
+                    //下发升级指令
+                    String templetStr = SysDictUtils.getValue(EnumMqttCmdTempletTask.Code.CODE.getCode(), EnumMqttCmdTempletTask.APPUPDATE.getCode(), String.class);
+                    String appId = LambdaUtils.getProperty(DeviceVersionUpServiceImpl.UpInfo::getAppId);
+                    String appKey = LambdaUtils.getProperty(DeviceVersionUpServiceImpl.UpInfo::getAppKey);
+                    String version = LambdaUtils.getProperty(DeviceVersionUpServiceImpl.UpInfo::getVersion);
+                    String src = LambdaUtils.getProperty(DeviceVersionUpServiceImpl.UpInfo::getSrc);
+                    JSONObject templet = JSONUtil.parseObj(templetStr);
+                    JSONObject data = templet.getJSONObject("data")
+                            .set(appId, upInfo.getAppId())
+                            .set(appKey, upInfo.getAppKey())
+                            .set(version, upInfo.getVersion())
+                            .set(src, upInfo.getSrc());
+                    templet.set("data", data);
+                    CommandMqtt commandMqtt = new CommandMqtt()
+                            .setDeviceId(deviceStatus.getDeviceId())
+                            .setTemplet(templet);
+                    mqttService.senCommand(Arrays.asList(commandMqtt));
+                });
         return R.ok();
     }
 

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

@@ -9,13 +9,17 @@ import com.xy.entity.DeviceVersionUp;
 import com.xy.mapper.DeviceVersionUpMapper;
 import com.xy.utils.*;
 import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiModelProperty;
 import io.swagger.annotations.ApiOperation;
-import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import lombok.experimental.Accessors;
 import org.springframework.stereotype.Service;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 
+import javax.validation.constraints.NotBlank;
 import java.time.LocalDateTime;
 
 import static com.xy.utils.PlusBeans.toIPage;
@@ -31,10 +35,14 @@ import static com.xy.utils.PlusBeans.toPageBean;
  * @since 2023-06-09
  */
 @Service
-@AllArgsConstructor
+@RequiredArgsConstructor
 @Api(tags = "设备版本升级记录表")
 public class DeviceVersionUpServiceImpl extends ServiceImpl<DeviceVersionUpMapper, DeviceVersionUp> implements DeviceVersionUpService {
 
+    private final RedisService<UpInfo> redisService;
+
+    public static String upInfoKey = "device:version:up";
+
     @PostMapping("page")
     @ApiOperation("分页查询")
     public R<PageBean<DeviceVersionUpDto.Vo>> page(@RequestBody DeviceVersionUpDto.Page page) {
@@ -70,4 +78,49 @@ public class DeviceVersionUpServiceImpl extends ServiceImpl<DeviceVersionUpMappe
         saveOrUpdate(deviceVersionUp);
         return R.ok();
     }
+
+    @PostMapping("saveUpdateUpInfoTask")
+    @ApiOperation("添加修改版本升级任务")
+    public R saveUpdateUpInfoTask(@RequestBody @Validated UpInfo upInfo) {
+        redisService.set(upInfoKey, upInfo);
+        return R.ok();
+    }
+
+    @PostMapping("queryUpInfoTask")
+    @ApiOperation("查询版本升级任务")
+    public R<UpInfo> queryUpInfoTask() {
+        UpInfo upInfo = redisService.get(upInfoKey);
+        return R.ok(upInfo);
+    }
+
+    @PostMapping("delUpInfoTask")
+    @ApiOperation("清除版本升级任务")
+    public R delUpInfoTask() {
+        redisService.remove(upInfoKey);
+        return R.ok();
+    }
+
+    @Data
+    @Accessors(chain = true)
+    public static class UpInfo {
+
+        @NotBlank(message = "version不能为空")
+        @ApiModelProperty(value = "版本号", required = true)
+        private String version;
+
+        @NotBlank(message = "appId不能为空")
+        @ApiModelProperty(value = "appId", required = true)
+        private String appId;
+
+        @NotBlank(message = "appKey不能为空")
+        @ApiModelProperty(value = "appKey", required = true)
+        private String appKey;
+
+        @NotBlank(message = "src不能为空")
+        @ApiModelProperty(value = "升级地址", required = true)
+        private String src;
+
+        @ApiModelProperty(value = "是否开启 默认=true")
+        private Boolean enable = true;
+    }
 }

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

@@ -180,6 +180,9 @@ public class DeviceStatusDto {
          */
         @ApiModelProperty("缺货状态")
         private Integer stockStatus;
+
+        @ApiModelProperty("版本询问")
+        private Boolean isUpVersion;
     }
 
 }