From f690786847f9c8eb3d89d04e4c731294ae1cb6d7 Mon Sep 17 00:00:00 2001 From: rqian <1206436827@qq.com> Date: Mon, 14 Sep 2026 13:06:08 +0800 Subject: [PATCH] =?UTF-8?q?#=E6=8E=A5=E5=8F=97=E8=B7=AF=E5=BE=84=E8=A7=84?= =?UTF-8?q?=E5=88=92=E6=8E=A5=E5=8F=A3=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../maibu/memory/DeviceSessionManager.java | 16 ++++++++-- .../com/maibu/mqtt/HostMessageHandler.java | 9 ++++++ .../maibu/service/DeviceThreadService.java | 32 ++++++++----------- 3 files changed, 36 insertions(+), 21 deletions(-) diff --git a/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java b/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java index f6d8b66..1275ff3 100644 --- a/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java +++ b/maibu-common/src/main/java/com/maibu/memory/DeviceSessionManager.java @@ -5,8 +5,12 @@ import com.maibu.core.business.device.Connector; import com.maibu.core.business.device.NettyDevice; import com.maibu.core.enums.ConnectorStatus; import com.maibu.core.enums.ConnectorType; +import com.maibu.utils.json.JsonUtils; + import io.netty.channel.Channel; import lombok.Data; +import lombok.extern.slf4j.Slf4j; + import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -23,6 +27,7 @@ import java.util.stream.Collectors; */ @Component @Data +@Slf4j public class DeviceSessionManager { private static final Logger logger = LoggerFactory.getLogger(DeviceSessionManager.class); @@ -187,7 +192,7 @@ public class DeviceSessionManager { public List getSlaveControl(String slaveId) { // 获取所有连接的主机 且是active状态的 return deviceMap.values().stream() - .filter(device -> ObjectUtil.equals(device.getCurrentSlaveId(), slaveId) + .filter(device -> Objects.equals(device.getCurrentSlaveId(), slaveId) && ConnectorStatus.ACTIVE.equals(device.getConnectorStatus())) .collect(Collectors.toList()); } @@ -195,14 +200,16 @@ public class DeviceSessionManager { public List getAllSlaveControl(String slaveId) { // 获取所有连接的主机 不区分状态 return deviceMap.values().stream() - .filter(device -> ObjectUtil.equals(device.getCurrentSlaveId(), slaveId)) + .filter(device -> !StringUtils.isEmpty(device.getCurrentSlaveId()) + && Objects.equals(device.getCurrentSlaveId(), slaveId) + && device.getDeviceType().equals(ConnectorType.MASTER)) .collect(Collectors.toList()); } public NettyDevice getCurrentControl(String slaveId) { // 获取所有连接的主机 不区分状态 return deviceMap.values().stream() - .filter(device -> ObjectUtil.equals(device.getCurrentSlaveId(), slaveId) + .filter(device -> Objects.equals(device.getCurrentSlaveId(), slaveId) && ConnectorStatus.ACTIVE.equals(device.getConnectorStatus())) .findFirst().orElse(null); } @@ -312,8 +319,11 @@ public class DeviceSessionManager { if (taskId == null) { return null; } + log.debug("sendMessage deviceMap：{},taskId={}", JsonUtils.toJsonString(deviceMap), taskId); for (String key : deviceMap.keySet()) { NettyDevice nettyDevice = deviceMap.get(key); + log.debug("sendMessage currentTaskId:{}，equal {}", JsonUtils.toJsonString(nettyDevice.getCurrentTaskId()), + Objects.equals(nettyDevice.getCurrentTaskId(), taskId)); if (nettyDevice != null && Objects.equals(nettyDevice.getCurrentTaskId(), taskId)) { return nettyDevice.getConnectorId(); } diff --git a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java index 3d91646..5e4529a 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java @@ -3,6 +3,7 @@ package com.maibu.mqtt; import java.util.ArrayList; import java.util.List; +import org.eclipse.paho.client.mqttv3.MqttException; import org.springframework.util.CollectionUtils; import com.fasterxml.jackson.databind.JsonNode; @@ -354,6 +355,14 @@ public class HostMessageHandler implements CustomMqttMessageHandler { } } } + //todo 推送实时位置和状态消息到外部 + try { + GlobalMemory.customMqttDeviceMonitor.publish(mqttClientId, String.format(MqttTopic.DEVICE_TASK_STATUS_TOPIC, taskId), + JsonUtils.toJsonString(sendDto)); + } catch (MqttException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } } if ("cancelled".equals(dto.getStatus())) { diff --git a/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java b/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java index c123ad6..6f3eaa4 100644 --- a/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java +++ b/maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java @@ -285,9 +285,17 @@ public class DeviceThreadService { List 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) { @@ -367,23 +375,11 @@ public class DeviceThreadService { device.setHasHost(true); deviceService.updateDeviceBySelf(device); } - // 推送json 找到设备绑定的主机进行推送 - // List controlMasters = - // sessionManager.getAllSlaveControl(device.getSerialNumber()); - // if (!CollectionUtils.isEmpty(controlMasters)) { - // controlMasters.forEach(x -> { - // DeviceStatusChangeDTO changeDTO = new DeviceStatusChangeDTO(); - // changeDTO.setDeviceId(data); - // changeDTO.setStatus(DeviceStatus.online.getCode()); - // changeDTO.setEvent(RespondCode.device_status_changed); - // ByteBuf buf = ctx.alloc().buffer(); - // CommandUtils.buildCommand(buf, changeDTO, CommandConstant.interaction); - // String content = buf.toString(CharsetUtil.UTF_8); - // logger.info("推送下位机登录状态消息:{}", content); - // if (x.getChannel() != null && x.getChannel().isActive()) { - // x.getChannel().writeAndFlush(buf); - // } - // }); + NettyDevice slaveDevice = sessionManager.getDevice(deviceId); + if (slaveDevice != null) { + slaveDevice.setOnlineStatus(1); + slaveDevice.setDevice(device); + } } else { // todo 推送设备上线 nettyDevice.status(1); @@ -392,7 +388,7 @@ public class DeviceThreadService { }); } } catch (Exception e) { - log.error("执行监测心跳线程错误：{}", e.getMessage()); + log.error("执行监测心跳线程错误", e); } }, 0, 500, TimeUnit.MILLISECONDS); }