Middleware/cn-data-platform
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始版本提交

Browse Source
This commit is contained in:
hzj
2025-01-21 16:22:57 +08:00
parent 04e3f48da8
commit 16f15b7cdd
6 changed files with 69 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
message/message-boot/src/main/java/com/njcn/message/websocket/WebSocketServer.java
View File

@@ -30,7 +30,7 @@ import java.util.concurrent.ConcurrentHashMap;
*/
@Slf4j
@Component
@ServerEndpoint(value ="/api/pushMessage/{userIdAndlineId}")
@ServerEndpoint(value ="/api/pushMessage/{userIdAndLineIdAndDevId}")
public class WebSocketServer {
@@ -46,13 +46,13 @@ public class WebSocketServer {
}
private static LineFeignClient lineFeignClient;
// 注入的时候，给类的 service 注入
@Autowired
public void setLineFeignClient(LineFeignClient lineFeignClient) {
WebSocketServer.lineFeignClient = lineFeignClient;
}
// private static LineFeignClient lineFeignClient;
//
// // 注入的时候，给类的 service 注入
// @Autowired
// public void setLineFeignClient(LineFeignClient lineFeignClient) {
// WebSocketServer.lineFeignClient = lineFeignClient;
// }
/**
* 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
*/
@@ -68,24 +68,24 @@ public class WebSocketServer {
/**
* 接收userId
*/
private String userIdAndlineId = "";
private String userIdAndLineIdAndDevId = "";
/**
* 连接建立成
* 功调用的方法
*/
@OnOpen
public void onOpen(Session session, @PathParam("userIdAndlineId") String userIdAndlineId) {
//lineId 是 userid+","+lineId
public void onOpen(Session session, @PathParam("userIdAndLineIdAndDevId") String userIdAndLineIdAndDevId) {
//lineId 是 userid+","+lineId+","+Devid
this.session = session;
this.userIdAndlineId = userIdAndlineId;
if (webSocketMap.containsKey(userIdAndlineId)) {
webSocketMap.remove(userIdAndlineId);
this.userIdAndLineIdAndDevId = userIdAndLineIdAndDevId;
if (webSocketMap.containsKey(userIdAndLineIdAndDevId)) {
webSocketMap.remove(userIdAndLineIdAndDevId);
//加入set中
webSocketMap.put(userIdAndlineId, this);
webSocketMap.put(userIdAndLineIdAndDevId, this);
} else {
//加入set中
webSocketMap.put(userIdAndlineId, this);
webSocketMap.put(userIdAndLineIdAndDevId, this);
//在线数加1
addOnlineCount();
}
@@ -100,12 +100,12 @@ public class WebSocketServer {
*/
@OnClose
public void onClose() {
if (webSocketMap.containsKey(userIdAndlineId)) {
webSocketMap.remove(userIdAndlineId);
if (webSocketMap.containsKey(userIdAndLineIdAndDevId)) {
webSocketMap.remove(userIdAndLineIdAndDevId);
//从set中删除
subOnlineCount();
}
log.info("监测点退出:" + userIdAndlineId + ",当前在线监测点数为:" + getOnlineCount());
log.info("监测点退出:" + userIdAndLineIdAndDevId + ",当前在线监测点数为:" + getOnlineCount());
}
/**
@@ -117,20 +117,20 @@ public class WebSocketServer {
@OnMessage
public void onMessage(String message, Session session) {
//会每30s发送请求1次
log.info("监测点消息:" + userIdAndlineId + ",报文:" + message);
log.info("监测点消息:" + userIdAndLineIdAndDevId + ",报文:" + message);
AskRealDataMessage askRealDataMessage = new AskRealDataMessage();
String[] split = userIdAndlineId.split(",");
String[] split = userIdAndLineIdAndDevId.split(",");
askRealDataMessage.setLine(split[1]);
//测试设备187
// askRealDataMessage.setDevSeries("fed656b21f89abb06204e8e4dc6c375a");
LineDetailDataVO data = lineFeignClient.getLineDetailData(split[1]).getData();
askRealDataMessage.setDevSeries(data.getDevId());
// LineDetailDataVO data = lineFeignClient.getLineDetailData(split[1]).getData();
askRealDataMessage.setDevSeries(split[2]);
BaseMessage baseMessage = new BaseMessage();
baseMessage.setMessageBoy(JSONObject.toJSONString(askRealDataMessage));
// 发送消息到topic1
askRealDataMessaggeTemplate.sendMember(baseMessage);
redisUtil.saveByKey("AskRealData:".concat(split[1]),"AskRealData:".concat(split[1]));
log.info("监测点连接:" + userIdAndlineId + ",当前在线监测点数为:" + getOnlineCount());
log.info("监测点连接:" + userIdAndLineIdAndDevId + ",当前在线监测点数为:" + getOnlineCount());
}
@@ -142,7 +142,7 @@ public class WebSocketServer {
@OnError
public void onError(Session session, Throwable error) {
log.error("监测点错误:" + this.userIdAndlineId + ",原因:" + error.getMessage());
log.error("监测点错误:" + this.userIdAndLineIdAndDevId + ",原因:" + error.getMessage());
error.printStackTrace();
}