diff --git a/src/main/java/com/youlai/boot/device/service/impl/DeviceOnlineServiceImpl.java b/src/main/java/com/youlai/boot/device/service/impl/DeviceOnlineServiceImpl.java index 6857145e..0c3d9f14 100644 --- a/src/main/java/com/youlai/boot/device/service/impl/DeviceOnlineServiceImpl.java +++ b/src/main/java/com/youlai/boot/device/service/impl/DeviceOnlineServiceImpl.java @@ -9,6 +9,7 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; +import tools.jackson.databind.ObjectMapper; import java.time.LocalDateTime; import java.util.ArrayList; @@ -33,6 +34,7 @@ public class DeviceOnlineServiceImpl implements DeviceOnlineService { private final RedisTemplate redisTemplate; private final MqttProperties mqttProperties; + private final ObjectMapper objectMapper; @Override public void reportHeartbeat(String sn, String ip) { @@ -101,6 +103,14 @@ public class DeviceOnlineServiceImpl implements DeviceOnlineService { if (value instanceof DeviceOnlineVO vo) { return vo; } + // GenericJacksonJsonRedisSerializer 反序列化后可能为 LinkedHashMap，需手动转换 + if (value != null) { + try { + return objectMapper.convertValue(value, DeviceOnlineVO.class); + } catch (Exception e) { + log.warn("转换设备在线状态失败, sn={}", sn, e); + } + } // 离线 DeviceOnlineVO offline = new DeviceOnlineVO(); offline.setSerialno(sn); @@ -115,12 +125,15 @@ public class DeviceOnlineServiceImpl implements DeviceOnlineService { return result; } List keys = new ArrayList<>(sns.size()); + // 记录有效 SN 的索引映射，避免空 SN 导致 keys 和 sns 索引不对齐 + List validSns = new ArrayList<>(sns.size()); for (String sn : sns) { if (StrUtil.isBlank(sn)) { continue; } String key = RedisConstants.Device.ONLINE.replace("{}", sn); keys.add(key); + validSns.add(sn); // 预置离线状态，供缺失 key 时填充 DeviceOnlineVO offline = new DeviceOnlineVO(); offline.setSerialno(sn); @@ -132,10 +145,19 @@ public class DeviceOnlineServiceImpl implements DeviceOnlineService { if (values != null) { for (int i = 0; i < values.size(); i++) { Object value = values.get(i); - String sn = sns.get(i); - if (value instanceof DeviceOnlineVO vo) { - result.put(sn, vo); + String sn = validSns.get(i); + if (value == null) { + continue; } + // GenericJacksonJsonRedisSerializer 反序列化 multiGet 结果时 + // 返回 LinkedHashMap，需手动转换为 DeviceOnlineVO + DeviceOnlineVO vo; + if (value instanceof DeviceOnlineVO v) { + vo = v; + } else { + vo = objectMapper.convertValue(value, DeviceOnlineVO.class); + } + result.put(sn, vo); } } } catch (Exception e) { diff --git a/src/main/java/com/youlai/boot/support/mqtt/MqttMessageHandler.java b/src/main/java/com/youlai/boot/support/mqtt/MqttMessageHandler.java index 54d9c650..64eb060a 100644 --- a/src/main/java/com/youlai/boot/support/mqtt/MqttMessageHandler.java +++ b/src/main/java/com/youlai/boot/support/mqtt/MqttMessageHandler.java @@ -5,6 +5,7 @@ import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import com.youlai.boot.config.property.MqttProperties; import com.youlai.boot.device.service.DeviceOnlineService; +import com.youlai.boot.support.sse.SseService; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.integration.annotation.ServiceActivator; @@ -31,10 +32,14 @@ public class MqttMessageHandler implements MessageHandler { private final DeviceOnlineService deviceOnlineService; private final MqttProperties mqttProperties; + private final SseService sseService; - public MqttMessageHandler(DeviceOnlineService deviceOnlineService, MqttProperties mqttProperties) { + public MqttMessageHandler(DeviceOnlineService deviceOnlineService, + MqttProperties mqttProperties, + SseService sseService) { this.deviceOnlineService = deviceOnlineService; this.mqttProperties = mqttProperties; + this.sseService = sseService; } /** @@ -95,6 +100,9 @@ public class MqttMessageHandler implements MessageHandler { // 默认视为心跳/在线，刷新 Redis TTL，并同步屏幕状态 deviceOnlineService.reportHeartbeat(sn, ip, screenState); } + + // Redis 更新后，通过 SSE 广播设备在线状态（含屏幕状态） + sseService.broadcastDeviceOnlineStatus(sn); } catch (Exception e) { log.error("解析设备状态上报失败, payload={}", payload, e); } diff --git a/src/main/java/com/youlai/boot/support/sse/SseService.java b/src/main/java/com/youlai/boot/support/sse/SseService.java index 6613a78d..e2ad037a 100644 --- a/src/main/java/com/youlai/boot/support/sse/SseService.java +++ b/src/main/java/com/youlai/boot/support/sse/SseService.java @@ -1,5 +1,7 @@ package com.youlai.boot.support.sse; +import com.youlai.boot.device.model.vo.DeviceOnlineVO; +import com.youlai.boot.device.service.DeviceOnlineService; import com.youlai.boot.support.sse.constant.SseEventConstants; import com.youlai.boot.support.sse.constant.SseTopics; import com.youlai.boot.support.sse.model.OnlineUserDTO; @@ -26,6 +28,7 @@ import java.util.Map; public class SseService { private final SseSessionManager sessionManager; + private final DeviceOnlineService deviceOnlineService; /** * 创建 SSE 连接 @@ -129,4 +132,54 @@ public class SseService { public int getOnlineUserCount() { return sessionManager.getOnlineUserCount(); } + + /** + * 广播设备在线状态（从 Redis 读取后推送） + * + * @param sn 设备序列号 + */ + public void broadcastDeviceOnlineStatus(String sn) { + DeviceOnlineVO deviceInfo = deviceOnlineService.getOnlineInfo(sn); + if (deviceInfo != null) { + sessionManager.broadcastDeviceOnlineStatus(deviceInfo); + log.debug("已广播设备[{}]在线状态: online={}, screenState={}", sn, deviceInfo.getOnline(), deviceInfo.getScreenState()); + } + } + + /** + * 广播设备屏幕状态（从 Redis 读取后推送） + * + * @param sn 设备序列号 + */ + public void broadcastDeviceScreenStatus(String sn) { + DeviceOnlineVO deviceInfo = deviceOnlineService.getOnlineInfo(sn); + if (deviceInfo != null) { + sessionManager.broadcastDeviceScreenStatus(deviceInfo); + log.debug("已广播设备[{}]屏幕状态: screenState={}", sn, deviceInfo.getScreenState()); + } + } + + /** + * 直接广播设备在线状态数据（无需从 Redis 查询，调用方已有数据时使用） + * + * @param deviceData 设备在线状态数据 + */ + public void broadcastDeviceOnlineStatus(DeviceOnlineVO deviceData) { + if (deviceData != null) { + sessionManager.broadcastDeviceOnlineStatus(deviceData); + log.debug("已广播设备[{}]在线状态: online={}", deviceData.getSerialno(), deviceData.getOnline()); + } + } + + /** + * 直接广播设备屏幕状态数据（无需从 Redis 查询，调用方已有数据时使用） + * + * @param deviceData 设备在线状态数据 + */ + public void broadcastDeviceScreenStatus(DeviceOnlineVO deviceData) { + if (deviceData != null) { + sessionManager.broadcastDeviceScreenStatus(deviceData); + log.debug("已广播设备[{}]屏幕状态: screenState={}", deviceData.getSerialno(), deviceData.getScreenState()); + } + } } diff --git a/src/main/java/com/youlai/boot/support/sse/SseSessionManager.java b/src/main/java/com/youlai/boot/support/sse/SseSessionManager.java index 55c25b24..c6a05686 100644 --- a/src/main/java/com/youlai/boot/support/sse/SseSessionManager.java +++ b/src/main/java/com/youlai/boot/support/sse/SseSessionManager.java @@ -217,6 +217,24 @@ public class SseSessionManager { } } + /** + * 广播设备在线状态到所有 SSE 连接 + * + * @param deviceData 设备在线状态数据 + */ + public void broadcastDeviceOnlineStatus(Object deviceData) { + broadcast(SseEventConstants.DEVICE_ONLINE_STATUS, deviceData); + } + + /** + * 广播设备屏幕状态到所有 SSE 连接 + * + * @param screenData 设备屏幕状态数据 + */ + public void broadcastDeviceScreenStatus(Object screenData) { + broadcast(SseEventConstants.DEVICE_SCREEN_STATUS, screenData); + } + /** * 心跳检测：每 30 秒向所有连接发送 ping 事件，清理已断开的僵尸连接 */ diff --git a/src/main/java/com/youlai/boot/support/sse/constant/SseEventConstants.java b/src/main/java/com/youlai/boot/support/sse/constant/SseEventConstants.java index 4bd71119..9e6155d4 100644 --- a/src/main/java/com/youlai/boot/support/sse/constant/SseEventConstants.java +++ b/src/main/java/com/youlai/boot/support/sse/constant/SseEventConstants.java @@ -16,4 +16,10 @@ public final class SseEventConstants { /** 在线用户数事件 */ public static final String ONLINE_USERS = "online-users"; + + /** 设备在线状态事件 */ + public static final String DEVICE_ONLINE_STATUS = "device-online-status"; + + /** 设备屏幕状态事件 */ + public static final String DEVICE_SCREEN_STATUS = "device-screen-status"; }