From 2eea735fd4ddf0ae047687780271ef3962d256cc Mon Sep 17 00:00:00 2001
From: 张世豪 <979909237@qq.com>
Date: 星期四, 25 十二月 2025 15:54:26 +0800
Subject: [PATCH] 替换了mqtt最新版
---
src/Mqttmessage/Client.java | 458 ++++++++++++++++++++++++++++++++++++++++++++++++++++++---
1 files changed, 432 insertions(+), 26 deletions(-)
diff --git a/src/Mqttmessage/Client.java b/src/Mqttmessage/Client.java
index 90ff9f1..ca1e876 100644
--- a/src/Mqttmessage/Client.java
+++ b/src/Mqttmessage/Client.java
@@ -3,6 +3,10 @@
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+
+import set.Setsys;
+import user.Usrdell;
/**
* MQTT瀹㈡埛绔伐鍏风被
@@ -15,6 +19,15 @@
private String clientId;
private MqttClient client;
private MqttConnectOptions options;
+ private int qos = 0; // 榛樿QoS绛夌骇
+ private Thread connectionMonitorThread; // 杩炴帴鐩戞帶绾跨▼
+ private volatile boolean monitoring = false; // 鐩戞帶鏍囧織
+
+ // 闈欐�佸彉閲忕敤浜庡瓨鍌ㄥ鎴风瀹炰緥
+ private static Client gpsClient;
+ private static Client responseClient;
+ // 閲嶈繛鏍囧織锛岄槻姝㈤噸澶嶉噸杩�
+ private static volatile boolean isReconnecting = false;
/**
* 鏋勯�犲嚱鏁�
@@ -23,12 +36,19 @@
* @param clientId 瀹㈡埛绔疘D锛屼笉鑳介噸澶�
*/
public Client(String host, String topic, String clientId) {
-
this.host = host;
this.topic = topic;
this.clientId = clientId;
this.options = new MqttConnectOptions();
this.options.setCleanSession(true);
+ // 璁剧疆杩炴帴瓒呮椂鏃堕棿锛堢锛�
+ this.options.setConnectionTimeout(30);
+ // 璁剧疆KeepAlive闂撮殧锛堢锛夛紝鐢ㄤ簬淇濇寔杩炴帴娲昏穬
+ this.options.setKeepAliveInterval(60);
+ // 璁剧疆鑷姩閲嶈繛
+ this.options.setAutomaticReconnect(true);
+ // 璁剧疆MQTT鐗堟湰锛屼娇鐢�3.1.1
+ this.options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
}
/**
@@ -37,11 +57,31 @@
*/
public void connect() throws MqttException {
if (client != null && client.isConnected()) {
+ System.out.println("MQTT瀹㈡埛绔凡杩炴帴锛孋lientId: " + clientId);
return;
}
- client = new MqttClient(host, clientId);
+
+ // 濡傛灉瀹㈡埛绔凡瀛樺湪浣嗘湭杩炴帴锛屽厛鍏抽棴
+ if (client != null) {
+ try {
+ client.close();
+ } catch (Exception e) {
+ // 蹇界暐鍏抽棴鏃剁殑寮傚父
+ }
+ client = null;
+ }
+
+ // 浣跨敤鍐呭瓨鎸佷箙鍖栵紝閬垮厤鏂囦欢閿佸畾闂
+ client = new MqttClient(host, clientId, new MemoryPersistence());
+ // 鍏堣缃洖璋冿紝鍐嶈繛鎺ワ紙浼犲叆Client瀹炰緥鍜岃闃呬俊鎭紝鐢ㄤ簬鑷姩閲嶈繛鍚庨噸鏂拌闃咃級
+ client.setCallback(new PushCallback(this, topic, qos));
+
+ // 鎵ц杩炴帴
client.connect(options);
- client.setCallback(new PushCallback());
+ System.out.println("MQTT杩炴帴鎴愬姛锛丆lientId: " + clientId + ", 鏈嶅姟鍣�: " + host + ", 涓婚: " + topic);
+
+ // 鍚姩杩炴帴鐩戞帶绾跨▼
+ startConnectionMonitor();
}
/**
@@ -50,10 +90,12 @@
* @throws MqttException 璁㈤槄澶辫触鏃舵姏鍑哄紓甯�
*/
public void subscribe(int qos) throws MqttException {
+ this.qos = qos; // 淇濆瓨QoS鍊硷紝鐢ㄤ簬閲嶈繛鍚庨噸鏂拌闃�
if (client == null || !client.isConnected()) {
connect();
}
client.subscribe(topic, qos);
+ System.out.println("宸茶闃呬富棰�: " + topic + ", QoS: " + qos);
}
/**
@@ -61,7 +103,8 @@
* @throws MqttException 璁㈤槄澶辫触鏃舵姏鍑哄紓甯�
*/
public void subscribe() throws MqttException {
- subscribe(2);
+ // 浣跨敤QoS 0纭繚娑堟伅鍙潬浼犻�掞紙鑷冲皯涓�娆★紝涓斾粎涓�娆★級
+ subscribe(0);
}
/**
@@ -75,6 +118,93 @@
}
/**
+ * 鍚姩杩炴帴鐩戞帶绾跨▼锛屽畾鏈熸鏌ヨ繛鎺ョ姸鎬佸苟鍦ㄦ柇寮�鏃跺皾璇曢噸杩�
+ */
+ private void startConnectionMonitor() {
+ if (monitoring) {
+ return; // 宸茬粡鍦ㄧ洃鎺т腑
+ }
+
+ monitoring = true;
+ connectionMonitorThread = new Thread(() -> {
+ while (monitoring && client != null) {
+ try {
+ Thread.sleep(5000); // 姣�5绉掓鏌ヤ竴娆�
+
+ if (client != null && !client.isConnected()) {
+ System.out.println("妫�娴嬪埌MQTT杩炴帴鏂紑锛屽皾璇曢噸杩�... ClientId: " + clientId);
+ try {
+ // 灏濊瘯閲嶆柊杩炴帴
+ if (!client.isConnected()) {
+ client.reconnect();
+ // 閲嶈繛鎴愬姛鍚庨噸鏂拌闃�
+ if (client.isConnected()) {
+ client.subscribe(topic, qos);
+ System.out.println("杩炴帴鐩戞帶锛氶噸杩炴垚鍔熷苟閲嶆柊璁㈤槄涓婚: " + topic);
+ }
+ }
+ } catch (Exception e) {
+ System.err.println("杩炴帴鐩戞帶锛氶噸杩炲け璐�: " + e.getMessage());
+ // 濡傛灉鑷姩閲嶈繛澶辫触锛屽皾璇曞畬鍏ㄩ噸鏂拌繛鎺�
+ try {
+ connect();
+ if (isConnected()) {
+ subscribe(qos);
+ }
+ } catch (Exception ex) {
+ System.err.println("杩炴帴鐩戞帶锛氬畬鍏ㄩ噸杩炲け璐�: " + ex.getMessage());
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ } catch (Exception e) {
+ System.err.println("杩炴帴鐩戞帶绾跨▼寮傚父: " + e.getMessage());
+ }
+ }
+ });
+ connectionMonitorThread.setDaemon(true);
+ connectionMonitorThread.setName("MQTT-ConnectionMonitor-" + clientId);
+ connectionMonitorThread.start();
+ System.out.println("宸插惎鍔∕QTT杩炴帴鐩戞帶绾跨▼: " + clientId);
+ }
+
+ /**
+ * 鍋滄杩炴帴鐩戞帶绾跨▼
+ */
+ private void stopConnectionMonitor() {
+ monitoring = false;
+ if (connectionMonitorThread != null) {
+ connectionMonitorThread.interrupt();
+ try {
+ connectionMonitorThread.join(1000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ connectionMonitorThread = null;
+ }
+ }
+
+ /**
+ * 鍏抽棴瀹㈡埛绔苟閲婃斁璧勬簮
+ */
+ public void close() {
+ stopConnectionMonitor(); // 鍋滄鐩戞帶绾跨▼
+ try {
+ if (client != null) {
+ if (client.isConnected()) {
+ client.disconnect();
+ }
+ client.close();
+ client = null;
+ }
+ } catch (Exception e) {
+ // 蹇界暐鍏抽棴鏃剁殑寮傚父
+ }
+ }
+
+ /**
* 妫�鏌ユ槸鍚﹀凡杩炴帴
* @return true琛ㄧず宸茶繛鎺ワ紝false琛ㄧず鏈繛鎺�
*/
@@ -91,29 +221,305 @@
}
/**
- * 绀轰緥鐢ㄦ硶
+ * 杩炴帴MQTT鏈嶅姟鍣ㄧ殑宸ュ叿鏂规硶
+ * 渚涘叾浠栫被鐩存帴璋冪敤锛岃繛鎺PS涓婚鍜屽搷搴斾富棰�
+ * @return true琛ㄧず杩炴帴鎴愬姛锛宖alse琛ㄧず杩炴帴澶辫触
*/
-
- public static void test() {
- try {
- String host = "tcp://39.99.43.227:1883";
- String deiveID="6258";
- String clientId = "hxzkMQTT";
- String clientId2 = "hxzkMQTT2";
- String topic = "mower/"+deiveID+"/gps";
- String topic2 = "mower/"+deiveID+"/response";
- Client mqttClient = new Client(host, topic, clientId);
- Client mqttClient1 = new Client(host, topic2, clientId2);
- mqttClient.connect();
- mqttClient.subscribe();
-
- mqttClient1.connect();
- mqttClient1.subscribe();
-
- // 淇濇寔绋嬪簭杩愯
- //Thread.sleep(Long.MAX_VALUE);
- } catch (MqttException e) {
- throw new RuntimeException(e);
+ public static boolean connectMQTT() {
+ // 闃叉閲嶅閲嶈繛
+ if (isReconnecting) {
+ System.out.println("MQTT姝ｅ湪閲嶈繛涓紝璺宠繃鏈閲嶈繛璇锋眰");
+ return false;
}
+
+ // 妫�鏌ユ槸鍚﹀凡缁忚繛鎺�
+ if (areClientsConnected()) {
+ System.out.println("MQTT瀹㈡埛绔凡杩炴帴锛屾棤闇�閲嶅杩炴帴");
+ return true;
+ }
+
+ isReconnecting = true;
+ try {
+ // 鍏堟柇寮�涔嬪墠鐨勮繛鎺�
+ disconnectAll();
+
+ boolean gpsSuccess = false;
+ boolean responseSuccess = false;
+
+ try {
+ String host = "tcp://39.99.43.227:1883";
+ String deiveID = Setsys.getMowerIdValue();
+ if (deiveID == null || deiveID.isEmpty()) {
+ System.err.println("璁惧ID涓虹┖锛屾棤娉曡繛鎺QTT");
+ return false;
+ }
+ // 娣诲姞鏃堕棿鎴崇‘淇濆鎴风ID鍞竴
+ long timestamp = System.currentTimeMillis();
+ String clientId = Usrdell.getUserEmail() + "mower" + "_" + timestamp;
+ String clientId2 = Usrdell.getUserEmail() + "response" + "_" + timestamp;
+ String topic = "mower/" + deiveID + "/gps";
+ String topic2 = "mower/" + deiveID + "/response";
+
+ // 杩炴帴GPS涓婚
+ try {
+ gpsClient = new Client(host, topic, clientId);
+ gpsClient.connect();
+ // 绋嶄綔寤惰繜锛岀‘淇濊繛鎺ョǔ瀹�
+ Thread.sleep(100);
+ if (gpsClient != null && gpsClient.isConnected()) {
+ gpsClient.subscribe();
+ gpsSuccess = true;
+ System.out.println("GPS涓婚MQTT杩炴帴骞惰闃呮垚鍔�");
+ }
+ } catch (MqttException e) {
+ System.err.println("GPS涓婚MQTT杩炴帴澶辫触: " + e.getMessage());
+ if (e.getCause() != null) {
+ System.err.println("澶辫触鍘熷洜: " + e.getCause().getMessage());
+ }
+ e.printStackTrace();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ System.err.println("杩炴帴杩囩▼琚腑鏂�");
+ }
+
+ // 杩炴帴鍝嶅簲涓婚
+ try {
+ responseClient = new Client(host, topic2, clientId2);
+ responseClient.connect();
+ // 绋嶄綔寤惰繜锛岀‘淇濊繛鎺ョǔ瀹�
+ Thread.sleep(100);
+ if (responseClient != null && responseClient.isConnected()) {
+ responseClient.subscribe();
+ responseSuccess = true;
+ System.out.println("鍝嶅簲涓婚MQTT杩炴帴骞惰闃呮垚鍔�");
+ }
+ } catch (MqttException e) {
+ System.err.println("鍝嶅簲涓婚MQTT杩炴帴澶辫触: " + e.getMessage());
+ if (e.getCause() != null) {
+ System.err.println("澶辫触鍘熷洜: " + e.getCause().getMessage());
+ }
+ e.printStackTrace();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ System.err.println("杩炴帴杩囩▼琚腑鏂�");
+ }
+
+ if (gpsSuccess && responseSuccess) {
+ System.out.println("鎵�鏈塎QTT涓婚杩炴帴骞惰闃呮垚鍔燂紒");
+ return true;
+ } else if (gpsSuccess || responseSuccess) {
+ System.out.println("閮ㄥ垎MQTT涓婚杩炴帴鎴愬姛");
+ return true;
+ } else {
+ System.err.println("鎵�鏈塎QTT涓婚杩炴帴澶辫触");
+ return false;
+ }
+ } catch (Exception e) {
+ System.err.println("MQTT杩炴帴杩囩▼鍙戠敓寮傚父: " + e.getMessage());
+ if (e.getCause() != null) {
+ System.err.println("寮傚父鍘熷洜: " + e.getCause().getMessage());
+ }
+ e.printStackTrace();
+ return false;
+ }
+ } finally {
+ isReconnecting = false;
+ }
+ }
+
+ /**
+ * 杩炴帴MQTT鏈嶅姟鍣ㄧ殑宸ュ叿鏂规硶锛堝甫鍙傛暟鐗堟湰锛�
+ * @param host MQTT鏈嶅姟鍣ㄥ湴鍧�锛屾牸寮忥細tcp://ip:port
+ * @param deviceId 璁惧ID
+ * @param userEmail 鐢ㄦ埛閭
+ * @return true琛ㄧず杩炴帴鎴愬姛锛宖alse琛ㄧず杩炴帴澶辫触
+ */
+ public static boolean connectMQTT(String host, String deviceId, String userEmail) {
+ // 闃叉閲嶅閲嶈繛
+ if (isReconnecting) {
+ System.out.println("MQTT姝ｅ湪閲嶈繛涓紝璺宠繃鏈閲嶈繛璇锋眰");
+ return false;
+ }
+
+ // 妫�鏌ユ槸鍚﹀凡缁忚繛鎺�
+ if (areClientsConnected()) {
+ System.out.println("MQTT瀹㈡埛绔凡杩炴帴锛屾棤闇�閲嶅杩炴帴");
+ return true;
+ }
+
+ isReconnecting = true;
+ try {
+ // 鍏堟柇寮�涔嬪墠鐨勮繛鎺�
+ disconnectAll();
+
+ boolean gpsSuccess = false;
+ boolean responseSuccess = false;
+
+ try {
+ if (deviceId == null || deviceId.isEmpty()) {
+ System.err.println("璁惧ID涓虹┖锛屾棤娉曡繛鎺QTT");
+ return false;
+ }
+ // 娣诲姞鏃堕棿鎴崇‘淇濆鎴风ID鍞竴
+ long timestamp = System.currentTimeMillis();
+ String clientId = userEmail + "mower" + "_" + timestamp;
+ String clientId2 = userEmail + "response" + "_" + timestamp;
+ String topic = "mower/" + deviceId + "/gps";
+ String topic2 = "mower/" + deviceId + "/response";
+
+ // 杩炴帴GPS涓婚
+ try {
+ gpsClient = new Client(host, topic, clientId);
+ gpsClient.connect();
+ // 绋嶄綔寤惰繜锛岀‘淇濊繛鎺ョǔ瀹�
+ Thread.sleep(100);
+ if (gpsClient != null && gpsClient.isConnected()) {
+ gpsClient.subscribe();
+ gpsSuccess = true;
+ System.out.println("GPS涓婚MQTT杩炴帴骞惰闃呮垚鍔�");
+ }
+ } catch (MqttException e) {
+ System.err.println("GPS涓婚MQTT杩炴帴澶辫触: " + e.getMessage());
+ if (e.getCause() != null) {
+ System.err.println("澶辫触鍘熷洜: " + e.getCause().getMessage());
+ }
+ e.printStackTrace();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ System.err.println("杩炴帴杩囩▼琚腑鏂�");
+ }
+
+ // 杩炴帴鍝嶅簲涓婚
+ try {
+ responseClient = new Client(host, topic2, clientId2);
+ responseClient.connect();
+ // 绋嶄綔寤惰繜锛岀‘淇濊繛鎺ョǔ瀹�
+ Thread.sleep(100);
+ if (responseClient != null && responseClient.isConnected()) {
+ responseClient.subscribe();
+ responseSuccess = true;
+ System.out.println("鍝嶅簲涓婚MQTT杩炴帴骞惰闃呮垚鍔�");
+ }
+ } catch (MqttException e) {
+ System.err.println("鍝嶅簲涓婚MQTT杩炴帴澶辫触: " + e.getMessage());
+ if (e.getCause() != null) {
+ System.err.println("澶辫触鍘熷洜: " + e.getCause().getMessage());
+ }
+ e.printStackTrace();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ System.err.println("杩炴帴杩囩▼琚腑鏂�");
+ }
+
+ if (gpsSuccess && responseSuccess) {
+ System.out.println("鎵�鏈塎QTT涓婚杩炴帴骞惰闃呮垚鍔燂紒");
+ return true;
+ } else if (gpsSuccess || responseSuccess) {
+ System.out.println("閮ㄥ垎MQTT涓婚杩炴帴鎴愬姛");
+ return true;
+ } else {
+ System.err.println("鎵�鏈塎QTT涓婚杩炴帴澶辫触");
+ return false;
+ }
+ } catch (Exception e) {
+ System.err.println("MQTT杩炴帴杩囩▼鍙戠敓寮傚父: " + e.getMessage());
+ if (e.getCause() != null) {
+ System.err.println("寮傚父鍘熷洜: " + e.getCause().getMessage());
+ }
+ e.printStackTrace();
+ return false;
+ }
+ } finally {
+ isReconnecting = false;
+ }
+ }
+
+ /**
+ * 鍒涘缓骞惰繛鎺QTT瀹㈡埛绔殑宸ュ叿鏂规硶
+ * @param host MQTT鏈嶅姟鍣ㄥ湴鍧�
+ * @param topic 璁㈤槄涓婚
+ * @param clientId 瀹㈡埛绔疘D
+ * @param qos 鏈嶅姟璐ㄩ噺绛夌骇锛岄粯璁�2
+ * @return Client瀹炰緥锛岃繛鎺ュけ璐ヨ繑鍥瀗ull
+ */
+ public static Client createAndConnect(String host, String topic, String clientId, int qos) {
+ try {
+ Client mqttClient = new Client(host, topic, clientId);
+ mqttClient.connect();
+ mqttClient.subscribe(qos);
+ System.out.println("MQTT瀹㈡埛绔垱寤哄苟璁㈤槄鎴愬姛锛屼富棰�: " + topic + ", ClientId: " + clientId);
+ return mqttClient;
+ } catch (MqttException e) {
+ System.err.println("MQTT瀹㈡埛绔垱寤哄け璐�: " + e.getMessage() + ", 涓婚: " + topic);
+ e.printStackTrace();
+ return null;
+ }
+ }
+
+ /**
+ * 鍒涘缓骞惰繛鎺QTT瀹㈡埛绔殑宸ュ叿鏂规硶锛堥粯璁oS涓�2锛�
+ * @param host MQTT鏈嶅姟鍣ㄥ湴鍧�
+ * @param topic 璁㈤槄涓婚
+ * @param clientId 瀹㈡埛绔疘D
+ * @return Client瀹炰緥锛岃繛鎺ュけ璐ヨ繑鍥瀗ull
+ */
+ public static Client createAndConnect(String host, String topic, String clientId) {
+ return createAndConnect(host, topic, clientId, 2);
+ }
+
+ /**
+ * 鏂紑鎵�鏈塎QTT杩炴帴
+ */
+ public static void disconnectAll() {
+ try {
+ if (gpsClient != null) {
+ gpsClient.close();
+ System.out.println("GPS涓婚MQTT杩炴帴宸叉柇寮�");
+ gpsClient = null;
+ }
+ if (responseClient != null) {
+ responseClient.close();
+ System.out.println("鍝嶅簲涓婚MQTT杩炴帴宸叉柇寮�");
+ responseClient = null;
+ }
+ } catch (Exception e) {
+ System.err.println("鏂紑MQTT杩炴帴澶辫触: " + e.getMessage());
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * 鑾峰彇GPS瀹㈡埛绔疄渚�
+ * @return GPS瀹㈡埛绔疄渚�
+ */
+ public static Client getGpsClient() {
+ return gpsClient;
+ }
+
+ /**
+ * 鑾峰彇鍝嶅簲瀹㈡埛绔疄渚�
+ * @return 鍝嶅簲瀹㈡埛绔疄渚�
+ */
+ public static Client getResponseClient() {
+ return responseClient;
+ }
+
+ /**
+ * 妫�鏌QTT杩炴帴鐘舵�侊紙闈欐�佹柟娉曪級
+ * @return true琛ㄧず宸茶繛鎺ワ紝false琛ㄧず鏈繛鎺�
+ */
+ public static boolean areClientsConnected() {
+ boolean gpsConnected = gpsClient != null && gpsClient.isConnected();
+ boolean responseConnected = responseClient != null && responseClient.isConnected();
+ return gpsConnected || responseConnected;
+ }
+
+ /**
+ * 绀轰緥鐢ㄦ硶锛堜繚鐣欏悜鍚庡吋瀹癸級
+ * @deprecated 璇蜂娇鐢� connectMQTT() 鏂规硶鏇夸唬
+ */
+ @Deprecated
+ public static void lianjiemqqt() {
+ connectMQTT();
}
}
\ No newline at end of file
--
Gitblit v1.10.0