本文主要是介绍Java MQTT实战应用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
《JavaMQTT实战应用》本文详解MQTT协议,涵盖其发布/订阅机制、低功耗高效特性、三种服务质量等级（QoS0/1/2）,以及客户端、代理、主题的核心概念,最后提供Linux部署教程、Sprin...
一、MQTT协议
MQTT（Message Queuing Telemetry Transport）是一种轻量级的发布/订阅式消息传递协议，专为物联网(IoT)和嵌入式设备设计，它简化了设备之间的通信，并优化带宽使用。
在MQTT中，消息的发送者称为“发布者”（Publisher）消息的接收者称为“订阅者”（Subscriber），而消息的中转站是“代理”（Broker）。发布者将消息发布到特定的“主题”（Topic），代理负责将消息转发给所有订阅了该主题的订阅者。这种模式解耦了消息的发送者和接收者，使得系统更加灵活和可扩展。
二、MQTT优点
- 低功耗、高效、可靠。
- 轻量级：协议设计简洁，消息头部开销小，适用于低带宽和低功耗设备。
- 支持发布/订阅模式：设备可以发布消息到主题，其他设备可以订阅对应的主题接收消息。这一模式解耦了消息生产者和消费者，简化了系统架构，提高了灵活性和可扩展性。
- 可拓展性和兼容性：MQTT允许使用不同的传输协议，包括TCP、WebSocket等。它的简单性使得它易于与其他协议和服务集成。
- 持久化会话：MQTT支持消息持久化，允许设备在断线后重新连接时恢复之前的会话状态，包括未完成的订阅和未收到的消息队列，这对于网络不稳定或经常断开的物联网环境尤为重要。
三、三种服务质量等级
- QoS = 0（最多一次）：消息最多被传递一次，可能丢失，但不会重复。此级别提供的可靠性最低，一旦消息被客户端发送出去，它不会等待任何确认，即“Fire and Forget”模式。这意味着发布者不会确认消息是否到达Broker，也不会尝试重传失败的消息）
- QoS = 1（至少一次）：消息至少被传递一次，可能会重复，但不会丢失。此级别保证消息至少被送达一次，但有可能被重复发送。在QoS 1下，Broker(消息队列服务器)会发送PUBACK确认消息给客户端，如果客户端没有收到确认，则会重发消息，直到收到确认为止。因此，虽然可以确保消息不会丢失，但也可能导致相同消息被多次接收
- QoS = 2（恰好一次）：消息保证被传递一次且仅一次，不会丢失也不会重复。这是MQTT提供的最高级别服务质量，确保每条消息只会被接收一次，提供最严格的可靠性保证。该机制通过一个复杂的四次握手过程实现，包括消息标识符的确认和释放，确保消息既不丢失也不重复
四、客户端、代理、主题
MQTT协议中编程，三个核心概念分别是客户端（Client）、代理（Broker）和主题（Topic），它们共同构成了MQTT通信的基础框架，实现了消息的发布与订阅机制。
1. 客户端（Client）：
作用：客户端可以是消息的发布者（Publisher）或订阅者（Subscriber），也可以同时具备这两种角色。发布者负责向MQTT系统中的某个主题发布消息；订阅者则订阅感兴趣的主题，以接收来自该主题的消息。客户端可以是传感器、手机应用、服务器程序等各种设备或应用。
相互关系：客户端不直接相互通信，而是通过Broker中转消息。发布者客户端向Broker发送消息，而订阅者客户端从Broker接收消息。
2. 代理（Broker）：
作用：Broker是MQTT通信的中心节点，它接收来自发布者客户端的消息，并根据消息中的主题分发给相应的订阅者客户端。Broker负责维护客户端的连接状态、存储消息（如果需要持久化）、管理主题的订阅关系等。
相互关系：Broker是客户端之间的中介，它管理着所有的消息流动。每个客户端都与Broker建立连接，无论发布还是订阅操作，都必须通过Broker来完成。
3. 主题（Topic）：
作用：主题是MQTT中消息的分类标签，类似于一个消息通道或者频道。每个消息都会关联一个主题，发布者通过指定主题来决定消息的去向，而订阅者通过订阅特定主题来接收相关消息。
相互关系：主题是连接发布者与订阅者的桥梁。发布者向特定主题发布消息，而订阅者则通过订阅这些主题来接收消息。Broker根据主题匹配规则，确保消息被正确地路由到已订阅该主题的所有客户端。主题可以是静态的字符串，也可以包含通配符（如"+“和”#”）来实现灵活的匹配规则。
五、实战应用
1. 安装部署(linux)
-- 拉取镜像
docker pull emqx/emqx:5.0.26
-- 安装容器
docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:5.0.26
2. 访问控制台
访问：ip:18083
默认的用户名密码：admin/public
3. 客户端认证
4. 创建用户
5. SpringBoot中整合
5.1 导入jar包
<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> </dependency> <dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-stream</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-configuration-processor</artifactId> <optional>true</optional> </dependency>
5.2 yml配置
mqtt: #MQTT-服务器连接地址，如果有多个，用逗号隔开 host: tcp://192.168.17.101:1883 #MQTT-连接服务器默认客户端ID，可以随便写 clientId: mqtt_test #MQTT-用户名 username: zhangsan #MQTT-密码 password: 123456 #MQTT-指定消息的推送和订阅主题 topic: test #连接超时 timeout: 100 #设置会话心跳时间 keepalive: 10
5.3 MqttConfig.Java
@Slf4j
@Configuration
@ConfigurationProperties("mqtt")
@Data
public class MqttConfig {
String host;
String clientId;
String topic;
String username;
String password;
Integer timeout;
Integer keepalive;
// MQTT客户端的配置类，可以设置mqtt服务器的账号和密码
@Bean
public MqttConnectOptions mqttConnectOptions() {
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
// 设置是否自动重连
options.setAutomaticReconnect(true);
// false 保持会话不被清理自动重连后才能收到订阅的主题消息（包括离线时发布的消息）
options.setCleanSession(true);
options.setConnectionTimeout(timeout);
options.setKeepAliveInterval(keepalive);
return options;
}
// MqttClient 类，MQTT的客户端类，可以去连接MQTT服务器
@Bean
public MqttClient mqttClient(MqttConnectOptions mqttConnectOptions) {
try {
MqttClient client = new MqttClient(host, clientId);
// 回调对象，监听消息的获取，采用的接口回调，可以获取对应订阅到的消息
client.setCallback(new MessageCallback(client, this.topic, mqttConnectOptions));
// 连接
client.connect(mqttConnectOptions());
return client;
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException("mqtt 连接异常");
}
}
}5.4 MessageCallback.java
/**
* consumer 消费者，对收到的消息进行处理
*/
//@Component
@Slf4j
public class MessageCallback implements MqttCallbackExtended {
private MqttClient client;
private String topic;
private MqttConnectOptions mqttConnectOptions;
public MessageCallback() {
}
public MessageCallback(MqttClient mqttClient, String topic, MqttConnectOptions mqttConnectOptions) {
this.client = mqttClient;
this.topic = topic;
this.mqttConnectOptions = mqttConnectOptions;
}
// 在客户端连接断开时触python发
@Override
public void connectionLost(Throwable throwable) {
if (client != null && !client.isConnected()) {
log.info("{}, 连接断开，正在reconnect....", client.getClientId());
try {
client.reconnect();
// client.connect(this.mqttConnectOptions);
} catch (MqttException e) {
e.printStackTrace();
}
} else {
log.info("未知异常，连接断开");
}
}
// 在客户端与服务器连接成功时触发
@Override
public void connectComplete(boolean b, String url) {
log.info("{} 上线了{} {}", client.getClientId(), b, url);
try {
client.subscribe(this.topic, 0);
} catch (MqttException e) {
e.printStackTrace();
}
}
// 在客户端收到订阅的消息时触发
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
log.info("接收消息主题 : " + topic);
log.info("接收消息内容 : " + new String(message.getPayload()));
String msg = new String(message.getPayload());
try {
jsONObject jsonObject = JSON.parseobject(msg);
String clientId = String.valueOf(jsonObject.get("clientid"));
if (topic.endsWith("disconnected")) {
log.info("设备{}已掉线", clientId);
} else if (topic.endsWith("connected")) {
log.info("设备{}已上线", clientId);
} else {
log.info("其他主题的消息");
}
} catch (JSONException e) {
log.error("JSON Format Parsing Exception : {}", msg);
}
}
// 在客户端发送消息至服务器成功时触发
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
log.info("deliveryComplpythonete---------" + token.isComplete());
}
}5.5 MqttUtil.java
@Component
@Slf4j
public class MqttUtil {
@Autowired(required = false)
private MqttClient client;
/**
* 订阅主题
*
* @param topic
* @param qos
*/
public void subscribe(String topic, int qos) {
try {
client.subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 订阅主题
*
* @param topic
*/
public void subscribe(String topic) {
try {
client.subscribe(topic);
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 发布消息
*
* @param qos 连接方式 0,1,2 默认0
* @param retained 是否保留最新的消息
* @param topic 订阅主题
* @param pushMessage 消息体
*/
public void publish(int qos, boolean retained, String topic, String pushMessage) {
MqttMessage message = new MqttMessage();
message.setQos(qos);
message.setRetained(retained);
message.setPayload(pushMessage.getBytes());
MqttTopic mqttTopic = client.getTopic(topic);
if (null == mqttTop编程ic) {
log.error("topic not exist");
}
MqttDeliveryToken token;
try {
// 发送消息
token = mqttTopic.publish(message);
token.waitForCompletion();
} catch (MqttPersistenceException e) {
e.printStackTrace();
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 发布消息
*
* @param topic 主题
* @param pushMessage 消息内容
*/
public void publish(String topic, String pushMessage) {
publish(0, true, topic, pushMessage);
}
}5.6 MqttController.java
@RestController
@Slf4j
public class MqttController {
@Autowired
MqttClient client;
@Autowired
MqttUtil mqttUtil;
@GetMapping("/send")
public String send() {
try {
for (int i = 0; i < 3; i++) {
mqttUtil.publish("test", "消息hello" + i);
log.info("发送成China编程功：{}", i);
Thread.sleep(1000);
}
} catch (Exception e) {
e.printStackTrace();
}
return "SUCCESS";
}
}六、MQTTX官网地址
MQTT客户端工具MQTTX下载地址 : MQTTX：全功能 MQTT 客户端工具
到此这篇关于Java MQTT实战应用的文章就介绍到这了,更多相关Java MQTT内容请搜索编程China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持China编程(www.chinasem.cn)！
这篇关于Java MQTT实战应用的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
