做文案图片上什么网站,深圳别墅设计公司首选,备案价公示网站,凡科自助建站系统引言
本文是springboot集成mqtt的一个实战案例。 gitee代码库地址#xff1a;源码地址
一、什么是MQTT
MQTT#xff08;Message Queuing Telemetry Transport#xff0c;消息队列遥测传输协议#xff09;#xff0c;是一种基于发布/订阅#xff08;publish/subscribe源码地址
一、什么是MQTT
MQTTMessage Queuing Telemetry Transport消息队列遥测传输协议是一种基于发布/订阅publish/subscribe模式的“轻量级”通讯协议该协议构建于 TCP/IP 协议上由 IBM 于 1999 年发明。MQTT 协议的主要特征是开放、简单、轻量级和易于实现这些特征使得它适用于受约束的应用环境如 网络受限网络带宽较低且传输不可靠 终端受限协议运行在嵌入式设备上嵌入式终端的处理器、内存等是受限的 MQTT 非常适用于物联网领域如传感器与服务器的通信、传感器信息采集等。
二、发布/订阅模式
发布/订阅模式Publish/Subscribe Pattern简称Pub/Sub是一种消息通信模式在这种模式下消息的发送者发布者不会将消息直接发送给特定的接收者订阅者。而是将代表消息内容的通知事件发布到一个特定的主题或频道上而订阅了这个主题的接收者会收到所有在这个主题上发布的通知。这种模式解耦了消息的发送者和接收者使得系统更加灵活和可扩展。
主要组成部分 发布者Publisher负责生成消息并将其发布到特定的主题或频道。 订阅者Subscriber注册对特定主题的兴趣并接收该主题上的所有消息。 消息代理Message Broker作为中间件它接收来自发布者的消息并将这些消息传递给所有相关的订阅者。
优点 解耦发布者和订阅者之间不需要直接交互这降低了系统的耦合度。 灵活性可以动态添加或删除订阅者不影响其他组件。 可扩展性系统容易扩展可以轻松增加新的发布者或订阅者。
缺点 复杂性引入了额外的组件如消息代理增加了系统的复杂性和管理成本。 性能开销消息的传递需要通过中间件可能会有延迟和性能损失。
应用场景 事件驱动架构在微服务架构中不同的服务通过发布/订阅模式进行异步通信。 数据流处理如实时数据分析多个组件可以订阅数据流并进行处理。 分布式系统用于跨系统或跨服务的消息传递。
发布/订阅模式并不是 MQTT 协议特有的模式很多消息中间件都有使用发布/订阅模式有同学可能认为这就是观察者模式还真不是这两个模式很容易混淆。观察者模式只有观察者 被观察者两个角色而发布/订阅模式还有一个经纪人 Broker往更深层次的讲观察者和被观察者是松耦合的关系而发布者和订阅者则完全不存在耦合。
三、Windows下安装MQTT消息服务器
非常遗憾EMQ X Broker 在 5.4.0 版本的发行版中已不支持 windows 版本的安装包了笔者从网上找了一个最后支持版本的压缩包已上传资源。
解压后在bin文件下使用cmd执行运行命令 .\emqx console访问MQTT管理页面 http://localhost:18083/#/ 用户名密码 admin/public
如果报错缺少Erlang环境需要自行安装下该环境 浏览器访问http://localhost:18083/#输入账号密码进入会要求你修改密码可以暂时跳过 四、Windows安装MQTT消息代理客户端MQTTX
下载地址MQTTX下载地址
点击免费下载
选择64位版本 下好后点击安装启动运行界面如下 语言是英文可以在设置按钮里调成中文。这个客户端代理主要是进行消息发送的测试服务。
五、新建MQTT集成项目
随便新建了一个springboot应用用的是JDK17在pom文件中引入如下依赖 !-- MQTT --dependencygroupIdorg.eclipse.paho/groupIdartifactIdorg.eclipse.paho.client.mqttv3/artifactIdversion1.2.5/version/dependency5.1 yml配置
server:port: 8081#允许循环依赖
spring:main:allow-circular-references: truecustomer:mqtt:broker: tcp://localhost:1883clientList:#发布客户端ID- clientId: nays_service#监听主题 同时订阅多个主题 使用 - 分割开subscribeTopic: mqtt/publish#用户名userName: admin#密码password: public#接收客户端ID- clientId: receive_service#监听主题 同时订阅多个主题 使用 - 分割开subscribeTopic: mqtt/receive#用户名userName: admin#密码password: public
5.2 Mqtt配置类
package com.hulei.mqttproject.config;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;import java.util.List;/*** Mqtt配置类*/
Data
Configuration
ConfigurationProperties(prefix customer.mqtt)
public class MqttConfig {/*** mqtt broker地址*/String broker;/*** 需要创建的MQTT客户端*/ListMqttClient clientList;
}
5.3 MQTT客户端
package com.hulei.mqttproject.config;import lombok.Data;/*** MQTT客户端*/
Data
public class MqttClient {/*** 客户端ID*/private String clientId;/*** 监听主题*/private String subscribeTopic;/*** 用户名*/private String userName;/*** 密码*/private String password;
}
5.4 MQTT客户端管理类
package com.hulei.mqttproject.config;import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** MQTT客户端管理类,如果客户端非常多后续可入redis缓存*/
Slf4j
Component
public class MqttClientManager {Value(${customer.mqtt.broker})private String mqttBroker;Resourceprivate MqttCallBackContext mqttCallBackContext;/*** 存储MQTT客户端*/public static MapString, MqttClient MQTT_CLIENT_MAP new ConcurrentHashMap();public MqttClient getMqttClientById(String clientId) {return MQTT_CLIENT_MAP.get(clientId);}/*** 创建mqtt客户端** param clientId 客户端ID* param subscribeTopic 订阅主题可为空* param userName 用户名可为空* param password 密码可为空*/public void createMqttClient(String clientId, String subscribeTopic, String userName, String password) {MemoryPersistence persistence new MemoryPersistence();try {MqttClient client new MqttClient(mqttBroker, clientId, persistence);MqttConnectOptions connOpts new MqttConnectOptions();if (null ! userName !userName.isEmpty()) {connOpts.setUserName(userName);}if (null ! password !password.isEmpty()) {connOpts.setPassword(password.toCharArray());}connOpts.setCleanSession(true);if (null ! subscribeTopic !subscribeTopic.isEmpty()) {AbsMqttCallBack callBack mqttCallBackContext.getCallBack(clientId);if (null callBack) {callBack mqttCallBackContext.getCallBack(default);}callBack.setClientId(clientId);callBack.setConnectOptions(connOpts);client.setCallback(callBack);}//连接mqtt服务端brokerclient.connect(connOpts);// 订阅主题if (null ! subscribeTopic !subscribeTopic.isEmpty()) {if (subscribeTopic.contains(-))client.subscribe(subscribeTopic.split(-));else {client.subscribe(subscribeTopic);}}MQTT_CLIENT_MAP.putIfAbsent(clientId, client);} catch (MqttException e) {log.error(Create mqttClient failed!, e);}}
}
5.5 MQTT客户端创建
package com.hulei.mqttproject.config;import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.List;/*** MQTT客户端创建*/
Component
Slf4j
public class MqttClientCreate {Resourceprivate MqttClientManager mqttClientManager;Resourceprivate MqttConfig mqttConfig;/*** 创建MQTT客户端*/PostConstructpublic void createMqttClient() {ListMqttClient mqttClientList mqttConfig.getClientList();for (MqttClient mqttClient : mqttClientList) {log.info({}, mqttClient);//创建客户端客户端IDdemo回调类跟客户端ID一致mqttClientManager.createMqttClient(mqttClient.getClientId(), mqttClient.getSubscribeTopic(), mqttClient.getUserName(), mqttClient.getPassword());}}
}
5.6 MQTT回调抽象类
package com.hulei.mqttproject.config;import jakarta.annotation.Resource;
import lombok.Getter;
import lombok.Setter;
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.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;/*** MQTT回调抽象类*/
Setter
Getter
Slf4j
public abstract class AbsMqttCallBack implements MqttCallback {private String clientId;private MqttConnectOptions connectOptions;ResourceMqttClientManager mqttClientManager;/*** 失去连接操作,进行重连** param throwable 异常*/Overridepublic void connectionLost(Throwable throwable) {try {if (null ! clientId) {if (null ! connectOptions) {mqttClientManager.getMqttClientById(clientId).connect(connectOptions);} else {mqttClientManager.getMqttClientById(clientId).connect();}}} catch (Exception e) {log.error({} reconnect failed!, e.getMessage(), e);}}/*** 接收订阅消息* param topic 主题* param mqttMessage 接收消息* throws Exception 异常*/Overridepublic void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {String content new String(mqttMessage.getPayload());handleReceiveMessage(topic, content);}/*** 消息发送成功** param iMqttDeliveryToken toke*/Overridepublic void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {log.info(消息发送成功);}/*** 处理接收的消息* param topic 主题* param message 消息内容*/protected abstract void handleReceiveMessage(String topic, String message);
}
5.7 MQTT订阅回调环境类
package com.hulei.mqttproject.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** MQTT订阅回调环境类*/
Component
Slf4j
public class MqttCallBackContext {private final MapString, AbsMqttCallBack callBackMap new ConcurrentHashMap();/*** 默认构造函数** param callBackMap 回调集合*/public MqttCallBackContext(MapString, AbsMqttCallBack callBackMap) {this.callBackMap.putAll(callBackMap);}/*** 获取MQTT回调类** param clientId 客户端ID* return MQTT回调类*/public AbsMqttCallBack getCallBack(String clientId) {return this.callBackMap.get(clientId);}
}
5.8 默认回调类
package com.hulei.mqttproject.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;/*** 默认回调*/
Slf4j
Component(default)
public class DefaultMqttCallBack extends AbsMqttCallBack {/*** param topic 主题* param message 消息内容*/Overrideprotected void handleReceiveMessage(String topic, String message) {log.info(接收到主题---{}, topic);log.info(接收到消息---{}, message);// 自定义消息处理业务}
}
六、测试服务类
package com.hulei.mqttproject.controller;import com.hulei.mqttproject.config.MqttClientManager;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;RestController
Slf4j
public class SendController {Resourceprivate MqttClientManager mqttClientManager;RequestMapping(/sendMessage)public String sendMessage(String topic){try {MqttMessage mqttMessage new MqttMessage(你好.getBytes());mqttClientManager.getMqttClientById(nays_service).publish(topic,mqttMessage);return 发送成功;} catch (Exception e) {log.error(发送失败,e);return 发送失败;}}
}
七、启动springboot
启动日志可以看到mqtt消息服务器连接成功 EMQX工具显示发布客户端和接收客户端均已成功注册 使用Apifox测试下SendController中的接口mqtt/receive是yaml中接收客户端订阅的主题当然也可以往mqtt/publish主题发mqtt中消息的发布者也可以订阅主题监听某些消息。
