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