root
/
agri
2
0
Fork 0
Code Issues Pull Requests Packages Projects 1 Releases Wiki Activity

暂提

feasure
xce 2026-01-16 02:19:52 +08:00
parent 0c520823aa
commit 6cfc8a1393
1 changed files with 12 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
agri-framework/src/main/java/com/agri/framework/interceptor/MqttMessageHandler.java
View File

@ -6,6 +6,8 @@ import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@ -14,6 +16,7 @@ import javax.annotation.PostConstruct;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Set; import java.util.Set;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -46,6 +49,7 @@ public class MqttMessageHandler {
@Value("${spring.mqtt.default-topic:dtu/+/up,frontend/+/control/+}") @Value("${spring.mqtt.default-topic:dtu/+/up,frontend/+/control/+}")
private String defaultTopic; private String defaultTopic;
Logger log = LoggerFactory.getLogger(MqttMessageHandler.class);
/** /**
* 初始化：订阅主题+设置回调 * 初始化：订阅主题+设置回调
*/ */
@ -245,6 +249,14 @@ public class MqttMessageHandler {
* @return 前端需要取消的MQTT主题列表（供前端批量取消） * @return 前端需要取消的MQTT主题列表（供前端批量取消）
*/ */
public List<String> unsubscribeAllDevice(String clientId) { public List<String> unsubscribeAllDevice(String clientId) {
// 适配低版本的Redis连接可用性校验（替换掉isRunning()）
try {
stringRedisTemplate.hasKey("test:connection");
} catch (Exception e) {
log.warn("Redis连接不可用，取消订阅操作跳过：{}", e.getMessage());
return Collections.emptyList();
}
// 步骤1：查询该前端订阅的所有设备ID（Redis中所有sub:*集合中包含该clientId的key） // 步骤1：查询该前端订阅的所有设备ID（Redis中所有sub:*集合中包含该clientId的key）
// 注意：生产环境建议用scan代替keys，避免阻塞Redis // 注意：生产环境建议用scan代替keys，避免阻塞Redis
Set<String> subKeys = stringRedisTemplate.keys("sub:*"); Set<String> subKeys = stringRedisTemplate.keys("sub:*");