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-03-12 16:21:17 +08:00
parent 646b58a21b
commit 3d4d793411
3 changed files with 44 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
message/message-boot/src/main/java/com/njcn/message/redis/RedisKeyExpirationListener.java Normal file
View File

@@ -0,0 +1,43 @@
package com.njcn.message.redis;
import com.njcn.message.websocket.WebSocketServer;
import com.njcn.redis.utils.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
/**
* Description:
* Date: 2025/03/12 下午 3:33【需求编号】
*监听器监听的是所有库的key事件
* @author clam
* @version V1.0.0
*/
@Slf4j
@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
@Autowired
private RedisUtil redisUtil;
public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
@Override
public void onMessage(Message message, byte[] pattern) {
// 用户做自己的业务处理即可,注意message.toString()可以获取失效的key
String expiredKey = message.toString();
if(expiredKey.contains("AskRealData:")){
log.info("失效expiredKey:{}", expiredKey);
String lineId = expiredKey.replace("AskRealData:","");
WebSocketServer.sendInfo("前置连接存在问题",lineId);
redisUtil.delete(expiredKey);
}
}
}

35
message/message-boot/src/main/java/com/njcn/message/websocket/ScheduledTasks.java
View File

@@ -1,35 +0,0 @@
package com.njcn.message.websocket;
import com.njcn.redis.utils.RedisUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* Description:
* Date: 2025/01/14 上午 10:24【需求编号】
*
* @author clam
* @version V1.0.0
*/
@Component
@EnableScheduling
public class ScheduledTasks {
@Autowired
private RedisUtil redisUtil;
@Scheduled(fixedRate = 15000) // 每15s执行一次
public void reportCurrentTime() {
List<String> likeListAllValues = (List<String>) redisUtil.getLikeListAllValues("AskRealData:");
likeListAllValues.forEach(temp->{
String lineId = temp.replace("AskRealData:","");
WebSocketServer.sendInfo("前置连接存在问题",lineId);
redisUtil.delete(temp);
});
}
}

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

@@ -128,7 +128,7 @@ public class WebSocketServer {
baseMessage.setMessageBody(JSONObject.toJSONString(askRealDataMessage));
// 发送消息到topic1
askRealDataMessaggeTemplate.sendMember(baseMessage);
redisUtil.saveByKey("AskRealData:".concat(split[1]),"AskRealData:".concat(split[1]));
redisUtil.saveByKeyWithExpire("AskRealData:".concat(split[1]),"AskRealData:".concat(split[1]),20L);
log.info("监测点连接:" + userIdAndLineIdAndDevId + ",当前在线监测点数为:" + getOnlineCount());
}