一文搞懂MQTT，如何在SpringBoot中使用MQTT实

　　之前介绍了RabbitMQ以及如何在SpringBoot项目中整合使用RabbitMQ，看过的朋友都说写的比较详细，希望再总结一下目前比较流行的MQTT。所以接下来，就来介绍什么MQTT？它在IoT中有着怎样的作用？如何在项目中使用MQTT？一、MQTT介绍1。1什么是MQTT？
　　MQTT（MessageQueuingTelemetryTransport，消息队列遥测传输协议），是一种基于发布订阅（publishsubscribe）模式的轻量级通讯协议，该协议构建于TCPIP协议上，由IBM在1999年发布。
　　MQTT最大优点在于用极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。
　　MQTT具有协议简洁、轻巧、可扩展性强、低开销、低带宽占用等优点，已经有PHP，JAVA，Python，C，C，Go等多个语言版本，基本可以使用在任何平台上。在物联网、小型设备、移动应用等方面有较广泛的应用，特别适合用来当做物联网的通信协议。
　　1。2MQTT特点
　　MQTT是一个基于客户端服务器的消息发布订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。
　　MQTT协议是为硬件性能有限，且工作在低带宽、不可靠的网络的远程传感器和控制设备通讯而设计的协议，它具有以下主要的几项特性：1。使用发布订阅消息模式，提供多对多的消息发布，解除应用程序耦合；2。对负载内容屏蔽的消息传输；3。使用TCPIP提供网络连接；4。支持三种消息发布服务质量（QoS）：QoS0（最多一次）：消息发布完全依赖底层TCPIP网络。会发生消息丢失或重复。这个级别可用于如下情况，环境传感器数据，丢失一次数据无所谓，因为不久后还会有第二次发送。QoS1（至少一次）：确保消息到达，但消息重复可能会发生。QoS2（只有一次）：确保消息到达一次。这个级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。5。传输数据小，开销很小（固定长度的头部是2字节），协议交换最小化，以降低网络流量；（用极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。）
　　1。3MQTT应用场景
　　MQTT作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有着广泛的应用。MQTT服务只负责消息的接收和传递，应用系统连接到MQTT服务器后，可以实现采集数据接收、解析、业务处理、存储入库、数据展示等功能。常见的应用场景主要有以下几个方面：
　　（1）消息推送：如PC端的推送公告，比如安卓的推送服务，还有一些即时通信软件如微信、易信等也是采用的推送技术。
　　（2）智能点餐：通过MQTT消息队列产品，消费者可在餐桌上扫码点餐，并与商家后端系统连接实现自助下单、支付。
　　（3）信息更新：实现商场超市等场所的电子标签、公共场所的多媒体屏幕的显示更新管理。
　　（4）扫码出站：最常见的停车场扫码缴费，自动起竿；地铁闸口扫码进出站。
　　二、MQTT的角色组成2。1MQTT的客户端和服务端
　　2。1。1服务端（Broker）
　　EMQX就是一个MQTT的Broker，emqx只是基于erlang语言开发的软件而已，其它的MQ还有ActiveMQ、RabbitMQ、HiveMQ等等。
　　EMQX服务端：https：www。emqx。iozhdownloads？osWindows
　　2。1。2客户端（发布订阅）
　　EMQX客户端：https：mqttx。appzh
　　这个是用来测试验证的客户端，实际项目是通过代码来实现我们消息的生产者和消费者。
　　2。2MQTT中的几个概念
　　相比RabbitMQ等消息队列，MQTT要相对简单一些，只有Broker、Topic、发布者、订阅者等几部分构成。接下来我们先简单整理下MQTT日常使用中最常见的几个概念：1。Topic主题：MQTT消息的主要传播途径，我们向主题发布消息，订阅主题，从主题中读取消息并进行。业务逻辑处理，主题是消息的通道2。生产者：MQTT消息的发送者，他们向主题发送消息3。消费者：MQTT消息的接收者，他们订阅自己需要的主题，并从中获取消息4。broker服务：消息转发器，消息是通过它来承载的，EMQX就是我们的broker，在使用中我们不用关心它的具体实现
　　其实，MQTT的使用流程就是：生产者给broker的某个topic发消息broker通过topic进行消息的传递订阅该主题的消费者拿到消息并进行相应的业务逻辑
　　三、EMQX的安装和使用
　　下面以Windows为例，演示Windows下如何安装和使用EXQX。
　　step1：下载EMQ安装包，配置EMQ环境
　　EMQX服务端：https：www。emqx。iozhdownloads？osWindows
　　step2：下载压缩包解压，cmd进入bin文件夹
　　step3：启动EMQX服务
　　在命令行输入：emqxstart启动服务，打开浏览器输入：http：localhost：18083进入登录页面。默认用户名密码adminpublic。登录成功后，会进入emqx的后台管理页面，如下图所示：
　　四、使用SpringBoot整合MQTT协议
　　前面介绍了MQTT协议以及如何安装和启动MQTT服务。接下来演示如何在SpringBoot项目中整合MQTT实现消息的订阅和发布。4。1创建工程
　　首先，创建springbootstartermqtt父工程，在父工程下分别创建消息的提供者springbootstartermqttprovider模块和消息的消费者springbootstartermqttconsumer模块。
　　4。2实现生产者
　　接下来，修改生产者模块springbootstartermqttprovider相关的代码，实现消息发布的功能模块。4。2。1导入依赖包
　　修改pom。xml文件，添加MQTT相关依赖，具体示例代码如下所示：dependenciesdependencygroupIdorg。springframework。bootgroupIdspringbootstarterartifactIddependencydependencygroupIdorg。springframework。bootgroupIdspringbootstarterwebartifactIddependencydependencygroupIdorg。springframework。bootgroupIdspringbootstartertestartifactIdscopetestscopedependencydependencygroupIdorg。springframework。integrationgroupIdspringintegrationmqttartifactIdversion5。3。2。RELEASEversiondependencydependencygroupIdorg。projectlombokgroupIdlombokartifactIdversion1。18。4versiondependencydependencies4。2。2修改配置文件
　　修改application。yml配置文件，增加MQTT相关配置。示例代码如下所示：spring：application：name：providerMQTT配置信息mqtt：MQTT服务地址，端口号默认11883，如果有多个，用逗号隔开url：tcp：127。0。0。1：11883用户名username：admin密码password：public客户端id（不能重复）client：id：provideridMQTT默认的消息推送主题，实际可在调用接口是指定default：topic：topicserver：port：80804。2。3消息生产者客户端配置
　　创建MqttProviderConfig配置类，读取application。yml中的相关配置，并初始化创建MQTT的连接。示例代码如下所示：importlombok。extern。slf4j。Slf4j；importorg。eclipse。paho。client。mqttv3。；importorg。eclipse。paho。client。mqttv3。persist。MemoryPersistence；importorg。springframework。beans。factory。annotation。Value；importorg。springframework。context。annotation。Configuration；importjavax。annotation。PostConstruct；ConfigurationSlf4jpublicclassMqttProviderConfig｛Value（｛spring。mqtt。username｝）privateStringusername；Value（｛spring。mqtt。password｝）privateStringpassword；Value（｛spring。mqtt。url｝）privateStringhostUrl；Value（｛spring。mqtt。client。id｝）privateStringclientId；Value（｛spring。mqtt。default。topic｝）privateStringdefaultTopic；客户端对象privateMqttClientclient；在bean初始化后连接到服务器PostConstructpublicvoidinit（）｛connect（）；｝客户端连接服务端publicvoidconnect（）｛try｛创建MQTT客户端对象clientnewMqttClient（hostUrl，clientId，newMemoryPersistence（））；连接设置MqttConnectOptionsoptionsnewMqttConnectOptions（）；是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos），客户端重连之后能获取到服务器在客户端断开连接期间推送的消息设置为true表示每次连接服务器都是以新的身份options。setCleanSession（true）；设置连接用户名options。setUserName（username）；设置连接密码options。setPassword（password。toCharArray（））；设置超时时间，单位为秒options。setConnectionTimeout（100）；设置心跳时间单位为秒，表示服务器每隔1。520秒的时间向客户端发送心跳判断客户端是否在线options。setKeepAliveInterval（20）；设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息options。setWill（willTopic，（clientId与服务器断开连接）。getBytes（），0，false）；设置回调client。setCallback（newMqttProviderCallBack（））；client。connect（options）；｝catch（MqttExceptione）｛e。printStackTrace（）；｝｝publicvoidpublish（intqos，booleanretained，Stringtopic，Stringmessage）｛MqttMessagemqttMessagenewMqttMessage（）；mqttMessage。setQos（qos）；mqttMessage。setRetained（retained）；mqttMessage。setPayload（message。getBytes（））；主题的目的地，用于发布订阅信息MqttTopicmqttTopicclient。getTopic（topic）；提供一种机制来跟踪消息的传递进度用于在以非阻塞方式（在后台运行）执行发布是跟踪消息的传递进度MqttDeliveryTokentoken；try｛将指定消息发布到主题，但不等待消息传递完成，返回的token可用于跟踪消息的传递状态一旦此方法干净地返回，消息就已被客户端接受发布，当连接可用，将在后台完成消息传递。tokenmqttTopic。publish（mqttMessage）；token。waitForCompletion（）；｝catch（MqttExceptione）｛e。printStackTrace（）；｝｝｝
　　4。2。4生产者客户端消息回调
　　创建MqttProviderCallBack类并继承MqttCallback，实现相关消息回调事件，示例代码如下图所示：importorg。eclipse。paho。client。mqttv3。IMqttDeliveryToken；importorg。eclipse。paho。client。mqttv3。MqttCallback；importorg。eclipse。paho。client。mqttv3。MqttMessage；publicclassMqttConsumerCallBackimplementsMqttCallback｛客户端断开连接的回调OverridepublicvoidconnectionLost（Throwablethrowable）｛System。out。println（与服务器断开连接，可重连）；｝消息到达的回调OverridepublicvoidmessageArrived（Stringtopic，MqttMessagemessage）throwsException｛System。out。println（String。format（接收消息主题：s，topic））；System。out。println（String。format（接收消息Qos：d，message。getQos（）））；System。out。println（String。format（接收消息内容：s，newString（message。getPayload（））））；System。out。println（String。format（接收消息retained：b，message。isRetained（）））；｝消息发布成功的回调OverridepublicvoiddeliveryComplete（IMqttDeliveryTokeniMqttDeliveryToken）｛System。out。println（String。format（接收消息成功））；｝｝4。2。5创建Controller控制器实现消息发布功能
　　创建SendController控制器类，实现消息的发送功能，示例代码如下所示：ControllerpublicclassSendController｛AutowiredprivateMqttProviderConfigproviderClient；RequestMapping（sendMessage）ResponseBodypublicStringsendMessage（intqos，booleanretained，Stringtopic，Stringmessage）｛try｛providerClient。publish（qos，retained，topic，message）；return发送成功；｝catch（Exceptione）｛e。printStackTrace（）；return发送失败；｝｝｝4。3实现消费者
　　前面完成了生成者消息发布的模块，接下来修改消费者模块springbootstartermqttconsumer实现消息订阅、处理的功能。4。3。1导入依赖包
　　修改pom。xml文件，添加MQTT相关依赖，具体示例代码如下所示：dependenciesdependencygroupIdorg。springframework。bootgroupIdspringbootstarterartifactIddependencydependencygroupIdorg。springframework。bootgroupIdspringbootstarterwebartifactIddependencydependencygroupIdorg。springframework。bootgroupIdspringbootstartertestartifactIdscopetestscopedependencydependencygroupIdorg。springframework。integrationgroupIdspringintegrationmqttartifactIdversion5。3。2。RELEASEversiondependencydependencygroupIdorg。projectlombokgroupIdlombokartifactIdversion1。18。4versiondependencydependencies4。3。2修改配置文件
　　修改application。yml配置文件，增加MQTT相关配置。示例代码如下所示：spring：application：name：consumerMQTT配置信息mqtt：MQTT服务端地址，端口默认为11883，如果有多个，用逗号隔开url：tcp：127。0。0。1：11883用户名username：admin密码password：public客户端id（不能重复）client：id：consumeridMQTT默认的消息推送主题，实际可在调用接口时指定default：topic：topicserver：port：80854。3。3消费者客户端配置
　　创建消费者客户端配置类MqttConsumerConfig，读取application。yml中的相关配置，并初始化创建MQTT的连接。示例代码如下所示：importjavax。annotation。PostConstruct；importorg。eclipse。paho。client。mqttv3。MqttClient；importorg。eclipse。paho。client。mqttv3。MqttConnectOptions；importorg。eclipse。paho。client。mqttv3。MqttException；importorg。eclipse。paho。client。mqttv3。persist。MemoryPersistence；importorg。springframework。beans。factory。annotation。Value；importorg。springframework。context。annotation。Configuration；ConfigurationpublicclassMqttConsumerConfig｛Value（｛spring。mqtt。username｝）privateStringusername；Value（｛spring。mqtt。password｝）privateStringpassword；Value（｛spring。mqtt。url｝）privateStringhostUrl；Value（｛spring。mqtt。client。id｝）privateStringclientId；Value（｛spring。mqtt。default。topic｝）privateStringdefaultTopic；客户端对象privateMqttClientclient；在bean初始化后连接到服务器PostConstructpublicvoidinit（）｛connect（）；｝客户端连接服务端publicvoidconnect（）｛try｛创建MQTT客户端对象clientnewMqttClient（hostUrl，clientId，newMemoryPersistence（））；连接设置MqttConnectOptionsoptionsnewMqttConnectOptions（）；是否清空session，设置为false表示服务器会保留客户端的连接记录，客户端重连之后能获取到服务器在客户端断开连接期间推送的消息设置为true表示每次连接到服务端都是以新的身份options。setCleanSession（true）；设置连接用户名options。setUserName（username）；设置连接密码options。setPassword（password。toCharArray（））；设置超时时间，单位为秒options。setConnectionTimeout（100）；设置心跳时间单位为秒，表示服务器每隔1。520秒的时间向客户端发送心跳判断客户端是否在线options。setKeepAliveInterval（20）；设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息options。setWill（willTopic，（clientId与服务器断开连接）。getBytes（），0，false）；设置回调client。setCallback（newMqttConsumerCallBack（））；client。connect（options）；订阅主题消息等级，和主题数组一一对应，服务端将按照指定等级给订阅了主题的客户端推送消息int〔〕qos｛1，1｝；主题String〔〕topics｛topic1，topic2｝；订阅主题client。subscribe（topics，qos）；｝catch（MqttExceptione）｛e。printStackTrace（）；｝｝断开连接publicvoiddisConnect（）｛try｛client。disconnect（）；｝catch（MqttExceptione）｛e。printStackTrace（）；｝｝订阅主题publicvoidsubscribe（Stringtopic，intqos）｛try｛client。subscribe（topic，qos）；｝catch（MqttExceptione）｛e。printStackTrace（）；｝｝｝4。3。4消费者客户端消息回调
　　创建MqttConsumerCallBack类并继承MqttCallback，实现相关消息回调事件，示例代码如下图所示：importorg。eclipse。paho。client。mqttv3。IMqttDeliveryToken；importorg。eclipse。paho。client。mqttv3。MqttCallback；importorg。eclipse。paho。client。mqttv3。MqttMessage；publicclassMqttConsumerCallBackimplementsMqttCallback｛客户端断开连接的回调OverridepublicvoidconnectionLost（Throwablethrowable）｛System。out。println（与服务器断开连接，可重连）；｝消息到达的回调OverridepublicvoidmessageArrived（Stringtopic，MqttMessagemessage）throwsException｛System。out。println（String。format（接收消息主题：s，topic））；System。out。println（String。format（接收消息Qos：d，message。getQos（）））；System。out。println（String。format（接收消息内容：s，newString（message。getPayload（））））；System。out。println（String。format（接收消息retained：b，message。isRetained（）））；｝消息发布成功的回调OverridepublicvoiddeliveryComplete（IMqttDeliveryTokeniMqttDeliveryToken）｛System。out。println（String。format（接收消息成功））；｝｝4。3。5创建Controller控制器，实现MQTT连接的建立和断开
　　接下来，创建Controller控制器MqttController，并实现MQTT连接的建立和断开等方法。示例代码如下所示：importcom。weiz。mqtt。config。MqttConsumerConfig；importorg。springframework。beans。factory。annotation。Autowired；importorg。springframework。beans。factory。annotation。Value；importorg。springframework。stereotype。Controller；importorg。springframework。web。bind。annotation。RequestMapping；importorg。springframework。web。bind。annotation。ResponseBody；ControllerpublicclassMqttController｛AutowiredprivateMqttConsumerConfigclient；Value（｛spring。mqtt。client。id｝）privateStringclientId；RequestMapping（connect）ResponseBodypublicStringconnect（）｛client。connect（）；returnclientId连接到服务器；｝RequestMapping（disConnect）ResponseBodypublicStringdisConnect（）｛client。disConnect（）；returnclientId与服务器断开连接；｝｝4。4测试验证
　　首先，分别启动生产者springbootstartermqttprovider和消费者springbootstartermqttconsumer两个项目，打开浏览器，输入地址http：localhost：18083，在EMQX管理界面可以看到连接上来的两个客户端。如下图所示：
　　接下来，调用生产者的消息发布接口验证消息发布是否成功。使用Pomstman调用消息发送接口：http：localhost：8080sendMessage，如下图所示：
　　通过上图可以发现，生产者模块已经把消息发送成功。接下来查看消费者模块，验证消息是否处理成功。如下图所示：
　　通过日志输出可以发现，消费者已经成功接收到生产者发送的消息，说明我们成功实现在SpringBoot项目中整合MQTT实现了消息的发布和订阅的功能。
　　最后
　　以上就是如何在SpringBoot中使用MQTT的详细内容，更多关于在SpringBoot中MQTT的使用大家可以去自己研究学习。比如：如何利用qos机制保证数据不会丢失？消息的队列和排序？集群模式下的应用？等等。