宁波网站建设制作公司排名,最好用的免费建站,做相同网站违法吗,建立网站实验总结MQTT#xff08;Message Queuing Telemetry Transport#xff0c;消息队列遥测传输协议#xff09;是一种轻量级的、基于发布/订阅模式的物联网通信协议。它构建于TCP/IP协议之上#xff0c;由IBM在1999年发布。MQTT的主要特点包括#xff1a;
轻量级与高效#xff1a;M…MQTTMessage Queuing Telemetry Transport消息队列遥测传输协议是一种轻量级的、基于发布/订阅模式的物联网通信协议。它构建于TCP/IP协议之上由IBM在1999年发布。MQTT的主要特点包括
轻量级与高效MQTT设计用于在带宽有限、网络不稳定的环境中工作具有较小的数据包开销和较低的带宽占用。高可靠性使用TCP协议传输确保消息传递的可靠性。发布/订阅模式支持一对多的消息发布降低应用程序之间的耦合度。广泛适用性广泛应用于物联网、智能家居、小型设备等领域特别适用于机器与机器M2M通信。
MQTT协议通过简单的发布和订阅机制实现了消息的可靠传输和分发是物联网领域中的重要通信协议之一。
1、需要了解的理论知识
1.1、MQTT 的工作原理
要了解 MQTT 的工作原理首先需要掌握以下几个概念MQTT 客户端、MQTT Broker、发布-订阅模式、主题、QoS。
MQTT 客户端
任何运行 MQTT 客户端库的应用或设备都是 MQTT 客户端。例如使用 MQTT 的即时通讯应用是客户端使用 MQTT 上报数据的各种传感器是客户端各种 MQTT 测试工具也是客户端。
MQTT Broker
MQTT Broker 是负责处理客户端请求的关键组件包括建立连接、断开连接、订阅和取消订阅等操作同时还负责消息的转发。一个高效强大的 MQTT Broker 能够轻松应对海量连接和百万级消息吞吐量从而帮助物联网服务提供商专注于业务发展快速构建可靠的 MQTT 应用。
发布-订阅模式
发布-订阅模式与客户端-服务器模式的不同之处在于它将发送消息的客户端发布者和接收消息的客户端订阅者进行了解耦。发布者和订阅者之间无需建立直接连接而是通过 MQTT Broker 来负责消息的路由和分发。
主题
MQTT 协议根据主题来转发消息。主题通过 / 来区分层级类似于 URL 路径例如test/topic
1.2、MQTT 的工作流程
在了解了 MQTT 的基本组件之后让我们来看看它的一般工作流程
客户端使用 TCP/IP 协议与 Broker 建立连接可以选择使用 TLS/SSL 加密来实现安全通信。客户端提供认证信息并指定会话类型Clean Session 或 Persistent Session。**客户端既可以向特定主题发布消息也可以订阅主题以接收消息。**当客户端发布消息时它会将消息发送给 MQTT Broker而当客户端订阅消息时它会接收与订阅主题相关的消息。MQTT Broker 接收发布的消息并将这些消息转发给订阅了对应主题的客户端。它根据 QoS 等级确保消息可靠传递并根据会话类型为断开连接的客户端存储消息。
2、代码实现
2.1、Maven依赖
MQTT协议有两个版本一个是3.x另一个是5.x。本文使用的是3.x
MQTT v3.1
!-- MQTT v3.1 --
dependencygroupIdorg.eclipse.paho/groupIdartifactIdorg.eclipse.paho.client.mqttv3/artifactIdversion1.2.5/version
/dependency!-- MQTT 5.0 --
!-- dependency--
!-- groupIdorg.eclipse.paho/groupId--
!-- artifactIdorg.eclipse.paho.mqttv5.client/artifactId--
!-- version1.2.5/version--
!-- /dependency--2.2、配置文件
broker这里使用免费的公共的服务也可以自己使用开源项目emqx搭建
# httpserver.port8091#server.servlet.context-path/hub# mqtt
mqtt.urltcp://broker.emqx.io:1883
mqtt.username
mqtt.password
mqtt.clientIdjava-mqtt-client
mqtt.defaultTopictest/topic
mqtt.cleanSessiontrue2.3、基于MQTT写个service
依赖了lombok简化代码按需导入
dependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional
/dependencyimport jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
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.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;Service
Slf4j
public class MqttService implements MqttCallback {Value(${mqtt.url})private String url;Value(${mqtt.username})private String username;Value(${mqtt.password})private String password;Value(${mqtt.clientId})private String clientId;Value(${mqtt.cleanSession})private boolean cleanSession;private MqttClient client;private int reconnectDelay 2000; // 初始重连延迟2秒private int maxReconnectAttempts 3; // 最大重连尝试次数private int reconnectAttempts 0;PostConstructpublic void connect() {try {client new MqttClient(url, clientId);MqttConnectOptions options new MqttConnectOptions();// cleanSession为 false 时表示创建一个持久会话在客户端断开连接时会话仍然保持并保存离线消息直到会话超时注销。// cleanSession为 true 时表示创建一个新的临时会话在客户端断开时会话自动销毁。// 注意持久会话恢复的前提是客户端使用固定的 Client ID 再次连接如果 Client ID 是动态的那么连接成功后将会创建一个新的持久会话。options.setCleanSession(cleanSession);// 禁用Paho的自动重连自己控制options.setAutomaticReconnect(false);if (username ! null !username.isEmpty()) {options.setUserName(username);options.setPassword(password.toCharArray());}client.setCallback(this);client.connect(options);// 订阅一个或多个主题client.subscribe(test/topic);} catch (MqttException e){log.error(---mqtt connect fail, e);}}// 实现MqttCallback的方法connectionLost, messageArrived, deliveryCompleteOverridepublic void connectionLost(Throwable cause) {log.info(---Connection lost! cause.getMessage());// 这里可以重新连接MQTT服务器reconnect();}private void reconnect() {if (reconnectAttempts maxReconnectAttempts) {reconnectAttempts;log.info(---Attempting to reconnect in reconnectDelay ms);// 使用ScheduledExecutorService方式实现延迟重连// 为简单起见使用Thread.sleeptry {Thread.sleep(reconnectDelay);} catch (InterruptedException e) {Thread.currentThread().interrupt();}reconnectDelay * 2; // 增大重连间隔connect(); // 尝试重新连接} else {log.warn(---Max reconnect attempts reached);// 可以考虑执行一些清理操作或通知操作}}Overridepublic void messageArrived(String topic, MqttMessage message) throws Exception {// 当消息到达时调用log.info(---Message arrived. Topic: topic Message: new String(message.getPayload()));// 处理消息的逻辑}Overridepublic void deliveryComplete(IMqttDeliveryToken token) {// 当消息被完全传送出去后调用log.info(---Delivery complete!);// 可以在这里处理一些发送完成后的清理工作}// 发送消息的方法public void publish(String topic, String payload) throws MqttException {MqttMessage message new MqttMessage(payload.getBytes());// QoS 0最多交付一次。可能丢失消息// QoS 1至少交付一次。可以保证收到消息但消息可能重复// QoS 2只交付一次。可以保证消息既不丢失也不重复message.setQos(2);client.publish(topic, message);log.info(---Message published {}, payload);}// 断开连接的方法PreDestroypublic void disconnect() throws MqttException {if (client ! null client.isConnected()) {client.disconnect();log.info(---Disconnected);}}
}2.4、基于MQTT写个service
写个控制器测试
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;RestController
RequestMapping(/mqtt)
public class MqttController {Autowiredprivate MqttService mqttService;RequestMapping(value /send, method {RequestMethod.GET, RequestMethod.POST})public ResponseEntityString sendMessage(RequestParam String topic, RequestParam String message) {topic test/topic;try {mqttService.publish(topic, message);return ResponseEntity.ok(Message sent successfully);} catch (MqttException e) {return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(Failed to send message: e.getMessage());}}
}浏览器输入http://localhost:8091/mqtt/send?topicmessage我来了 控制台日志如下符合预期
2024-08-10 23:13:17 [MQTT Call: java-mqtt-client] INFO cn.talktrip.mqtt.MqttService - ---Message arrived. Topic: test/topic Message: 我来了
2024-08-10 23:13:17 [http-nio-8091-exec-8] INFO cn.talktrip.mqtt.MqttService - ---Message published 我来了
2024-08-10 23:13:17 [MQTT Call: java-mqtt-client] INFO cn.talktrip.mqtt.MqttService - ---Delivery complete!参考文档https://www.emqx.com/zh/blog/the-easiest-guide-to-getting-started-with-mqtt
