SpringBoot整合MQTT實踐開發

MQTT簡介：

MQTT（Message Queuing Telemetry Transport，消息隊列遙測傳輸協議），是一種基於發布/訂閱（publish/subscribe）模式的輕量級協議，該協議構建於TCP/IP協議之上，MQTT最大優點在於，可以以極少的代碼和有限的帶寬，為連接遠程設備提供實時可靠的消息服務（這是特點）。作為一種低開銷、低帶寬占用的即時通訊協議，使其在物聯網、小型設備、移動應用等方面有較廣泛的應用。

MQTT應用范圍：

MQTT是一個基於客戶端-服務器的消息發布/訂閱傳輸協議。MQTT協議是輕量、簡單、開放和易於實現的，這些特點使它適用范圍非常廣泛。在很多情況下，包括受限的環境中，如：機器與機器（M2M）通信和物聯網（IoT）。其在，通過衛星鏈路通信傳感器、偶爾撥號的醫療設備、智能家居、及一些小型化設備中已廣泛使用。

分享給大家一套SpringBoot的視頻教程，這套視頻由淺入深，帶你體驗Spring Boot的極速開發過程，內容豐富，涵蓋了SpringBoot開發的方方面面，並且同步更新到Spring Boot 2.x系列的最新版本，讓你一次性拿下Spring Boot開發框架。

視頻教程在線觀看：

SpringBoot框架從入門到實踐視頻課程 - Spring Boot - 動力節點在線 (bjpowernode.com)

資料及源碼免費下載：

SpringBoot學習視頻_全套SpringBoot基礎教程免費下載 - 動力節點 (bjpowernode.com)

引入maven依賴

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

1、Mqtt

@Component
public class Mqtt implements CommandLineRunner {


@Autowired
private CommitUserInfo commitUserInfol;
@Autowired
private UserInfoDao userInfoDao; // 數據庫CRUD接口
@Autowired
private ApplicationRedis applicationRedis; // redis 接口
@Autowired
JudgeEquipmentService judgeEquipmentService; // 查詢/修改---當前設備號碼狀態

@Override
public void run(String... args) throws Exception {
applicationRedis.test();
MqttConnectionUtils.start(userInfoDao,applicationRedis,judgeEquipmentService,commitUserInfol);

}



}

2.MqttConnectionUtils

private static MqttClient client;

private static MqttConnectOptions connectOptions;

private static String TOPIC;

private static String clientId;

private static final Logger LOG = LogManager.getLogger(MqttConnectionUtils.class);
static {
try {
clientId = UUID.randomUUID().toString().trim().replaceAll("-", "");
client = new MqttClient("tcp://****:1883",clientId);
connectOptions=new MqttConnectOptions();
connectOptions.setCleanSession(false);
connectOptions.setUserName("用戶名");
connectOptions.setPassword(PropertiesReader.getPassword().toCharArray());//密碼 connectOptions.setConnectionTimeout(10);
client.setTimeToWait(10000);
client.connect(connectOptions);
TOPIC = PropertiesReader.getTopic();
} catch (MqttException e) {
e.printStackTrace();
}
}


/**
* 發送數據
*/

public static void publish(String topic,String content) throws MqttException {
MqttMessage message=new MqttMessage(content.getBytes());
message.setQos(1);
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//設置日期格式
LOG.info("發送時間========"+df.format(new Date()));
LOG.info(topic+"主題發送成功，內容:"+message);
client.publish(topic,message);
}

/**
* 接收數據
*/

public static void start(UserInfoDao userInfoDao, ApplicationRedis applicationRedis,
JudgeEquipmentService judgeEquipmentService, CommitUserInfo commitUserInfol) throws MqttException {
MqttTopic topic = client.getTopic(TOPIC);
// setWill方法，如果項目中需要知道客戶端是否掉線可以調用該方法。設置最終端口的通知消息
connectOptions.setWill(topic, "close".getBytes(), 2, true);

// 訂閱消息
int[] Qos = { 1 };
String[] topic1 = { TOPIC };
client.subscribe(topic1, Qos);
// 設置回調
client.setCallback(new PushCallback(userInfoDao,applicationRedis,judgeEquipmentService,commitUserInfol));
LOG.info("WIFI版啟動成功=================");
}


/**
* mqtt重連
*/
public static void reConnect() {
while (true){
try {
if (null != client && !(client.isConnected())) {
Thread.sleep(1000);
clientId = UUID.randomUUID().toString().trim().replaceAll("-", "");
client.connect(connectOptions);
LOG.info("=======嘗試重新連接==============");
break;
}
} catch (MqttException | InterruptedException e) {
LOG.info("=======重新連接失敗:{}==============", e.toString());
continue;
}
}

}

3.PushCallback

import com.alibaba.fastjson.JSONObject;

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.springframework.beans.factory.annotation.Autowired;

import java.io.UnsupportedEncodingException;
import java.util.*;

public class PushCallback implements MqttCallback{

public static PushCallback pushCallback;

@Autowired
private UserInfoDao userInfoDao; // 數據庫CRUD接口

@Autowired
private CommitUserInfo commitUserInfol;

@Autowired
private ApplicationRedis applicationRedis; // redis 接口
@Autowired
JudgeEquipmentService judgeEquipmentService; // 查詢/修改---當前設備號碼狀態



public PushCallback(UserInfoDao userInfoDao,ApplicationRedis applicationRedis,
JudgeEquipmentService judgeEquipmentService, CommitUserInfo commitUserInfol){
this.userInfoDao = userInfoDao;
this.applicationRedis = applicationRedis;
this.judgeEquipmentService = judgeEquipmentService;
this.commitUserInfol = commitUserInfol;
}
@Override
public void connectionLost(Throwable throwable) {
// 連接丟失后，一般在這里面進行重連
System.out.println("WIFI版======連接斷開，可以做重連");
MqttConnectionUtils.reConnect();
}

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
// subscribe后得到的消息會執行到這里面
String messages = new String(message.getPayload());
if(!messages.equals("close")){
System.out.println("接收消息主題 : " + topic);
System.out.println("接收消息Qos : " + message.getQos());
System.out.println("接收消息內容 : " + new String(message.getPayload()));
try {

perform(topic,json);

}catch (Exception e){

}

}


}

@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
System.out.println("deliveryComplete---------" +iMqttDeliveryToken.isComplete());
}

public void perform(String topicP,JSONObject json) throws MqttException, UnsupportedEncodingException {
//你的業務模塊
}


}

 

 

 

---

 

轉載自：CSDN作者JAVA葉知秋

原文鏈接：https://blog.csdn.net/qq_37996327/article/details/104797737