tongtong/youlai-boot
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(device): 新增设备在线状态和屏幕状态SSE广播，并修复Redis反序列化问题

Browse Source 
- MqttMessageHandler 收到心跳后通过 SseService 广播设备在线状态（含屏幕状态）
- SseService/SseSessionManager 新增设备在线状态和屏幕状态的广播方法
- DeviceOnlineServiceImpl 修复 Redis 反序列化为 LinkedHashMap 时的类型转换问题
- DeviceOnlineServiceImpl 修复批量查询时空 SN 导致索引不对齐的问题
This commit is contained in:
TongTongStudio
2026-08-09 20:06:03 +08:00
parent ba5b9eddc8
commit 369a940344
5 changed files with 111 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/main/java/com/youlai/boot/device/service/impl/DeviceOnlineServiceImpl.java
View File

@@ -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<String, Object> 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<String> keys = new ArrayList<>(sns.size());
// 记录有效 SN 的索引映射，避免空 SN 导致 keys 和 sns 索引不对齐
List<String> 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) {

10
src/main/java/com/youlai/boot/support/mqtt/MqttMessageHandler.java
View File

@@ -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);
}

53
src/main/java/com/youlai/boot/support/sse/SseService.java
View File

@@ -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());
}
}
}

18
src/main/java/com/youlai/boot/support/sse/SseSessionManager.java
View File

@@ -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 事件，清理已断开的僵尸连接
*/

6
src/main/java/com/youlai/boot/support/sse/constant/SseEventConstants.java
View File

@@ -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";
}