文章目录
前言
MQTT是一种轻量级的物联网通信协议，基于客户端-服务器的消息发布/订阅传输协议，支持QoS级别，适用于低带宽、高延迟的网络环境。它具有精简的协议设计，开放的消息协议，以及广泛应用于物联网(IOT)、M2M通信、消息推送和智能设备等领域。MQTT协议涉及发布者、订阅者和消息代理（Broker）的角色，以及连接、订阅、发布消息的过程，并包含会话保持和心跳机制，确保消息的可靠传输。
MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。
以下为使用Springboot实现MQTT发布订阅消息相关代码：
一、相关pom依赖
<!-- MQTT -->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
二、相关代码
1.MQTT工具类
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;
/**
* MQTT工具类
*/
@Slf4j
@Component
public class MQTTConnect {
private String HOST = "tcp://127.0.0.1:1883"; //mqtt服务器的地址和端口号
private final String clientId = "Test";
private final String username = "test";
private final String password = "123456";
private MqttClient mqttClient;
/**
* 测试订阅消息
*/
public static void main(String[] args) throws Exception {
MQTTConnect mqttConnect = new MQTTConnect();
mqttConnect.start();
//订阅消息
mqttConnect.sub("/topic/#",0);
}
public void start() throws MqttException {
// host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
mqttClient = new MqttClient(HOST, clientId + System.currentTimeMillis(), new MemoryPersistence());
// MQTT的连接设置
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
// 设置超时时间 单位为秒
options.setConnectionTimeout(10);///默认：30
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
options.setCleanSession(false);//默认：true
// 设置断开后重新连接(设置为true时将启用自动重新连接)
options.setAutomaticReconnect(true);//默认：false
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
options.setKeepAliveInterval(20);//默认：60
// 设置回调
mqttClient.setCallback(new Callback());
mqttClient.connect(options);
}
/**
* 自定义mqtt连接
* @param host
* @param clientId
* @param userName
* @param passWord
* @param connectionTimeout
* @param cleanSession
* @param automaticReconnect
* @param keepAliveInterval
* @param mqttCallback
* @throws MqttException
*/
public void start(String host,String clientId, String userName, String passWord,
int connectionTimeout, boolean cleanSession,boolean automaticReconnect,
int keepAliveInterval,MqttCallback mqttCallback) throws MqttException {
// host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
mqttClient = new MqttClient(host, clientId + System.currentTimeMillis(), new MemoryPersistence());
// MQTT的连接设置
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(userName);
options.setPassword(passWord.toCharArray());
// 设置超时时间 单位为秒
options.setConnectionTimeout(connectionTimeout);///默认：30
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
options.setCleanSession(cleanSession);//默认：true
// 设置断开后重新连接(设置为true时将启用自动重新连接)
options.setAutomaticReconnect(automaticReconnect);//默认：false
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
options.setKeepAliveInterval(keepAliveInterval);//默认：60
// 设置回调
mqttClient.setCallback(mqttCallback);
mqttClient.connect(options);
}
/**
* 关闭MQTT连接
*/
public void close() throws MqttException {
mqttClient.disconnect();
mqttClient.close();
}
/**
* 向某个主题发布消息 默认qos：1
*
* @param topic:发布的主题
* @param msg：发布的消息
*/
public void pub(String topic, String msg) throws MqttException {
MqttMessage mqttMessage = new MqttMessage();
//mqttMessage.setQos(2);
mqttMessage.setPayload(msg.getBytes());
MqttTopic mqttTopic = mqttClient.getTopic(topic);
MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
token.waitForCompletion();
}
/**
* 向某个主题发布消息
*
* @param topic: 发布的主题
* @param msg: 发布的消息
* @param qos: 消息质量 Qos：0、1、2
*/
public void pub(String topic, String msg, int qos) throws MqttException {
MqttMessage mqttMessage = new MqttMessage();
mqttMessage.setQos(qos);
mqttMessage.setPayload(msg.getBytes());
MqttTopic mqttTopic = mqttClient.getTopic(topic);
MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
token.waitForCompletion();
}
/**
* 订阅某一个主题 ，此方法默认的的Qos等级为：1
*
* @param topic 主题
*/
public void sub(String topic) throws MqttException {
mqttClient.subscribe(topic);
}
/**
* 订阅某一个主题，可携带Qos
*
* @param topic 所要订阅的主题
* @param qos 消息质量：0、1、2
*/
public void sub(String topic, int qos) throws MqttException {
mqttClient.subscribe(topic, qos);
}
}
2.MQTT回调函数
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
/**
* 常规MQTT回调函数
* MqttCallback 接口定义了用于处理 MQTT 客户端异步事件的方法。
* 当使用 Eclipse Paho MQTT 客户端库时，你可以实现这个接口来接收连接状态的变化和消息传递的通知。
*/
@Slf4j
public class Callback implements MqttCallback {
/**
* MQTT 断开连接会执行此方法
*
* 方法说明：当客户端与 MQTT 服务器之间的连接丢失时，此方法被调用。
* @param throwable 表示导致连接丢失的原因，通常为一个 Throwable 对象
*/
@Override
public void connectionLost(Throwable throwable) {
log.info("断开了MQTT连接 ：{}", throwable.getMessage());
log.error(throwable.getMessage(), throwable);
}
/**
* publish发布成功后会执行到这里
*
* 方法说明：当一个消息的交付完成并且所有必要的确认都已收到时，此方法被调用。
* 注意事项：
* 对于 QoS 0 消息，在消息被网络层接收后会调用此方法。
* 对于 QoS 1 消息，在接收到 PUBACK 包后调用此方法。
* 对于 QoS 2 消息，在接收到 PUBCOMP 包后调用此方法。
* @param iMqttDeliveryToken the delivery token associated with the message.
*/
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("发布消息成功");
}
/**
* subscribe订阅后得到的消息会执行到这里
*
* 方法说明：当从服务器接收到一条新消息时，此方法被调用
* 注意事项：
* 在此方法中抛出任何异常将会导致客户端关闭，并且未确认的消息可能会被重新发送。
* 如果在此方法执行期间有其他消息到达，它们将被缓存直到此方法返回。
* @param topic 消息发布的主题名称
* @param message 实际的消息内容，类型为 MqttMessage
* @throws Exception
*/
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
// TODO 此处可以将订阅得到的消息进行业务处理、数据存储
log.info("收到来自 " + topic + " 的消息：{}", new String(message.getPayload()));
}
}
3.订阅消息
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
/**
* 项目启动 监听主题
*/
@Slf4j
@Component
public class MQTTListener implements ApplicationListener<ContextRefreshedEvent> {
private final MQTTConnect server;
@Autowired
public MQTTListener(MQTTConnect server) {
this.server = server;
}
@Override
public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
try {
server.start();
server.sub("/topic/#");
log.info("-----------消息订阅成功-----------");
} catch (Exception e) {
log.error(e.getMessage(), e);
log.error("-----------消息订阅失败-----------");
}
}
}
4.发布消息
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class MqttTestController {
@Autowired
private MQTTConnect mqttConnect;
@GetMapping("mqttTest01")
public void test(@RequestParam("msg") String msg,
@RequestParam("topic") String topic,@RequestParam("qos") int qos) throws Exception {
mqttConnect.pub(topic, msg,qos);
}
}
三、安装mosquitto
1.mosquitto简介
Mosquitto是用C语言实现MQTT协议的Broker。是一款实现了消息推送协议 MQTT v3.1 的开源消息代理软件，提供轻量级的，支持可发布/可订阅的的消息推送模式，使设备对设备之间的短消息通信变得简单，比如现在应用广泛的低功耗传感器，手机、嵌入式计算机、微型控制器等移动设备。一个典型的应用案例就是 Andy Stanford-ClarkMosquitto(MQTT协议创始人之一)在家中实现的远程监控和自动化。
2.下载
官网：https://mosquitto.org/download/
百度网盘：https://pan.baidu.com/s/1nOECYW3hrigcrhbQlcZU_Q?pwd=1234
本文中使用mosquitto当MQTT Broker来处理客户端发布或订阅的消息。
下载完安装一直点击next即可，需要注意的是安装完成后打开服务查看mosquitto是否启动(可能会出现端口冲突导致启动失败)
四、安装MQTT.fx
1.MQTT.fx简介
MQTT.fx是一个基于Eclipse Paho用Java编写的MQTT客户端软件。支持通过Topic订阅和发布消息，用来前期模拟设备和物理云平台等调试。
本文中使用MQTT.fx来模拟设备发送消息
2.下载
官网暂时无法打开
百度云盘：https://pan.baidu.com/s/1txsSLKHi5IqUb1HooOHT-g?pwd=1234
3.使用
下载完安装一直点击next即可，安装成功后先编辑连接信息
编辑连接
主界面
五、java订阅消息
启动java程序
/**
* 测试订阅消息
*/
public static void main(String[] args) throws Exception {
MQTTConnect mqttConnect = new MQTTConnect();
mqttConnect.start();
//订阅消息
mqttConnect.sub("/topic/#",0);
}
点击发送消息
java输出台
19:55:13.276 [MQTT Call: Test] INFO com.xxx.mqtt.Callback - 收到来自 /topic/qf 的消息：hello my name is qf
也可以点击脚步进行编辑要持续发送的消息。
JS脚本代码
var Thread = Java.type("java.lang.Thread");
function execute(action) {
out("Test Script: " + action.getName());
for (var i = 0; i < 100; i++) {
switchON();
Thread.sleep(500);
switchOFF();
Thread.sleep(500);
}
action.setExitCode(0);
action.setResultText("done.");
out("Test Script: Done");
return action;
}
function switchON() {
out("fountain ON");
mqttManager.publish("/topic/qf", "message01");
}
function switchOFF() {
out("fountain OFF");
mqttManager.publish("/topic/qf", "message02");
}
function out(message){
output.print(message);
}
本文只简单介绍使用，如果需要了解更多MQTT.fx相关内容，可以查看以下的文章：
https://blog.csdn.net/qq_33406883/article/details/107492604
六、java发布消息
启动springboot项目，通过接口发布消息，
http://127.0.0.1:8080/mqttTest01?topic=/topic/qf&msg=hello qf !&qos=0
MQTT.fx即可收到消息
参考文章：java连接MQTT服务器(Springboot整合MQTT)
标签：订阅,Java,String,param,topic,MQTT,消息,public From： https://blog.csdn.net/weixin_46425661/article/details/141347730