From 7c585586e9bea943161676bd9d127e81123891c3 Mon Sep 17 00:00:00 2001
From: baoshiwei <baoshiwei@shlanbao.cn>
Date: 星期三, 11 十二月 2024 11:01:35 +0800
Subject: [PATCH] Merge branch 'refs/heads/master' into herb
---
jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttConfig.java | 47 +++++++++++++++++++++++++++++++++++++----------
1 files changed, 37 insertions(+), 10 deletions(-)
diff --git a/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttConfig.java b/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttConfig.java
old mode 100644
new mode 100755
index 4871e16..f669d79
--- a/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttConfig.java
+++ b/jeecg-module-dry/jeecg-module-dry-biz/src/main/java/org/jeecg/modules/dry/mqtt/MqttConfig.java
@@ -15,6 +15,7 @@
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.RedisTemplate;
import java.util.*;
@@ -35,6 +36,8 @@
private String mqttClientId;
@Value(value = "${jeecg.mqtt.role}")
private String role;
+ @Value(value = "${jeecg.mqtt.enable}")
+ private boolean enable;
@Autowired
private MqttSampleCallback mqttSampleCallback;
@@ -44,6 +47,8 @@
private RedisUtil redisUtil;
@Autowired
private EmqxApi emqxApi;
+ @Autowired
+ private RedisTemplate redisTemplate;
@Bean
@@ -56,6 +61,7 @@
* mqtt杩炴帴閰嶇疆
*/
private void conn() {
+ if (!isEnable()) return;
MemoryPersistence persistence = new MemoryPersistence();
MqttConnectOptions mqttConnOpt = new MqttConnectOptions();
mqttConnOpt.setUserName(mqttName);
@@ -97,6 +103,13 @@
// 璁㈤槄绉熸埛瀹炴椂鏁版嵁
mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA);
System.out.println("admin璁㈤槄" + MqttConstant.TENANT_UP_PREFIX_REALTIME_DATA);
+ // 璁㈤槄绉熸埛鎶ヨ鏁版嵁
+ mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_REAL_FAULT_DATA);
+ System.out.println("admin璁㈤槄" + MqttConstant.TENANT_UP_PREFIX_REAL_FAULT_DATA);
+ mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_FAULT_DATA);
+ System.out.println("admin璁㈤槄" + MqttConstant.TENANT_UP_PREFIX_FAULT_DATA);
+ mqttClient.subscribe(MqttConstant.TENANT_UP_PREFIX_EQU);
+ System.out.println("admin璁㈤槄" + MqttConstant.TENANT_UP_PREFIX_EQU);
initClients();
break;
@@ -104,7 +117,7 @@
case "user":
//鏅�氬鎴风鍙渶璁㈤槄鑷韩鐩稿叧娑堟伅
mqttClient.subscribe(MqttConstant.SERVICE_DOWN_PREFIX + "/" + mqttClientId + "/#");
- mqttClient.subscribe(MqttConstant.SERVICE_REQ_PREFIX + "/" + mqttClientId.substring(mqttClientId.lastIndexOf("_")+1) + "/#");
+ mqttClient.subscribe(MqttConstant.SERVICE_REQ_PREFIX);
System.err.println("user璁㈤槄" + MqttConstant.SERVICE_DOWN_PREFIX + "/" + mqttClientId + "/#");
break;
@@ -122,13 +135,14 @@
//閲嶈繛
private void reconn() {
+ if (!isEnable()) return;
Timer timer = new Timer();
TimerTask task = new TimerTask() {
@Override
public void run() {
// 鍦ㄨ繖閲岀紪鍐欏畾鏃舵墽琛岀殑浠诲姟閫昏緫
- System.out.println("瀹氭椂浠诲姟鎵ц锛�" + new java.util.Date());
+ //System.out.println("瀹氭椂浠诲姟鎵ц锛�" + new java.util.Date());
if (mqttUtil.getMqttClient() == null || !mqttUtil.getMqttClient().isConnected()) {
try {
conn();
@@ -148,7 +162,14 @@
* 鏈嶅姟绔紙admin瑙掕壊锛夊惎鍔ㄦ椂鏌ヨ鎵�鏈夎澶囧苟缂撳瓨鍒皉edis
*/
private void initClients() {
- redisUtil.del(MqttConstant.MQTT_ONLINE_CLIENT);
+ //鍒濆鍖栨椂鍏堝垹闄ゆ墍鏈夊湪绾胯澶�
+ Set keys = redisTemplate.keys( String.format(MqttConstant.MQTT_ONLINE_CLIENT,"*"));
+ if (keys != null && !keys.isEmpty()) {
+ keys.forEach(key -> System.out.println("鍒濆鍖栧垹闄ゅ湪绾胯澶�: " + key));
+ redisTemplate.delete(keys);
+ } else {
+ System.out.println("鍒濆鍖栨棤鍦ㄧ嚎璁惧: " + MqttConstant.MQTT_ONLINE_CLIENT);
+ }
JSONObject clients = emqxApi.queryEmqx(EmqxApi.CMD_CLIENTS);
//TODO 鏍规嵁emqx杩斿洖缂栧啓瀹炰綋绫�
@@ -169,17 +190,23 @@
//鏄惁杩炴帴
Boolean connected = obj.getBoolean("connected");
item.put("connected", connected);
- //
- String[] info = clientid.split("-");
- item.put("type", info[0]);
- item.put("tenantId", info[1]);
- item.put("code", info[2]);
+ //鏍规嵁clientid瑙ｆ瀽(娉ㄦ剰閰嶇疆鏂囦欢涓璫lientid鏍煎紡 渚嬶細client-1000)
+ try {
+ String[] info = clientid.split("-");
+ item.put("type", info[0]);
+ item.put("tenantId", info[1]);
+ //item.put("code", info[2]);
- if (connected) {
- redisUtil.set(MqttConstant.MQTT_ONLINE_CLIENT + clientid, item);
+ if (connected) {
+ redisUtil.hset(String.format(MqttConstant.MQTT_ONLINE_CLIENT,item.get("tenantId").toString()) , clientid, item);
+ }
+ }catch (Exception e){
+ e.printStackTrace();
}
+
+
}
}
}
--
Gitblit v1.9.3