by2025/SmartParks
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(property): 用 Server-Sent Events (SSE) 替代 WebSocket 实现消息推送

Browse Source
This commit is contained in:
zcxlsm
2025-09-06 13:16:40 +08:00
parent 22c6efea63
commit 0bf8cf45e8
8 changed files with 111 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
ruoyi-api/ruoyi-api-resource/src/main/java/org/dromara/resource/api/RemoteMessageService.java
View File

@@ -23,4 +23,13 @@ public interface RemoteMessageService {
* @param message 消息内容
*/
void publishAll(String message);
/**
* 向指定的用户的指定会话发送消息
*
* @param userId 要发送消息的用户id
* @param token 用户的会话令牌
* @param message 要发送的消息内容
*/
void sendMessage(Long userId, String token, String message);
}

16
ruoyi-api/ruoyi-api-resource/src/main/java/org/dromara/resource/api/RemoteMessageServiceStub.java
View File

@@ -44,4 +44,20 @@ public class RemoteMessageServiceStub implements RemoteMessageService {
log.warn("推送功能未开启或服务未找到");
}
}
/**
* 向指定的用户的指定会话发送消息
*
* @param userId 要发送消息的用户id
* @param token 用户的会话令牌
* @param message 要发送的消息内容
*/
@Override
public void sendMessage(Long userId, String token, String message) {
try {
remoteMessageService.sendMessage(userId, token, message);
} catch (Exception e) {
log.warn("推送功能未开启或服务未找到");
}
}
}

28
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java
View File

