|
|
@@ -10,7 +10,6 @@ import cn.iocoder.yudao.module.pms.dal.mysql.IotDeviceMapper;
|
|
|
import cn.iocoder.yudao.module.pms.dal.mysql.TDDeviceMapper;
|
|
|
import cn.iocoder.yudao.module.pms.service.yanfan.YfDeviceService;
|
|
|
import com.alibaba.fastjson.JSON;
|
|
|
-import com.alibaba.fastjson.JSONObject;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.apache.commons.lang3.StringUtils;
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
@@ -21,6 +20,7 @@ import javax.annotation.Resource;
|
|
|
import java.sql.Timestamp;
|
|
|
import java.text.SimpleDateFormat;
|
|
|
import java.util.*;
|
|
|
+import java.util.function.Function;
|
|
|
import java.util.stream.Collectors;
|
|
|
|
|
|
@Component
|
|
|
@@ -38,99 +38,95 @@ public class TdCronJob implements JobHandler {
|
|
|
@Override
|
|
|
@TenantIgnore
|
|
|
public String execute(String param) throws Exception {
|
|
|
-
|
|
|
- //1.查询源设备,构建sn -> YfDeviceDO map,避免循环内stream findFirst
|
|
|
List<YfDeviceDO> allDevice = yfDeviceService.getAllDevice();
|
|
|
- Map<String, YfDeviceDO> yfDeviceMap = allDevice.stream()
|
|
|
- .collect(Collectors.toMap(YfDeviceDO::getSerialNumber, e -> e));
|
|
|
- List<String> rawSnList = new ArrayList<>(yfDeviceMap.keySet());
|
|
|
- if(CollUtil.isEmpty(rawSnList)){
|
|
|
- return "";
|
|
|
- }
|
|
|
|
|
|
- //2.查询iot设备
|
|
|
- List<IotDeviceDO> devices = iotDeviceMapper.selectByCodeIn(rawSnList);
|
|
|
- if (CollUtil.isEmpty(devices)) {
|
|
|
- return "";
|
|
|
- }
|
|
|
-
|
|
|
- // ===================== TDengine批量预处理(替代循环内tableIfExist、selectLastTime) =====================
|
|
|
- // 子表完整表名:device_${sn}
|
|
|
- List<String> fullTbNameList = rawSnList.stream()
|
|
|
- .map(sn -> "device_" + sn)
|
|
|
+ // ① serialNumber(小写) -> YfDeviceDO,替代循环内 stream filter
|
|
|
+ Map<String, YfDeviceDO> yfDeviceMap = allDevice.stream()
|
|
|
+ .collect(Collectors.toMap(
|
|
|
+ d -> d.getSerialNumber().toLowerCase(),
|
|
|
+ Function.identity(),
|
|
|
+ (a, b) -> a // 重复时保留前者
|
|
|
+ ));
|
|
|
+
|
|
|
+ List<String> codes = new ArrayList<>(yfDeviceMap.keySet());
|
|
|
+ List<IotDeviceDO> devices = iotDeviceMapper.selectByCodeIn(codes);
|
|
|
+
|
|
|
+ // ② 预取所有存在的 device 表名(一次查询,放入 Set)
|
|
|
+ Set<String> existTables = deviceMapper.selectAllDeviceTables().stream()
|
|
|
+ .map(String::toLowerCase)
|
|
|
+ .collect(Collectors.toSet());
|
|
|
+
|
|
|
+ // ③ 筛选出表存在的设备,组装子表名列表
|
|
|
+ List<String> targetTableNames = allDevice.stream()
|
|
|
+ .map(d -> "device_" + d.getSerialNumber().toLowerCase())
|
|
|
+ .filter(existTables::contains)
|
|
|
.collect(Collectors.toList());
|
|
|
- for (String s : fullTbNameList) {
|
|
|
- System.out.println("--------------"+s);
|
|
|
- }
|
|
|
- //批量查询哪些子表真实存在
|
|
|
- List<String> existFullTbList = deviceMapper.batchGetExistFullTbName(fullTbNameList);
|
|
|
- existFullTbList.forEach(device -> {
|
|
|
- System.out.println("%%%%%%%%%%%%%%%%"+device);
|
|
|
- });
|
|
|
- // key:完整表名 device_xxx ; value:原始sn
|
|
|
- Map<String, String> fullTbToRawSnMap = new HashMap<>();
|
|
|
- for (String sn : rawSnList) {
|
|
|
- fullTbToRawSnMap.put("device_" + sn, sn);
|
|
|
- }
|
|
|
|
|
|
- //批量查询【带identity过滤条件】每个设备最大ts
|
|
|
- Map<String, Timestamp> rawSnLastTsMap = new HashMap<>();
|
|
|
- if (CollUtil.isNotEmpty(existFullTbList)) {
|
|
|
- List<SnTsVO> snTsVOList = deviceMapper.batchQueryFilterLastTs(existFullTbList);
|
|
|
- System.out.println("=========================="+JSON.toJSONString(snTsVOList));
|
|
|
- for (SnTsVO vo : snTsVOList) {
|
|
|
- String rawSn = fullTbToRawSnMap.get(vo.getTbName());
|
|
|
- if(rawSn != null && vo.getTs() != null){
|
|
|
- rawSnLastTsMap.put(rawSn, vo.getTs());
|
|
|
- }
|
|
|
+ // ④ 超级表一次查询全部设备的最新 ts → Map<serialNumber小写, Timestamp>
|
|
|
+ Map<String, Timestamp> lastTimeMap = new HashMap<>();
|
|
|
+ if (CollUtil.isNotEmpty(targetTableNames)) {
|
|
|
+ List<DeviceLastTimeVO> lastTimeList = deviceMapper.selectLastTimeBatch(targetTableNames);
|
|
|
+ for (DeviceLastTimeVO vo : lastTimeList) {
|
|
|
+ // tbname 格式为 device_xxx,去掉前缀得到 serialNumber
|
|
|
+ String serialNumber = vo.getTbname().toLowerCase().replace("device_", "");
|
|
|
+ lastTimeMap.put(serialNumber, vo.getTs());
|
|
|
}
|
|
|
}
|
|
|
|
|
|
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
|
|
-
|
|
|
- // ===================== Redis批量获取hash字段,减少网络IO =====================
|
|
|
- Set<String> deviceCodeSet = devices.stream().map(IotDeviceDO::getDeviceCode).collect(Collectors.toSet());
|
|
|
- Map<String, Map<Object, Object>> redisHashCache = new HashMap<>();
|
|
|
- for (String code : deviceCodeSet) {
|
|
|
- String hashKey = "TSLV:" + code;
|
|
|
- List<Object> multiGet = redisTemplate.opsForHash().multiGet(hashKey, Arrays.asList("lat", "lng", "online"));
|
|
|
- Map<Object, Object> fieldMap = new HashMap<>();
|
|
|
- fieldMap.put("lat", multiGet.get(0));
|
|
|
- fieldMap.put("lng", multiGet.get(1));
|
|
|
- fieldMap.put("online", multiGet.get(2));
|
|
|
- redisHashCache.put(code, fieldMap);
|
|
|
- }
|
|
|
-
|
|
|
- // ===================== 业务组装,循环内部不再任何DB调用 =====================
|
|
|
List<IotDeviceDO> deviceDOS = new ArrayList<>();
|
|
|
+
|
|
|
+ // ⑤ 循环内纯内存操作,零数据库查询
|
|
|
for (IotDeviceDO device : devices) {
|
|
|
String deviceCode = device.getDeviceCode();
|
|
|
- YfDeviceDO yfDevice = yfDeviceMap.get(deviceCode);
|
|
|
- if (yfDevice == null) {
|
|
|
+ YfDeviceDO e = yfDeviceMap.get(deviceCode.toLowerCase());
|
|
|
+ if (e == null) {
|
|
|
continue;
|
|
|
}
|
|
|
- device.setYfDeviceId(yfDevice.getDeviceId());
|
|
|
- String sn = yfDevice.getSerialNumber();
|
|
|
|
|
|
- //从预加载map拿ts,完全替代循环调用selectLastTime
|
|
|
- Timestamp ts = rawSnLastTsMap.get(sn);
|
|
|
+ device.setYfDeviceId(e.getDeviceId());
|
|
|
+
|
|
|
+ // 从 Map 取最新时间,替代循环内 selectLastTime
|
|
|
+ Timestamp ts = lastTimeMap.get(e.getSerialNumber().toLowerCase());
|
|
|
if (ts != null) {
|
|
|
device.setLastInlineTime(sdf.format(ts));
|
|
|
- } else {
|
|
|
- device.setLastInlineTime(null);
|
|
|
}
|
|
|
|
|
|
- //填充在线状态逻辑
|
|
|
- fillInlineStatus(device, yfDevice);
|
|
|
- //填充经纬度、redis在线标记
|
|
|
- fillLocationAndOnline(device, redisHashCache.get(deviceCode));
|
|
|
+ if (e.getStatus() == 1) { // 未激活状态
|
|
|
+ if (StringUtils.isNotBlank(device.getLastInlineTime())) {
|
|
|
+ device.setIfInline(DateUtils.checkIfFullDayDifference(device.getLastInlineTime()) ? 4 : 3);
|
|
|
+ } else {
|
|
|
+ device.setIfInline(4);
|
|
|
+ }
|
|
|
+
|
|
|
+ // Redis 一次 hgetAll 替代三次 hget
|
|
|
+ Map<Object, Object> tslv = redisTemplate.opsForHash().entries("TSLV:" + device.getDeviceCode());
|
|
|
+ Object lat = tslv.get("lat");
|
|
|
+ Object lng = tslv.get("lng");
|
|
|
+ Object online = tslv.get("online");
|
|
|
+
|
|
|
+ if (Objects.nonNull(lat)) {
|
|
|
+ device.setLat(Double.valueOf((String) JSON.parseObject(lat.toString()).get("value")));
|
|
|
+ }
|
|
|
+ if (Objects.nonNull(lng)) {
|
|
|
+ device.setLng(Double.valueOf((String) JSON.parseObject(lng.toString()).get("value")));
|
|
|
+ }
|
|
|
+ if (Objects.nonNull(online)) {
|
|
|
+ String value = String.valueOf(JSON.parseObject(online.toString()).get("value"));
|
|
|
+ device.setIfInline("true".equals(value) ? 3 : 4);
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ if (StringUtils.isNotBlank(device.getLastInlineTime())) {
|
|
|
+ device.setIfInline(DateUtils.checkIfFullDayDifference(device.getLastInlineTime()) ? 4 : e.getStatus());
|
|
|
+ } else {
|
|
|
+ device.setIfInline(4);
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
deviceDOS.add(device);
|
|
|
}
|
|
|
|
|
|
- if (CollUtil.isNotEmpty(deviceDOS)) {
|
|
|
- iotDeviceMapper.updateBatch(deviceDOS);
|
|
|
- }
|
|
|
+ iotDeviceMapper.updateBatch(deviceDOS);
|
|
|
return "";
|
|
|
// List<String> codes = new ArrayList<>();
|
|
|
// List<YfDeviceDO> allDevice = yfDeviceService.getAllDevice();
|
|
|
@@ -209,62 +205,4 @@ public class TdCronJob implements JobHandler {
|
|
|
String abc = "2026-01-21 10:25:40";
|
|
|
System.out.println(DateUtils.checkIfFullDayDifference(abc));
|
|
|
}
|
|
|
-
|
|
|
-
|
|
|
- /**
|
|
|
- * 填充在线状态逻辑,原业务逻辑完全保留
|
|
|
- */
|
|
|
- private void fillInlineStatus(IotDeviceDO device, YfDeviceDO yfDevice) throws Exception {
|
|
|
- String lastInlineTime = device.getLastInlineTime();
|
|
|
- Integer yfStatus = yfDevice.getStatus();
|
|
|
- if (yfStatus == 1) {
|
|
|
- //未激活状态
|
|
|
- if (StringUtils.isNotBlank(lastInlineTime)) {
|
|
|
- if (DateUtils.checkIfFullDayDifference(lastInlineTime)) {
|
|
|
- device.setIfInline(4);
|
|
|
- } else {
|
|
|
- device.setIfInline(3);
|
|
|
- }
|
|
|
- } else {
|
|
|
- device.setIfInline(4);
|
|
|
- }
|
|
|
- } else {
|
|
|
- if (StringUtils.isNotBlank(lastInlineTime)) {
|
|
|
- if (DateUtils.checkIfFullDayDifference(lastInlineTime)) {
|
|
|
- device.setIfInline(4);
|
|
|
- } else {
|
|
|
- device.setIfInline(yfStatus);
|
|
|
- }
|
|
|
- } else {
|
|
|
- device.setIfInline(4);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 填充经纬度、redis的online状态
|
|
|
- */
|
|
|
- private void fillLocationAndOnline(IotDeviceDO device, Map<Object, Object> redisFields) {
|
|
|
- if (redisFields == null) {
|
|
|
- return;
|
|
|
- }
|
|
|
- Object latObj = redisFields.get("lat");
|
|
|
- if (Objects.nonNull(latObj)) {
|
|
|
- JSONObject jsonObject = JSON.parseObject(latObj.toString());
|
|
|
- device.setLat(Double.valueOf(String.valueOf(jsonObject.get("value"))));
|
|
|
- }
|
|
|
-
|
|
|
- Object lngObj = redisFields.get("lng");
|
|
|
- if (Objects.nonNull(lngObj)) {
|
|
|
- JSONObject jsonObject = JSON.parseObject(lngObj.toString());
|
|
|
- device.setLng(Double.valueOf(String.valueOf(jsonObject.get("value"))));
|
|
|
- }
|
|
|
-
|
|
|
- Object onlineObj = redisFields.get("online");
|
|
|
- if (Objects.nonNull(onlineObj)) {
|
|
|
- JSONObject jsonObject = JSON.parseObject(onlineObj.toString());
|
|
|
- String value = String.valueOf(jsonObject.get("value"));
|
|
|
- device.setIfInline("true".equals(value) ? 3 : 4);
|
|
|
- }
|
|
|
- }
|
|
|
}
|