429 lines
22 KiB
Java
429 lines
22 KiB
Java
package com.maibu.service;
|
||
|
||
import java.io.IOException;
|
||
import java.lang.reflect.Field;
|
||
import java.math.BigDecimal;
|
||
import java.time.LocalDateTime;
|
||
import java.util.ArrayList;
|
||
import java.util.Collection;
|
||
import java.util.Comparator;
|
||
import java.util.List;
|
||
import java.util.Map;
|
||
import java.util.concurrent.ConcurrentHashMap;
|
||
import java.util.concurrent.ScheduledExecutorService;
|
||
import java.util.concurrent.TimeUnit;
|
||
import java.util.stream.Collectors;
|
||
|
||
import com.maibu.core.business.dto.DeviceMapPushDTO;
|
||
import com.maibu.core.business.dto.WebMapMessageDTO;
|
||
import com.maibu.core.enums.*;
|
||
import com.maibu.utils.json.JsonUtils;
|
||
import org.apache.commons.lang3.StringUtils;
|
||
import org.checkerframework.checker.units.qual.s;
|
||
import org.springframework.beans.factory.annotation.Autowired;
|
||
import org.springframework.beans.factory.annotation.Value;
|
||
import org.springframework.core.io.Resource;
|
||
import org.springframework.core.io.ResourceLoader;
|
||
import org.springframework.stereotype.Service;
|
||
import org.springframework.util.CollectionUtils;
|
||
|
||
import com.alibaba.fastjson2.JSON;
|
||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||
import com.maibu.constant.Constant;
|
||
import com.maibu.constant.NettyCacheKey;
|
||
import com.maibu.core.business.Device;
|
||
import com.maibu.core.business.DeviceRunStatistics;
|
||
import com.maibu.core.business.DeviceRunningStatusHistory;
|
||
import com.maibu.core.business.DeviceStatusRecordDTO;
|
||
import com.maibu.core.business.ErrorIdentificationStandard;
|
||
import com.maibu.core.business.alarm_center.AlarmMessage;
|
||
import com.maibu.core.business.device.NettyDevice;
|
||
import com.maibu.core.business.inter.WebsocketMesDispather;
|
||
import com.maibu.core.host.HeartBeatDTO;
|
||
import com.maibu.core.host.LocationMessage;
|
||
import com.maibu.core.redis.RedisCache;
|
||
import com.maibu.dto.DeviceErrorPushDTO;
|
||
import com.maibu.mapper.DeviceRunStatisticsMapper;
|
||
import com.maibu.mapper.ErrorIdentificationStandardMapper;
|
||
import com.maibu.mapper.HostLocationMapper;
|
||
import com.maibu.memory.DeviceSessionManager;
|
||
import com.maibu.memory.GlobalMemory;
|
||
import com.maibu.memory.SiteMemory;
|
||
import com.maibu.mqtt.MqttTopic;
|
||
import com.maibu.utils.CompareUtils;
|
||
|
||
import cn.hutool.core.io.IORuntimeException;
|
||
import cn.hutool.core.io.IoUtil;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
|
||
@Slf4j
|
||
@Service
|
||
public class DeviceThreadService {
|
||
|
||
@Autowired
|
||
private ResourceLoader resourceLoader;
|
||
|
||
@Autowired
|
||
private RedisCache redisCache;
|
||
|
||
@Autowired
|
||
private DeviceSessionManager sessionManager;
|
||
|
||
@Autowired
|
||
private ErrorIdentificationStandardMapper standardMapper;
|
||
|
||
@Autowired
|
||
private WebsocketMesDispather websocketMesDispather;
|
||
|
||
@Autowired
|
||
private AlertPushService alertPushService;
|
||
|
||
@Autowired
|
||
private IDeviceService deviceService;
|
||
|
||
@Value("${mqtt.client-id}")
|
||
private String mqttClientId;
|
||
|
||
@Autowired
|
||
private ScheduledExecutorService executor;
|
||
|
||
@Autowired
|
||
private HostLocationMapper hostLocationMapper;
|
||
|
||
@Autowired
|
||
private DeviceRunStatisticsMapper deviceRunStatisticsMapper;
|
||
|
||
public void deviceErrorMonitor() throws InterruptedException, IOException {
|
||
initStandard();
|
||
|
||
// TODO fix 后续是否分多个线程 一个site一个线程
|
||
// 有还需要更新接受故障方式。
|
||
executor.scheduleWithFixedDelay(() -> {
|
||
try {
|
||
String key = NettyCacheKey.deviceRunningStatusKey;
|
||
Collection<String> keys = redisCache.getListKeyByPrefix(key);
|
||
if (!CollectionUtils.isEmpty(keys)) {
|
||
List<DeviceErrorPushDTO> pushList = new ArrayList<>();
|
||
keys.forEach(k -> {
|
||
DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(k);
|
||
if (recordDTO != null && !CollectionUtils.isEmpty(recordDTO.getHistories())) {
|
||
recordDTO.getHistories()
|
||
.sort(Comparator.comparing(DeviceRunningStatusHistory::getCreateTime).reversed());
|
||
DeviceRunningStatusHistory history = recordDTO.getHistories().get(0);
|
||
|
||
String[] parts = k.split(":");
|
||
String deviceId = parts[1];
|
||
NettyDevice slaveDevice = sessionManager.getDevice(deviceId);
|
||
if (slaveDevice != null && slaveDevice.getOnlineStatus() == 1) {
|
||
Long siteId = slaveDevice.getDevice().getSiteId();
|
||
Long orgId = slaveDevice.getDevice().getOrgId();
|
||
log.debug("compare siteId:{},orgId:{}", siteId, orgId);
|
||
SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
|
||
if (siteMemory != null) {
|
||
List<AlarmMessage> compared = compareErrorStandard(siteId, orgId, history,
|
||
siteMemory.standards.get(ErrorSource.MOWER));
|
||
DeviceErrorPushDTO pushDTO = new DeviceErrorPushDTO();
|
||
pushDTO.setDeviceId(deviceId);
|
||
pushDTO.setTime(System.currentTimeMillis());
|
||
pushDTO.setDetails(compared);
|
||
pushDTO.setEvent(RespondCode.device_error_push);
|
||
pushList.add(pushDTO);
|
||
}
|
||
}
|
||
}
|
||
});
|
||
// 按实际deviceId分组，推送错误信息
|
||
Map<String, List<DeviceErrorPushDTO>> pushMap = pushList.stream()
|
||
.collect(java.util.stream.Collectors.groupingBy(DeviceErrorPushDTO::getDeviceId));
|
||
pushMap.forEach((deviceId, dtos) -> {
|
||
// pushErrorMessage(deviceId, JSON.toJSONString(dtos));
|
||
// todo 判断错误级别，是否需要推送邮件 或者钉钉。
|
||
// 找到这个设备的场站 添加场站是否开启关闭 钉钉或者邮件推送，推送邮件 添加人员是否订阅邮件，钉钉是加入钉钉群
|
||
alertPushService.messagePushHandler(deviceId, dtos);
|
||
});
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("执行监测错误线程错误：{}", e.getMessage());
|
||
}
|
||
}, 0, 5, TimeUnit.SECONDS);
|
||
}
|
||
|
||
public void initStandard() throws IOException {
|
||
GlobalMemory.getAllSiteMemory().forEach(siteMemory -> {
|
||
if (siteMemory != null && siteMemory.standards == null) {
|
||
LambdaQueryWrapper<ErrorIdentificationStandard> queryWrapper = new LambdaQueryWrapper<>();
|
||
queryWrapper.eq(ErrorIdentificationStandard::getSiteId, siteMemory.sysSite.getId());
|
||
List<ErrorIdentificationStandard> list = standardMapper.selectList(queryWrapper);
|
||
if (CollectionUtils.isEmpty(list)) {
|
||
try {
|
||
// 读取resource目录下的device-config.json文件
|
||
Resource resource = resourceLoader.getResource("classpath:errorStandard.json");
|
||
// 使用fastjson2转换为DeviceConfig对象
|
||
String jsonContent;
|
||
jsonContent = IoUtil.readUtf8(resource.getInputStream());
|
||
List<ErrorIdentificationStandard> newStandards = JSON.parseArray(jsonContent,
|
||
ErrorIdentificationStandard.class);
|
||
if (!CollectionUtils.isEmpty(newStandards)) {
|
||
newStandards.forEach(x -> {
|
||
x.setOrgId(siteMemory.sysSite.getOrgId());
|
||
x.setSiteId(siteMemory.sysSite.getId());
|
||
});
|
||
standardMapper.insertBatch(newStandards);
|
||
siteMemory.standards = newStandards.stream()
|
||
.filter(e -> e.getErrorSource() != null)
|
||
.collect(Collectors.groupingBy(
|
||
ErrorIdentificationStandard::getErrorSource,
|
||
ConcurrentHashMap::new,
|
||
Collectors.toList()));
|
||
|
||
}
|
||
} catch (IORuntimeException e) {
|
||
// TODO Auto-generated catch block
|
||
e.printStackTrace();
|
||
} catch (IOException e) {
|
||
// TODO Auto-generated catch block
|
||
e.printStackTrace();
|
||
}
|
||
} else {
|
||
siteMemory.standards = list.stream()
|
||
.filter(e -> e.getErrorSource() != null)
|
||
.collect(Collectors.groupingBy(
|
||
ErrorIdentificationStandard::getErrorSource,
|
||
ConcurrentHashMap::new,
|
||
Collectors.toList()));
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
public void pushErrorMessage(String deviceId, String dto) {
|
||
// NettyDevice device = sessionManager.getDevice(deviceId);
|
||
// if (device != null) {
|
||
// ByteBuf buf = Unpooled.buffer();
|
||
// CommandUtils.buildCommand(buf, dto, CommandConstant.interaction);
|
||
// if (device.getChannel() != null && device.getChannel().isActive()) {
|
||
// device.getChannel().writeAndFlush(buf);
|
||
// }
|
||
// }
|
||
try {
|
||
GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId,
|
||
String.format(MqttTopic.DEVICE_ERROR_PUSH_TOPIC, deviceId), dto);
|
||
// GlobalMemory.mqttClientUtil.publish(String.format(MqttTopic.DEVICE_ERROR_PUSH_TOPIC,
|
||
// deviceId), dto);
|
||
List<NettyDevice> controlMasters = sessionManager.getAllSlaveControl(deviceId);
|
||
if (!CollectionUtils.isEmpty(controlMasters)) {
|
||
controlMasters.forEach(x -> {
|
||
websocketMesDispather.dispather(x.getConnectorId(), dto);
|
||
log.info("推送错误信息 device:{},content:{}", deviceId, dto);
|
||
});
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("推送错误信息失败 device:{}error:{}", deviceId, e.getMessage());
|
||
}
|
||
}
|
||
|
||
public List<AlarmMessage> compareErrorStandard(Long siteId, Long orgId, DeviceRunningStatusHistory history,
|
||
List<ErrorIdentificationStandard> standards) {
|
||
if (history == null || CollectionUtils.isEmpty(standards))
|
||
return null;
|
||
List<AlarmMessage> list = new ArrayList<>();
|
||
standards.forEach(standard -> {
|
||
String fieldName = standard.getField();
|
||
String compareValues = standard.getCompareValues();
|
||
CompareEnum compareType = standard.getCompareType();
|
||
String targetValue = knownClassFieldFinding(fieldName, history);
|
||
log.debug("compare2 compareType:{} compareValues:{},targetValue:{}", compareType, compareValues,
|
||
targetValue);
|
||
if (!StringUtils.isEmpty(compareValues) && !StringUtils.isEmpty(targetValue)) {
|
||
boolean result = CompareUtils.compare(compareType, targetValue, compareValues);
|
||
log.debug("compare3 result:{}", result);
|
||
if (result) {
|
||
AlarmMessage alarm = new AlarmMessage();
|
||
alarm.setContent(standard.getErrorDescription());
|
||
alarm.setDeviceSn(history.getDeviceId());
|
||
alarm.setDeviceName(history.getDeviceId());
|
||
alarm.setLevel(standard.getErrorLevel());
|
||
alarm.setLocation(history.getLatitude() + "," + history.getLongitude());
|
||
alarm.setName(standard.getErrorName());
|
||
alarm.setNo(standard.getErrorCode());
|
||
alarm.setSuggestion(standard.getSuggestion());
|
||
alarm.setStatus("未处理");
|
||
alarm.setTimeValues(targetValue);
|
||
alarm.setTime(LocalDateTime.now());
|
||
alarm.setOrgId(orgId);
|
||
alarm.setSiteId(siteId);
|
||
list.add(alarm);
|
||
}
|
||
}
|
||
});
|
||
return list;
|
||
}
|
||
|
||
private static String knownClassFieldFinding(String fieldName, DeviceRunningStatusHistory history) {
|
||
try {
|
||
// 1. 获取类的Class对象
|
||
Class<?> historyClass = DeviceRunningStatusHistory.class;
|
||
Field field = historyClass.getDeclaredField(fieldName);
|
||
field.setAccessible(true);
|
||
Object value = field.get(history);
|
||
return (String) value;
|
||
} catch (NoSuchFieldException | IllegalAccessException e) {
|
||
System.out.println("获取字段失败: " + e.getMessage());
|
||
log.error("获取属性值失败:{}", e.getMessage());
|
||
}
|
||
return null;
|
||
}
|
||
|
||
/**
|
||
* 监控心跳
|
||
*/
|
||
public void heartBeatMonitor() {
|
||
executor.scheduleWithFixedDelay(() -> {
|
||
try {
|
||
Long nowTime = System.currentTimeMillis();
|
||
List<HeartBeatDTO> heartBeatDTOS = GlobalMemory.customMqttDeviceMonitor.getAllHeartBeat();
|
||
if (!CollectionUtils.isEmpty(heartBeatDTOS)) {
|
||
heartBeatDTOS.forEach(x -> {
|
||
if (x == null) {
|
||
return;
|
||
}
|
||
String deviceId = x.getSn();
|
||
Long lastActiveTime = x.getSysTimestamp();
|
||
NettyDevice nettyDevice = sessionManager.getDevice(x.getSn());
|
||
|
||
if (lastActiveTime == null) {
|
||
log.error("设备心跳时间为空，deviceId={}", deviceId);
|
||
return;
|
||
}
|
||
if (nowTime - lastActiveTime >= Constant.hostOfflineInterval) {
|
||
// 超时了判断离线
|
||
if (nettyDevice != null) {
|
||
nettyDevice.status(DeviceStatus.offline.getType());
|
||
// todo 推送设备离线
|
||
String key = NettyCacheKey.latestPositonKey + deviceId;
|
||
// 保存一下最新一次的数据到 数据库
|
||
LocationMessage locationMessage = redisCache.getCacheObject(key);
|
||
if (locationMessage != null) {
|
||
LambdaQueryWrapper<LocationMessage> queryWrapper = new LambdaQueryWrapper<>();
|
||
queryWrapper.eq(LocationMessage::getDeviceId, deviceId);
|
||
LocationMessage existing = hostLocationMapper.selectOne(queryWrapper);
|
||
if (existing != null) {
|
||
locationMessage.setId(existing.getId());
|
||
}
|
||
hostLocationMapper.saveOrUpdate(locationMessage);
|
||
}
|
||
Device device = deviceService.selectDeviceBySerialNumber(deviceId);
|
||
if (device != null) {
|
||
device.setOnlineStatus(DeviceStatus.offline.getType());
|
||
device.setUpdateTime(LocalDateTime.now());
|
||
deviceService.updateDeviceBySelf(device);
|
||
}
|
||
|
||
Long orgId = nettyDevice.getDevice().getOrgId();
|
||
Long siteId = nettyDevice.getDevice().getSiteId();
|
||
SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
|
||
if (siteMemory != null) {
|
||
DeviceRunStatistics deviceRunStatistics = siteMemory.getRunStatistics(deviceId);
|
||
if (deviceRunStatistics != null) {
|
||
LambdaQueryWrapper<DeviceRunStatistics> queryWrapper = new LambdaQueryWrapper<>();
|
||
queryWrapper.eq(DeviceRunStatistics::getDeviceId, deviceId);
|
||
DeviceRunStatistics existing = deviceRunStatisticsMapper
|
||
.selectOne(queryWrapper);
|
||
if (existing != null) {
|
||
deviceRunStatistics.setId(existing.getId());
|
||
deviceRunStatistics.setDistance(
|
||
existing.getDistance() + deviceRunStatistics.getDistance());
|
||
deviceRunStatistics.setWorkArea(
|
||
existing.getWorkArea() + deviceRunStatistics.getWorkArea());
|
||
long time = System.currentTimeMillis() - nettyDevice.getLatestLoginTime();
|
||
deviceRunStatistics.setTime(existing.getTime() + time / (1000 * 60)); // 累加时长，单位为分钟
|
||
} else {
|
||
long time = System.currentTimeMillis() - nettyDevice.getLatestLoginTime();
|
||
deviceRunStatistics.setTime(time / (1000 * 60)); // 累加时长，单位为分钟
|
||
}
|
||
deviceRunStatisticsMapper.saveOrUpdate(deviceRunStatistics);
|
||
}
|
||
}
|
||
|
||
sessionManager.removeDevice(deviceId);
|
||
// TODO 统计总里程、总作业面积
|
||
|
||
}
|
||
} else {
|
||
Device device = deviceService.selectDeviceBySerialNumber(deviceId);
|
||
if (device == null) {
|
||
device = new Device();
|
||
device.setStatus(3);
|
||
device.setOnlineStatus(DeviceStatus.online.getType());
|
||
device.setSerialNumber(deviceId);
|
||
device.setDeviceName(deviceId);
|
||
device.setProductId(-1L);
|
||
device.setTenantId(-1L);
|
||
device.setFirmwareVersion(BigDecimal.valueOf(1.0));
|
||
device.setProductName("割草机产品MC700");
|
||
device.setCreateTime(LocalDateTime.now());
|
||
device.setUpdateTime(LocalDateTime.now());
|
||
device.setHasHost(true);
|
||
deviceService.insertDeviceBySelf(device);
|
||
} else {
|
||
device.setStatus(3);
|
||
device.setOnlineStatus(DeviceStatus.online.getType());
|
||
device.setUpdateTime(LocalDateTime.now());
|
||
device.setHasHost(true);
|
||
deviceService.updateDeviceBySelf(device);
|
||
}
|
||
if (nettyDevice == null) {
|
||
sessionManager.registerSlaveDevice(deviceId, ConnectorStatus.ACTIVE, null);
|
||
NettyDevice slaveDevice = sessionManager.getDevice(deviceId);
|
||
if (slaveDevice != null) {
|
||
slaveDevice.setOnlineStatus(1);
|
||
slaveDevice.setDevice(device);
|
||
}
|
||
} else {
|
||
// todo 推送设备上线
|
||
nettyDevice.status(1);
|
||
nettyDevice.setDevice(device);
|
||
}
|
||
}
|
||
});
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("执行监测心跳线程错误", e);
|
||
}
|
||
}, 0, 500, TimeUnit.MILLISECONDS);
|
||
}
|
||
|
||
|
||
/**
|
||
* 监控心跳
|
||
*/
|
||
public void mapMessagePush() {
|
||
executor.scheduleWithFixedDelay(() -> {
|
||
try {
|
||
List<SiteMemory> list = GlobalMemory.getAllSiteMemory();
|
||
if (!CollectionUtils.isEmpty(list)) {
|
||
list.forEach(x -> {
|
||
Map<String, DeviceMapPushDTO> map = x.getAllDeviceMapPush();
|
||
if (!CollectionUtils.isEmpty(map)) {
|
||
List<DeviceMapPushDTO> pushList = new ArrayList<>(map.values());
|
||
WebMapMessageDTO dto = new WebMapMessageDTO();
|
||
dto.setData(pushList);
|
||
dto.setType(MesType.mapPush);
|
||
try {
|
||
websocketMesDispather.mapMessageDispather(x.getSiteId(), JsonUtils.toJsonString(pushList));
|
||
} catch (Exception e) {
|
||
log.error("实时地图数据websocket推送失败: siteId={}, error={}", x.getSiteId(),
|
||
e.getMessage(), e);
|
||
}
|
||
|
||
}
|
||
});
|
||
}
|
||
} catch (Exception e) {
|
||
log.error("执行监测心跳线程错误：{}", e.getMessage());
|
||
}
|
||
}, 0, 100, TimeUnit.MILLISECONDS);
|
||
}
|
||
|
||
}
|