@@ -132,6 +132,34 @@ public class SseEmitterManager {
}
}
/**
* 向指定的用户的指定会话发送消息
*
* @param userId 要发送消息的用户id
* @param token 用户的会话令牌
* @param message 要发送的消息内容
*/
public void sendMessage(Long userId, String token, String message) {
Map<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.get(userId);
if (MapUtil.isNotEmpty(emitters)) {
for (Map.Entry<String, SseEmitter> entry : emitters.entrySet()){
if (entry.getKey().equals(token)) {
try {
entry.getValue().send(SseEmitter.event()
.name("message")
.data(message));
} catch (Exception e) {
SseEmitter remove = emitters.remove(entry.getKey());
if (remove != null) {
remove.complete();
}
}
}
}
}
}
/**
* 本机全用户会话发送消息
*

14
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/utils/SseMessageUtils.java
View File

@@ -38,6 +38,20 @@ public class SseMessageUtils {
MANAGER.sendMessage(userId, message);
}
/**
* 向指定的用户的指定会话发送消息
*
* @param userId 要发送消息的用户id
* @param token 用户的会话令牌
* @param message 要发送的消息内容
*/
public static void sendMessage(Long userId, String token, String message) {
if (!isEnable()) {
return;
}
MANAGER.sendMessage(userId, token, message);
}
/**
* 本机全用户会话发送消息
*

5
ruoyi-modules/Property/pom.xml
View File

@@ -125,6 +125,11 @@
<artifactId>ruoyi-common-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-sse</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>

2
ruoyi-modules/Property/src/main/java/org/dromara/property/domain/bo/residentBo/ResidentPersonBo.java
View File

@@ -114,7 +114,7 @@ public class ResidentPersonBo extends BaseEntity {
/**
* 状态
*/
private Integer state = 1;
private Integer state;
/**
* 备注

63
ruoyi-modules/Property/src/main/java/org/dromara/property/service/impl/smartDevicesImpl/TbMeterInfoServiceImpl.java
View File

@@ -1,8 +1,10 @@
package org.dromara.property.service.impl.smartDevicesImpl;
import cn.dev33.satoken.stp.StpUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.Assert;
import com.alibaba.fastjson.JSONObject;
import org.apache.dubbo.config.annotation.DubboReference;
import org.dromara.common.core.domain.TreeNode;
import org.dromara.common.core.utils.MapstructUtils;
import org.dromara.common.core.utils.StringUtils;
@@ -15,8 +17,6 @@ import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.satoken.utils.LoginHelper;
import org.dromara.common.websocket.dto.WebSocketMessageDto;
import org.dromara.common.websocket.utils.WebSocketUtils;
import org.dromara.property.domain.bo.TbFloorBo;
import org.dromara.property.domain.vo.TbBuildingVo;
import org.dromara.property.domain.vo.TbCommunityVo;
@@ -27,6 +27,7 @@ import org.dromara.property.service.ITbCommunityService;
import org.dromara.property.service.ITbFloorService;
import org.dromara.property.tasks.HeartbeatTasks;
import org.dromara.property.utils.MeterRecordUtil;
import org.dromara.resource.api.RemoteMessageService;
import org.dromara.system.api.model.LoginUser;
import org.springframework.stereotype.Service;
import org.dromara.property.domain.bo.smartDevicesBo.TbMeterInfoBo;
@@ -61,6 +62,9 @@ public class TbMeterInfoServiceImpl implements ITbMeterInfoService {
private final MeterRecordUtil meterRecordUtil;
private final HeartbeatTasks heartbeatTasks;
@DubboReference
private RemoteMessageService remoteMessageService;
/**
* 查询水电气
*
@@ -85,8 +89,7 @@ public class TbMeterInfoServiceImpl implements ITbMeterInfoService {
Page<TbMeterInfoVo> result = baseMapper.selectVoPage(pageQuery.build(), lqw);
// 创建楼层ID到楼层名称的映射，避免重复查询
List<TbFloorVo> floorList = floorService.queryList(new TbFloorBo());
Map<Long, String> floorMap = floorList.stream()
.collect(Collectors.toMap(TbFloorVo::getId, TbFloorVo::getFloorName, (key1, key2) -> key1));
Map<Long, String> floorMap = floorList.stream().collect(Collectors.toMap(TbFloorVo::getId, TbFloorVo::getFloorName, (key1, key2) -> key1));
// 为每个灯控信息设置楼层名称
result.getRecords().forEach(record -> {
@@ -270,13 +273,13 @@ public class TbMeterInfoServiceImpl implements ITbMeterInfoService {
// 获取当前登录用户
LoginUser user = LoginHelper.getLoginUser();
String tokenValue = StpUtil.getTokenValue();
if (user == null) {
heartbeatTasks.stopTask("Meter_Status_Reading");
return;
}
// 初始化WebSocket消息
WebSocketMessageDto webSocketMessage = new WebSocketMessageDto();
JSONObject jsonObject = new JSONObject();
jsonObject.put("type", "meter");
@@ -291,23 +294,17 @@ public class TbMeterInfoServiceImpl implements ITbMeterInfoService {
heartbeatTasks.stopTask("Meter_Status_Reading");
jsonObject.put("readingTime", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
jsonObject.put("data", new ArrayList<>());
webSocketMessage.setMessage(jsonObject.toString());
webSocketMessage.setSessionKeys(List.of(user.getUserId()));
WebSocketUtils.publishMessage(webSocketMessage);
remoteMessageService.sendMessage(user.getUserId(), tokenValue, jsonObject.toString());
return;
}
// 获取唯一的主机IP列表
String[] hostIpArr = meterInfoVoList.stream()
.map(TbMeterInfoVo::getHostIp)
.distinct()
.toArray(String[]::new);
String[] hostIpArr = meterInfoVoList.stream().map(TbMeterInfoVo::getHostIp).distinct().toArray(String[]::new);
// 统计每个IP对应的仪表数量
Map<String, Long> ipCountMap = new HashMap<>();
for (String ip : hostIpArr) {
ipCountMap.put(ip, baseMapper.selectCount(new LambdaQueryWrapper<TbMeterInfo>()
.eq(TbMeterInfo::getHostIp, ip)));
ipCountMap.put(ip, baseMapper.selectCount(new LambdaQueryWrapper<TbMeterInfo>().eq(TbMeterInfo::getHostIp, ip)));
}
// 启动定时任务
@@ -319,44 +316,39 @@ public class TbMeterInfoServiceImpl implements ITbMeterInfoService {
meterResults = meterRecordUtil.getMeterStatus(ipCountMap, hostIpArr);
} catch (Exception e) {
// 获取数据失败，设置所有仪表通信状态为0并发送消息
handleMeterCommunicationFailure(user, jsonObject, meterInfoVoList, webSocketMessage);
handleMeterCommunicationFailure(user, jsonObject, meterInfoVoList, tokenValue);
return;
}
// 处理仪表读数和状态
processMeterResults(user, jsonObject, meterResults, meterInfoVoList, webSocketMessage);
processMeterResults(user, jsonObject, meterResults, meterInfoVoList, tokenValue);
}, 30000);
}
/**
* 处理仪表通信失败的情况
*
* @param meterInfoVoList 仪表列表
* @param jsonObject 消息对象
* @param webSocketMessage WebSocket消息
* @param user 当前用户
* @param meterInfoVoList 仪表列表
* @param jsonObject 消息对象
* @param user 当前用户
*/
private void handleMeterCommunicationFailure(LoginUser user, JSONObject jsonObject, List<TbMeterInfoVo> meterInfoVoList, WebSocketMessageDto webSocketMessage) {
private void handleMeterCommunicationFailure(LoginUser user, JSONObject jsonObject, List<TbMeterInfoVo> meterInfoVoList, String tokenValue) {
meterInfoVoList.forEach(item -> item.setCommunicationState(0L));
jsonObject.put("data", meterInfoVoList);
webSocketMessage.setMessage(jsonObject.toString());
webSocketMessage.setSessionKeys(List.of(user.getUserId()));
WebSocketUtils.publishMessage(webSocketMessage);
remoteMessageService.sendMessage(user.getUserId(), tokenValue, jsonObject.toString());
}
/**
* 处理仪表读数结果
*
* @param meterInfoVoList 仪表列表
* @param meterResults 读数结果
* @param jsonObject 消息对象
* @param webSocketMessage WebSocket消息
* @param user 当前用户
* @param meterInfoVoList 仪表列表
* @param meterResults 读数结果
* @param jsonObject 消息对象
* @param user 当前用户
*/
private void processMeterResults(LoginUser user, JSONObject jsonObject, List<MeterResult> meterResults, List<TbMeterInfoVo> meterInfoVoList, WebSocketMessageDto webSocketMessage) {
private void processMeterResults(LoginUser user, JSONObject jsonObject, List<MeterResult> meterResults, List<TbMeterInfoVo> meterInfoVoList, String tokenValue) {
// 创建IP到结果的映射，提高查找效率
Map<String, MeterResult> meterResultMap = meterResults.stream()
.collect(Collectors.toMap(MeterResult::getIp, Function.identity(), (v1, v2) -> v1));
Map<String, MeterResult> meterResultMap = meterResults.stream().collect(Collectors.toMap(MeterResult::getIp, Function.identity(), (v1, v2) -> v1));
for (TbMeterInfoVo item : meterInfoVoList) {
MeterResult meterResult = meterResultMap.get(item.getHostIp());
@@ -373,8 +365,7 @@ public class TbMeterInfoServiceImpl implements ITbMeterInfoService {
try {
int codeIndex = Integer.parseInt(meterCode);
if (codeIndex >= 0 && codeIndex < collectionValue.size()) {
BigDecimal initReading = BigDecimal.valueOf(collectionValue.get(codeIndex))
.setScale(2, RoundingMode.HALF_UP);
BigDecimal initReading = BigDecimal.valueOf(collectionValue.get(codeIndex)).setScale(2, RoundingMode.HALF_UP);
item.setInitReading(initReading);
item.setCommunicationState(initReading.compareTo(BigDecimal.ZERO) == 0 ? 0L : 1L);
} else {
@@ -386,8 +377,6 @@ public class TbMeterInfoServiceImpl implements ITbMeterInfoService {
}
jsonObject.put("data", meterInfoVoList);
webSocketMessage.setMessage(jsonObject.toString());
webSocketMessage.setSessionKeys(List.of(user.getUserId()));
WebSocketUtils.publishMessage(webSocketMessage);
remoteMessageService.sendMessage(user.getUserId(), tokenValue, jsonObject.toString());
}
}

12
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/dubbo/RemoteMessageServiceImpl.java
View File

@@ -45,4 +45,16 @@ public class RemoteMessageServiceImpl implements RemoteMessageService {
SseMessageUtils.publishAll(message);
}
/**
* 向指定的用户的指定会话发送消息
*
* @param userId 要发送消息的用户id
* @param token 用户的会话令牌
* @param message 要发送的消息内容
*/
@Override
public void sendMessage(Long userId, String token, String message) {
SseMessageUtils.sendMessage(userId, token, message);
}
}