• 導入依賴包

<dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.0</version></dependency>

• 編寫客戶端連接配置類

package com.ha.mqtt.config.mqtt;import com.ha.common.core.utils.IdUtils; import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;@Configuration public class MqttClientConfig {@Value("${ha-config.mqtt.host}")private String host;@Value("${ha-config.mqtt.username}")private String username;@Value("${ha-config.mqtt.password}")private String password;@Value("${ha-config.mqtt.topic}")private String topic;private MqttClient client = null;private MqttAsyncClient mqttAsyncClient = null;@Beanpublic void initMqttClient() {try {String clientid = IdUtils.fastSimpleUUID();// MQTT的連接設置MqttConnectOptions options = new MqttConnectOptions();// 設置是否清空session,這里如果設置為false表示服務器會保留客戶端的連接記錄，這里設置為true表示每次連接到服務器都以新的身份連接options.setCleanSession(true);// 設置連接的用戶名options.setUserName(username);// 設置連接的密碼options.setPassword(password.toCharArray());// 設置超時時間 單位為秒options.setConnectionTimeout(10);// 設置會話心跳時間 單位為秒 服務器會每隔1.5*20秒的時間向客戶端發送個消息判斷客戶端是否在線，但這個方法并沒有重連的機制options.setKeepAliveInterval(20);//是否自動重連options.setAutomaticReconnect(true);/*//同步阻塞式客戶端// host為主機名，test為clientid即連接MQTT的客戶端ID，一般以客戶端唯一標識符表示，MemoryPersistence設置clientid的保存形式，默認為以內存保存client = new MqttClient(host, clientid, new MemoryPersistence());// 設置回調函數client.setCallback(new MqttPushClienCallback());client.connect(options);//訂閱消息if (topic.contains(";")) {client.subscribe(topic.split(";"));} else {client.subscribe(topic);}*///異步非阻塞式客戶端mqttAsyncClient = new MqttAsyncClient(host, clientid, new MemoryPersistence());mqttAsyncClient.setCallback(new MqttPushClienCallback());mqttAsyncClient.connect(options);if (topic.contains(";")) {String[] split = topic.split(";");int[] qos = new int[split.length];for (int i = 0; i < qos.length; ++i) {qos[i] = 1;}mqttAsyncClient.subscribe(split, qos);} else {mqttAsyncClient.subscribe(topic, 1);}} catch (Exception e) {e.printStackTrace();}}/*** 訂閱主題** @param topic 主題*/public void subTopic(String topic) throws MqttException {subTopic(topic, 1);}/*** 訂閱主題*/public void subTopic(String topic, int qos) throws MqttException {client.subscribe(topic, qos);}}

• 連接信息在配置文件

• 信息接收，異常重新類

package com.ha.mqtt.config.mqtt;import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;@Component public class MqttPushClienCallback implements MqttCallback {private static Logger logger = LoggerFactory.getLogger(MqttPushClienCallback.class);@Autowiredprivate MqttClientConfig mqttClientConfig;@Overridepublic void connectionLost(Throwable throwable) {try {logger.info("MQTT連接斷開，發起重連......");logger.info("............正在連接............");mqttClientConfig.initMqttClient();//心跳主題mqttClientConfig.subTopic("mqtt/face/heartbeat");//其他主題mqttClientConfig.subTopic("mqtt/face/by_ing112233/Rec");mqttClientConfig.subTopic("mqtt/face/by_ing112233/Snap");mqttClientConfig.subTopic("mqtt/face/by_ing112233/Ack");} catch (MqttException e) {e.printStackTrace();}}/*** 接收mqttf服務信息*/@Overridepublic void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {System.out.println(".......接收到mqtt信息.......");logger.info("topid：{}", topic);logger.info("mqttMessage：{}", mqttMessage.getPayload());}@Overridepublic void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {} }

• 結果輸出信息

?

?

?

總結

以上是生活随笔為你收集整理的Spring boot mqtt客户端的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。