dyf/fys-Multi-tenant
1
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

在线状态修改优化

Browse Source
This commit is contained in:
chenyouting
2025-09-13 16:46:05 +08:00
parent 64c81ac44e
commit 39c8375f08
3 changed files with 20 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
fys-admin/src/main/java/com/fuyuanshen/global/mqtt/listener/RedisKeyExpirationListener.java
View File

@ -49,14 +49,6 @@ public class RedisKeyExpirationListener implements MessageListener {
handleFunctionAccessExpired(element); handleFunctionAccessExpired(element);
} }
if(expiredKey.endsWith(DEVICE_ONLINE_STATUS_KEY_PREFIX)) { if(expiredKey.endsWith(DEVICE_ONLINE_STATUS_KEY_PREFIX)) {
// threadPoolTaskExecutor.execute(() -> {
// log.info("设备离线：{}", expiredKey);
// String element = expiredKey.substring(GlobalConstants.GLOBAL_REDIS_KEY.length() + DEVICE_KEY_PREFIX.length(), expiredKey.length() - DEVICE_ONLINE_STATUS_KEY_PREFIX.length());
// UpdateWrapper<Device> deviceUpdateWrapper = new UpdateWrapper<>();
// deviceUpdateWrapper.eq("device_imei", element);
// deviceUpdateWrapper.set("online_status", 0);
// deviceMapper.update(deviceUpdateWrapper);
// });
threadPoolTaskExecutor.execute(() -> { threadPoolTaskExecutor.execute(() -> {
log.info("设备离线：{}", expiredKey); log.info("设备离线：{}", expiredKey);
@ -72,9 +64,9 @@ public class RedisKeyExpirationListener implements MessageListener {
try { try {
QueryWrapper<Device> queryWrapper = new QueryWrapper<>(); QueryWrapper<Device> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("device_imei", element); queryWrapper.eq("device_imei", element);
queryWrapper.eq("online_status", 0); queryWrapper.eq("online_status", 1);
Long count = deviceMapper.selectCount(queryWrapper); Long count = deviceMapper.selectCount(queryWrapper);
if(count == 0){ if(count > 0){
UpdateWrapper<Device> deviceUpdateWrapper = new UpdateWrapper<>(); UpdateWrapper<Device> deviceUpdateWrapper = new UpdateWrapper<>();
deviceUpdateWrapper.eq("device_imei", element); deviceUpdateWrapper.eq("device_imei", element);
deviceUpdateWrapper.set("online_status", 0); deviceUpdateWrapper.set("online_status", 0);

6
fys-admin/src/main/java/com/fuyuanshen/global/mqtt/receiver/ReceiverMessageHandler.java
View File

@ -49,12 +49,12 @@ public class ReceiverMessageHandler implements MessageHandler {
String[] subStr = receivedTopic.split("/"); String[] subStr = receivedTopic.split("/");
String deviceImei = subStr[1]; String deviceImei = subStr[1];
if(StringUtils.isNotBlank(deviceImei)){ if(StringUtils.isNotBlank(deviceImei)){
//在线状态
String deviceOnlineStatusRedisKey = GlobalConstants.GLOBAL_REDIS_KEY+ DEVICE_KEY_PREFIX+ deviceImei + DeviceRedisKeyConstants.DEVICE_ONLINE_STATUS_KEY_PREFIX ;
RedisUtils.setCacheObject(deviceOnlineStatusRedisKey, "1", Duration.ofSeconds(62));
String queueKey = MqttMessageQueueConstants.MQTT_MESSAGE_QUEUE_KEY; String queueKey = MqttMessageQueueConstants.MQTT_MESSAGE_QUEUE_KEY;
String dedupKey = MqttMessageQueueConstants.MQTT_MESSAGE_DEDUP_KEY; String dedupKey = MqttMessageQueueConstants.MQTT_MESSAGE_DEDUP_KEY;
RedisUtils.offerDeduplicated(queueKey,dedupKey,deviceImei, Duration.ofHours(24)); RedisUtils.offerDeduplicated(queueKey,dedupKey,deviceImei, Duration.ofHours(24));
//在线状态
String deviceOnlineStatusRedisKey = GlobalConstants.GLOBAL_REDIS_KEY+ DEVICE_KEY_PREFIX+ deviceImei + DeviceRedisKeyConstants.DEVICE_ONLINE_STATUS_KEY_PREFIX ;
RedisUtils.setCacheObject(deviceOnlineStatusRedisKey, "1", Duration.ofSeconds(62));
} }
String state = payloadDict.getStr("state"); String state = payloadDict.getStr("state");

21
fys-admin/src/main/java/com/fuyuanshen/global/queue/MqttMessageConsumer.java
View File

@ -2,9 +2,12 @@ package com.fuyuanshen.global.queue;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper; import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.fuyuanshen.common.core.constant.GlobalConstants;
import com.fuyuanshen.common.core.utils.StringUtils;
import com.fuyuanshen.common.redis.utils.RedisUtils; import com.fuyuanshen.common.redis.utils.RedisUtils;
import com.fuyuanshen.equipment.domain.Device; import com.fuyuanshen.equipment.domain.Device;
import com.fuyuanshen.equipment.mapper.DeviceMapper; import com.fuyuanshen.equipment.mapper.DeviceMapper;
import com.fuyuanshen.global.mqtt.constants.DeviceRedisKeyConstants;
import jakarta.annotation.PostConstruct; import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy; import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -16,6 +19,8 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static com.fuyuanshen.global.mqtt.constants.DeviceRedisKeyConstants.DEVICE_KEY_PREFIX;
@Service @Service
@Slf4j @Slf4j
public class MqttMessageConsumer { public class MqttMessageConsumer {
@ -93,17 +98,21 @@ public class MqttMessageConsumer {
String threadName = Thread.currentThread().getName(); String threadName = Thread.currentThread().getName();
try { try {
log.info("业务处理线程 {} 开始处理消息: {}", threadName, message); log.info("业务处理线程 {} 开始处理消息: {}", threadName, message);
String deviceOnlineStatusRedisKey = GlobalConstants.GLOBAL_REDIS_KEY+ DEVICE_KEY_PREFIX+ message + DeviceRedisKeyConstants.DEVICE_ONLINE_STATUS_KEY_PREFIX ;
QueryWrapper<Device> queryWrapper = new QueryWrapper<>(); String deviceOnlineStatusRedis = RedisUtils.getCacheObject(deviceOnlineStatusRedisKey);
queryWrapper.eq("device_imei", message); if(StringUtils.isBlank(deviceOnlineStatusRedis)){
queryWrapper.eq("online_status", 1);
Long count = deviceMapper.selectCount(queryWrapper);
if(count == 0){
UpdateWrapper<Device> updateWrapper = new UpdateWrapper<>(); UpdateWrapper<Device> updateWrapper = new UpdateWrapper<>();
updateWrapper.eq("device_imei", message) updateWrapper.eq("device_imei", message)
.set("online_status", 1); .set("online_status", 1);
deviceMapper.update(updateWrapper); deviceMapper.update(updateWrapper);
} }
// QueryWrapper<Device> queryWrapper = new QueryWrapper<>();
// queryWrapper.eq("device_imei", message);
// queryWrapper.eq("online_status", 1);
// Long count = deviceMapper.selectCount(queryWrapper);
// if(count == 0){
//
// }
// 模拟业务处理耗时 // 模拟业务处理耗时
// Thread.sleep(200); // Thread.sleep(200);