springboot集成mqtt超級(jí)詳細(xì)步驟
更新時(shí)間：2023年06月28日 16:42:11 作者：想要一百塊
這篇文章主要介紹了springboot集成mqtt超級(jí)詳細(xì)步驟，本文分步驟結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下
springboot集成MQTT步驟
1. 引入pom依賴
<!-- mqtt -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>2. application.yml
## MQTT##
mqtt:
host: tcp://192.168.10.198:1883
userName: root
passWord: 123456
qos: 1
clientId: ClientId_local #ClientId_local必須唯一 比如你已經(jīng)定了叫ABC 那你就一直叫ABC 其他地方就不要使用ABC了
timeout: 10
keepalive: 20
topic1: A/pick/warn/# #符號(hào)是代表整個(gè)warn下面的全部子主題 沒(méi)有理解的話 可以百度仔細(xì)理解一下
topic2: A/cmd/resp
topic3: ABCF
topic4: ABCHapplication.properties
## MQTT## mqtt.host=tcp://192.168.10.198:1883 mqtt.clientId=ClientId_local mqtt.username=admin mqtt.password=123456 mqtt.timeout=10 mqtt.keepalive=20 mqtt.topic1=A/pick/warn/#
3. MqttConfiguration.java
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
1. @author WXY
2. @date 2022/6/29 20:42
*/
@Configuration
public class MqttConfiguration {
private static final Logger log = LoggerFactory.getLogger(MqttConfiguration.class);
@Value("${mqtt.host}")
String host;
@Value("${mqtt.username}")
String username;
@Value("${mqtt.password}")
String password;
@Value("${mqtt.clientId}")
String clientId;
@Value("${mqtt.timeout}")
int timeOut;
@Value("${mqtt.keepalive}")
int keepAlive;
@Value("${mqtt.topic1}")
String topic1;
@Value("${mqtt.topic2}")
String topic2;
@Value("${mqtt.topic3}")
String topic3;
@Value("${mqtt.topic4}")
String topic4;
@Bean//注入spring
public MyMQTTClient myMQTTClient() {
MyMQTTClient myMQTTClient = new MyMQTTClient(host, username, password, clientId, timeOut, keepAlive);
for (int i = 0; i < 10; i++) {
try {
myMQTTClient.connect();
//不同的主題
// myMQTTClient.subscribe(topic1, 1);
// myMQTTClient.subscribe(topic2, 1);
// myMQTTClient.subscribe(topic3, 1);
// myMQTTClient.subscribe(topic4, 1);
return myMQTTClient;
} catch (MqttException e) {
log.error("MQTT connect exception,connect time = " + i);
try {
Thread.sleep(2000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
}
}
return myMQTTClient;
}
public String getTopic1() {
return topic1;
}
public void setTopic1(String topic1) {
this.topic1 = topic1;
}
public String getTopic2() {
return topic2;
}
public void setTopic2(String topic2) {
this.topic2 = topic2;
}
public String getTopic3() {
return topic3;
}
public void setTopic3(String topic3) {
this.topic3 = topic3;
}
public String getTopic4() {
return topic4;
}
public void setTopic4(String topic4) {
this.topic4 = topic4;
}
}4. MyMQTTClient.java
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
1. @author WXY
2. @date 2022/6/29 20:43
*/
public class MyMQTTClient {
private static final Logger LOGGER = LoggerFactory.getLogger(MyMQTTClient.class);
private static MqttClient client;
private String host;
private String username;
private String password;
private String clientId;
private int timeout;
private int keepalive;
public MyMQTTClient(String host, String username, String password, String clientId, int timeOut, int keepAlive) {
this.host = host;
this.username = username;
this.password = password;
this.clientId = clientId;
this.timeout = timeOut;
this.keepalive = keepAlive;
}
public static MqttClient getClient() {
return client;
}
public static void setClient(MqttClient client) {
MyMQTTClient.client = client;
}
/**
* 設(shè)置mqtt連接參數(shù)
*
* @param username
* @param password
* @param timeout
* @param keepalive
* @return
*/
public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(timeout);
options.setKeepAliveInterval(keepalive);
options.setCleanSession(true);
options.setAutomaticReconnect(true);
return options;
}
/**
* 連接mqtt服務(wù)端，得到MqttClient連接對(duì)象
*/
public void connect() throws MqttException {
if (client == null) {
client = new MqttClient(host, clientId, new MemoryPersistence());
client.setCallback(new MyMQTTCallback(MyMQTTClient.this));
}
MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);
if (!client.isConnected()) {
client.connect(mqttConnectOptions);
} else {
client.disconnect();
client.connect(mqttConnectOptions);
}
LOGGER.info("MQTT connect success");//未發(fā)生異常，則連接成功
}
/**
* 發(fā)布，默認(rèn)qos為0，非持久化
*
* @param pushMessage
* @param topic
*/
public void publish(String pushMessage, String topic) {
publish(pushMessage, topic, 0, false);
}
/**
* 發(fā)布消息
*
* @param pushMessage
* @param topic
* @param qos
* @param retained:留存
*/
public void publish(String pushMessage, String topic, int qos, boolean retained) {
MqttMessage message = new MqttMessage();
message.setPayload(pushMessage.getBytes());
message.setQos(qos);
message.setRetained(retained);
MqttTopic mqttTopic = MyMQTTClient.getClient().getTopic(topic);
if (null == mqttTopic) {
LOGGER.error("topic is not exist");
}
MqttDeliveryToken token;//Delivery:配送
synchronized (this) {//注意：這里一定要同步，否則，在多線程publish的情況下，線程會(huì)發(fā)生死鎖，分析見(jiàn)文章最后補(bǔ)充
try {
token = mqttTopic.publish(message);//也是發(fā)送到執(zhí)行隊(duì)列中，等待執(zhí)行線程執(zhí)行，將消息發(fā)送到消息中間件
token.waitForCompletion(1000L);
} catch (MqttPersistenceException e) {
e.printStackTrace();
} catch (MqttException e) {
e.printStackTrace();
}
}
}
/**
* 訂閱某個(gè)主題
*
* @param topic
* @param qos
*/
public void subscribe(String topic, int qos) {
try {
MyMQTTClient.getClient().subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 取消訂閱主題
*
* @param topic 主題名稱
*/
public void cleanTopic(String topic) {
if (client != null && client.isConnected()) {
try {
client.unsubscribe(topic);
} catch (MqttException e) {
e.printStackTrace();
}
} else {
System.out.println("取消訂閱失??！");
}
}
}5. MyMQTTCallback.java
import cn.hutool.core.util.CharsetUtil;
import com.alibaba.fastjson.JSON;
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 java.util.Map;
/**
* @author WXY
* @date 2022/6/29 20:43
*/
public class MyMQTTCallback implements MqttCallbackExtended {
//手動(dòng)注入
private MqttConfiguration mqttConfiguration = SpringUtils.getBean(MqttConfiguration.class);
private static final Logger log = LoggerFactory.getLogger(MyMQTTCallback.class);
private MyMQTTClient myMQTTClient;
public MyMQTTCallback(MyMQTTClient myMQTTClient) {
this.myMQTTClient = myMQTTClient;
}
/**
* 丟失連接，可在這里做重連
* 只會(huì)調(diào)用一次
*
* @param throwable
*/
@Override
public void connectionLost(Throwable throwable) {
log.error("mqtt connectionLost 連接斷開(kāi)，5S之后嘗試重連: {}", throwable.getMessage());
long reconnectTimes = 1;
while (true) {
try {
if (MyMQTTClient.getClient().isConnected()) {
//判斷已經(jīng)重新連接成功 需要重新訂閱主題 可以在這個(gè)if里面訂閱主題 或者 connectComplete（方法里面） 看你們自己選擇
log.warn("mqtt reconnect success end 重新連接 重新訂閱成功");
return;
}
reconnectTimes+=1;
log.warn("mqtt reconnect times = {} try again... mqtt重新連接時(shí)間 {}", reconnectTimes, reconnectTimes);
MyMQTTClient.getClient().reconnect();
} catch (MqttException e) {
log.error("mqtt斷連異常", e);
}
try {
Thread.sleep(5000);
} catch (InterruptedException e1) {
}
}
}
/**
* @param topic
* @param mqttMessage
* @throws Exception
* subscribe后得到的消息會(huì)執(zhí)行到這里面
*/
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
log.info("接收消息主題 : {}，接收消息內(nèi)容 : {}", topic, new String(mqttMessage.getPayload()));
//發(fā)布消息主題
if (topic.equals("embed/resp")){
Map maps = (Map) JSON.parse(new String(mqttMessage.getPayload(), CharsetUtil.UTF_8));
//你自己的業(yè)務(wù)接口
insertCmdResults(maps);
}
//接收?qǐng)?bào)警主題
if (topic.equals("embed/warn")){
Map maps = (Map) JSON.parse(new String(mqttMessage.getPayload(), CharsetUtil.UTF_8));
//你自己的業(yè)務(wù)接口
insertPushAlarm(maps);
}
}
/**
*連接成功后的回調(diào) 可以在這個(gè)方法執(zhí)行 訂閱主題 生成Bean的 MqttConfiguration方法中訂閱主題 出現(xiàn)bug
*重新連接后 主題也需要再次訂閱 將重新訂閱主題放在連接成功后的回調(diào) 比較合理
* @param reconnect
* @param serverURI
*/
@Override
public void connectComplete(boolean reconnect,String serverURI){
log.info("MQTT 連接成功，連接方式：{}",reconnect?"重連":"直連");
//訂閱主題
myMQTTClient.subscribe(mqttConfiguration.topic1, 1);
myMQTTClient.subscribe(mqttConfiguration.topic2, 1);
myMQTTClient.subscribe(mqttConfiguration.topic3, 1);
myMQTTClient.subscribe(mqttConfiguration.topic4, 1);
}
/**
* 消息到達(dá)后
* subscribe后，執(zhí)行的回調(diào)函數(shù)
*
* @param s
* @param mqttMessage
* @throws Exception
*/
/**
* publish后，配送完成后回調(diào)的方法
*
* @param iMqttDeliveryToken
*/
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
}
}6. MqttMsg.java
/**
* @author WXY
* @date 2022/6/29 20:44
*/
public class MqttMsg {
private String name = "";
private String content = "";
private String time = "";
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
public String getTime() {
return time;
}
public void setTime(String time) {
this.time = time;
}
@Override
public String toString() {
return "MqttMsg{" +
"name='" + name + '\'' +
", content='" + content + '\'' +
", time='" + time + '\'' +
'}';
}
}7. MqttController.java
import com.gjwl.common.core.mqtt.MqttMsg;
import com.gjwl.common.core.mqtt.MyMQTTClient;
import net.sf.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedList;
import java.util.Queue;
/**
* @author WXY
* @date 2022/6/29 20:44
*/
@RestController
@RequestMapping("/sun/mqtt")
public class MqttController {
@Autowired
private MyMQTTClient myMQTTClient;
@Value("${mqtt.topic1}")
String topic1;
@Value("${mqtt.topic2}")
String topic2;
@Value("${mqtt.topic3}")
String topic3;
@Value("${mqtt.topic4}")
String topic4;
Queue<String> msgQueue = new LinkedList<>();
int count = 1;
@PostMapping("/getMsg")
@ResponseBody
public synchronized void mqttMsg(MqttMsg mqttMsg) {
System.out.println("隊(duì)列元素?cái)?shù)量：" + msgQueue.size());
System.out.println("***************" + mqttMsg.getName() + ":" + mqttMsg.getContent() + "****************");
//時(shí)間格式化
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String time = df.format(new Date());
mqttMsg.setTime(time);
mqttMsg.setContent(mqttMsg.getContent() + "——后臺(tái)編號(hào)：" + count);
count++;
//map轉(zhuǎn)json
JSONObject json = JSONObject.fromObject(mqttMsg);
String sendMsg = json.toString();
System.out.println(sendMsg);
//隊(duì)列添加元素
boolean flag = msgQueue.offer(sendMsg);
if (flag) {
//發(fā)布消息 topic2 是你要發(fā)送到那個(gè)通道里面的主題 比如我要發(fā)送到topic2主題消息
myMQTTClient.publish(msgQueue.poll(), topic2);
System.out.println("時(shí)間戳" + System.currentTimeMillis());
}
System.out.println("隊(duì)列元素?cái)?shù)量：" + msgQueue.size());
}
}8.SpringUtils.java
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import com.gjwl.common.utils.StringUtils;
/**
* spring工具類 方便在非spring管理環(huán)境中獲取bean
*
* @author wxy
*/
@Component
public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware
{
/** Spring應(yīng)用上下文環(huán)境 */
private static ConfigurableListableBeanFactory beanFactory;
private static ApplicationContext applicationContext;
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException
{
SpringUtils.beanFactory = beanFactory;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException
{
SpringUtils.applicationContext = applicationContext;
}
/**
* 獲取對(duì)象
*
* @param name
* @return Object 一個(gè)以所給名字注冊(cè)的bean的實(shí)例
* @throws org.springframework.beans.BeansException
*
*/
@SuppressWarnings("unchecked")
public static <T> T getBean(String name) throws BeansException
{
return (T) beanFactory.getBean(name);
}
/**
* 獲取類型為requiredType的對(duì)象
*
* @param clz
* @return
* @throws org.springframework.beans.BeansException
*
*/
public static <T> T getBean(Class<T> clz) throws BeansException
{
T result = (T) beanFactory.getBean(clz);
return result;
}
/**
* 如果BeanFactory包含一個(gè)與所給名稱匹配的bean定義，則返回true
*
* @param name
* @return boolean
*/
public static boolean containsBean(String name)
{
return beanFactory.containsBean(name);
}
/**
* 判斷以給定名字注冊(cè)的bean定義是一個(gè)singleton還是一個(gè)prototype。 如果與給定名字相應(yīng)的bean定義沒(méi)有被找到，將會(huì)拋出一個(gè)異常（NoSuchBeanDefinitionException）
*
* @param name
* @return boolean
* @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
*
*/
public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException
{
return beanFactory.isSingleton(name);
}
/**
* @param name
* @return Class 注冊(cè)對(duì)象的類型
* @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
*
*/
public static Class<?> getType(String name) throws NoSuchBeanDefinitionException
{
return beanFactory.getType(name);
}
/**
* 如果給定的bean名字在bean定義中有別名，則返回這些別名
*
* @param name
* @return
* @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
*
*/
public static String[] getAliases(String name) throws NoSuchBeanDefinitionException
{
return beanFactory.getAliases(name);
}
/**
* 獲取aop代理對(duì)象
*
* @param invoker
* @return
*/
@SuppressWarnings("unchecked")
public static <T> T getAopProxy(T invoker)
{
return (T) AopContext.currentProxy();
}
/**
* 獲取當(dāng)前的環(huán)境配置，無(wú)配置返回null
*
* @return 當(dāng)前的環(huán)境配置
*/
public static String[] getActiveProfiles()
{
return applicationContext.getEnvironment().getActiveProfiles();
}
/**
* 獲取當(dāng)前的環(huán)境配置，當(dāng)有多個(gè)環(huán)境配置時(shí)，只獲取第一個(gè)
*
* @return 當(dāng)前的環(huán)境配置
*/
public static String getActiveProfile()
{
final String[] activeProfiles = getActiveProfiles();
return StringUtils.isNotEmpty(activeProfiles) ? activeProfiles[0] : null;
}
}9.測(cè)試
發(fā)送和接收 springboot后臺(tái)日志
到此這篇關(guān)于springboot集成mqtt(超級(jí)無(wú)敵詳細(xì))的文章就介紹到這了,更多相關(guān)springboot集成mqtt內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Fastjson2是Fastjson的升級(jí)版,提供了更好的性能和擴(kuò)展性,它支持多種方式解析JSON數(shù)據(jù),包括將JSON字符串轉(zhuǎn)換為Java對(duì)象、嵌套的JSON對(duì)象和數(shù)組,以及轉(zhuǎn)換成Map或List,本文介紹Java fastjson2 解析JSON用法,感興趣的朋友一起看看吧2025-02-02
因?yàn)樵谖覀儗懙某绦虍?dāng)中可能會(huì)經(jīng)常使用到synchronized關(guān)鍵字，因此JVM對(duì)synchronized做出了很多優(yōu)化，而在本篇文章當(dāng)中我們將仔細(xì)介紹JVM對(duì)synchronized的偏向鎖的細(xì)節(jié)2023-02-02
java8中的Collectors.groupingBy用法詳解
這篇文章主要介紹了java8中的Collectors.groupingBy用法詳解，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
詳解SpringBoot實(shí)現(xiàn)fastdfs防盜鏈功能的示例代碼
我們可以通過(guò)fastdfs實(shí)現(xiàn)一個(gè)分布式文件系統(tǒng)，如果fastdfs部署在外網(wǎng)，那么任何一個(gè)人知道了上傳接口，就可以實(shí)現(xiàn)文件的上傳和訪問(wèn)。那么如何阻止他人訪問(wèn)我們fastdfs服務(wù)器上的文件呢？此處就需要使用fastdfs的防盜鏈功能，本文就來(lái)講講如何實(shí)現(xiàn)這一功能2022-10-10
這篇文章主要介紹了將Sublime Text 2配置為Java的IDE的教程,包括能讓Sublime這個(gè)文本編輯器編譯和運(yùn)行Java程序等,需要的朋友可以參考下2015-07-07
這篇文章主要介紹了修改IDEA代碼左側(cè)折疊線顏色的操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-02-02
本文主要介紹了SpringBoot?內(nèi)置工具類的使用，文中通過(guò)示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-12-12
MyBatis-Plus通用枚舉自動(dòng)關(guān)聯(lián)注入的實(shí)現(xiàn)
本文主要介紹了MyBatis-Plus通用枚舉自動(dòng)關(guān)聯(lián)注入的實(shí)現(xiàn)，解決了繁瑣的配置，讓 mybatis 優(yōu)雅的使用枚舉屬性，感興趣的可以一起來(lái)了解一下2021-06-06
java.lang.OutOfMemoryError: Metaspace異常解決的方法
這篇文章主要介紹了java.lang.OutOfMemoryError: Metaspace異常解決的方法，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
mybatis使用@mapkey獲取的結(jié)果的鍵(key)為null問(wèn)題
這篇文章主要介紹了mybatis使用@mapkey獲取的結(jié)果的鍵(key)為null問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-06-06