springboot + rabbitmq 做智能家居,我也没想到会这么简单
本文收录在个人博客:www.chengxy-nds.top,共享技术资源,共同进步
前一段有幸参与到一个智能家居项目的开发,由于之前都没有过这方面的开发经验,所以对智能硬件的开发模式和技术栈都颇为好奇。
产品是一款可燃气体报警器,如果家中燃气泄露浓度到达一定阈值,报警器检测到并上传气体浓度值给后台,后台以电话、短信、微信等方式,提醒用户家中可能有气体泄漏。
用户还可能向报警器发一些关闭报警、调整音量的指令等。整体功能还是比较简单的,大致的逻辑如下图所示:
但当我真正的参与其中开发时,其实有一点小小的失望,因为在整个研发过程中,并没用到什么新的技术,还是常规的几种中间件,只不过换个用法而已。
技术选型用rabbitmq
来做核心的组件,主要考虑到运维成本低,组内成员使用的熟练度比较高。
下面和小伙伴分享一下如何用 springboot
+ rabbitmq
搭建物联网(IOT
)平台,其实智能硬件也没想象的那么高不可攀!
很多小伙伴可能有点懵?rabbitmq
不是消息队列吗?怎么又能做智能硬件了?
其实rabbitmq
有两种协议,我们平时接触的消息队列是用的AMQP
协议,而用在智能硬件中的是MQTT
协议。
一、什么是 MQTT协议?
MQTT
全称(Message Queue Telemetry Transport):一种基于发布/订阅(publish
/subscribe
)模式的轻量级
通讯协议,通过订阅相应的主题来获取消息,是物联网(Internet of Thing
)中的一个标准传输协议。
该协议将消息的发布者(publisher
)与订阅者(subscriber
)进行分离,因此可以在不可靠的网络环境中,为远程连接的设备提供可靠的消息服务,使用方式与传统的MQ有点类似。
TCP
协议位于传输层,MQTT
协议位于应用层,MQTT
协议构建于TCP/IP
协议上,也就是说只要支持TCP/IP
协议栈的地方,都可以使用MQTT
协议。
二、为什么要用 MQTT协议?
MQTT
协议为什么在物联网(IOT)中如此受偏爱?而不是其它协议,比如我们更为熟悉的 HTTP
协议呢?
首先
HTTP
协议它是一种同步协议,客户端请求后需要等待服务器的响应。而在物联网(IOT)环境中,设备会很受制于环境的影响,比如带宽低、网络延迟高、网络通信不稳定等,显然异步消息协议更为适合IOT
应用程序。HTTP
是单向的,如果要获取消息客户端必须发起连接,而在物联网(IOT)应用程序中,设备或传感器往往都是客户端,这意味着它们无法被动地接收来自网络的命令。通常需要将一条命令或者消息,发送到网络上的所有设备上。
HTTP
要实现这样的功能不但很困难,而且成本极高。
三、MQTT协议介绍
前边说过MQTT
是一种轻量级的协议,它只专注于发消息, 所以此协议的结构也非常简单。
MQTT数据包
在MQTT
协议中,一个MQTT
数据包由:固定头
(Fixed header)、 可变头
(Variable header)、 消息体
(payload)三部分构成。
- 固定头(Fixed header),所有数据包中都有固定头,包含数据包类型及数据包的分组标识。
- 可变头(Variable header),部分数据包类型中有可变头。
- 内容消息体(Payload),存在于部分数据包类,是客户端收到的具体消息内容。
1、固定头
固定头部,使用两个字节,共16位:
(4-7)位表示消息类型,使用4位二进制表示,可代表如下的16种消息类型,不过 0 和 15位置属于保留待用,所以共14种消息事件类型。
DUP Flag(重试标识)
DUP Flag:保证消息可靠传输,消息是否已送达的标识。默认为0,只占用一个字节,表示第一次发送,当值为1时,表示当前消息先前已经被传送过。
QoS Level(消息质量等级)
QoS Level:消息的质量等级,后边会详细介绍
RETAIN(持久化)
值为
1
:表示发送的消息需要一直持久保存,而且不受服务器重启影响,不但要发送给当前的订阅者,且以后新加入的客户端订阅了此Topic
,订阅者也会马上得到推送。
注意:新加入的订阅者,只会取出最新的一个RETAIN flag = 1
的消息推送。值为
0
:仅为当前订阅者推送此消息。
Remaining Length(剩余长度)
在当前消息中剩余的byte
(字节)数,包含可变头部和消息体payload。
2、可变头
固定头部仅定义了消息类型和一些标志位,一些消息的元数据需要放入可变头部中。可变头部内容字节长度 + 消息体payload = 剩余长度。
可变头部居于固定头部和payload中间,包含了协议名称,版本号,连接标志,用户授权,心跳时间等内容。
可变头存在于这些类型的消息:PUBLISH (QoS > 0)、PUBACK、PUBREC、PUBREL、PUBCOMP、SUBSCRIBE、SUBACK、UNSUBSCRIBE、UNSUBACK。
3、消息体payload
消息体payload只存在于CONNECT
、PUBLISH
、SUBSCRIBE
、SUBACK
、UNSUBSCRIBE
这几种类型的消息:
CONNECT
:包含客户端的ClientId
、订阅的Topic
、Message
以及用户名
和密码
。PUBLISH
:向对应主题发送消息。SUBSCRIBE
:要订阅的主题以及QoS
。SUBACK
:服务器对于SUBSCRIBE
所申请的主题及QoS
进行确认和回复。UNSUBSCRIBE
:取消要订阅的主题。
消息质量(QoS )
消息质量
(Quality of Service),即消息的发送质量,发布者(publisher
)和订阅者(subscriber
)都可以指定qos
等级,有QoS 0
、QoS 1
、QoS 2
三个等级。
下边分别说明一下这三个等级的区别。
1、Qos 0:At most once
(至多一次),只发送一次消息,不保证消息是否成功送达,没有确认机制,消息可能会丢失或重复。
2、Qos 1:At least once
(至少一次),相对于QoS 0
而言Qos 1
增加了ack
确认机制,发送者(publisher
)推送消息到MQTT代理(broker
)时,两者自身都会先持久化消息,只有当publisher
或者 Broker
分别收到 PUBACK
确认时,才会删除自身持久化的消息,否则就会重发。
但有个问题,尽管我们可以通过确认来保证一定收到客户端 或 服务器的message
,可我们却不能保证仅收到一次message
,也就是当客户端publisher
没收到Broker
的puback
或者 Broker
没有收到subscriber
的puback
,那么就会一直重发。
publisher -> broker 大致流程:
- publisher store msg -> publish ->broker (传递message)
- broker -> puback -> publisher delete msg (确认传递成功)
3、Qos 2:Exactly once
(只有一次),相对于QoS 1
,QoS 2
升级实现了仅接受一次message
,publisher
和 broker
同样对消息进行持久化,其中 publisher
缓存了message
和 对应的msgID
,而 broker
缓存了 msgID
,可以保证消息不重复,由于又增加了一个confirm
机制,整个流程变得复杂很多。
publisher -> broker 大致流程:
- publisher store msg -> publish ->broker -> broker store
- msgID(传递message) broker -> puberc (确认传递成功)
- publisher -> pubrel ->broker delete msgID (告诉broker删除msgID)
- broker -> pubcomp -> publisher delete msg (告诉publisher删除msg)
LWT(最后遗嘱)
LWT
全称为 Last Will and Testament
,其实遗嘱是一个由客户端预先定义好的主题和对应消息,附加在CONNECT
的数据包中,包括遗愿主题
、遗愿 QoS
、遗愿消息
等。
当MQTT代理 Broker
检测到有客户端client
非正常断开连接时,再由服务器主动发布此消息,然后相关的订阅者会收到消息。
举个栗子:聊天室中所有人都订阅一个叫talk
的主题 ,但小富由于网络抖动突然断开了链接,这时聊天室中所有订阅主题 talk
的客户端都会收到一个 “小富离开聊天室
” 的遗愿消息。
遗嘱的相关参数:
Will Flag
:是否使用 LWT,1 开启Will Topic
:遗愿主题名,不可使用通配符Will Qos
:发布遗愿消息时使用的 QoSWill Retain
:遗愿消息的 Retain 标识Will Message
:遗愿消息内容
那客户端Client
有哪些场景是非正常断开连接呢?
Broker
检测到底层的 I/O 异常;- 客户端 未能在心跳
Keep Alive
的间隔内和Broker
进行消息交互; - 客户端 在关闭底层
TCP
连接前没有发送DISCONNECT
数据包; - 客户端 发送错误格式的数据包到
Broker
,导致关闭和客户端的连接等。
注意:当客户端通过发布 DISCONNECT
数据包断开连接时,属于正常断开连接,并不会触发 LWT
的机制,与此同时Broker
还会丢弃掉当前客户端在连接时指定的相关 LWT
参数。
四、MQTT协议应用场景
MQTT
协议广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等领域。使用的场景也是非常非常多,下边列举一些:
- 物联网M2M通信,物联网大数据采集
- Android消息推送,WEB消息推送
- 移动即时消息,例如Facebook Messenger
- 智能硬件、智能家具、智能电器
- 车联网通信,电动车站桩采集
- 智慧城市、远程医疗、远程教育
- 电力、石油与能源等行业市场
五、代码实现
具体 rabbitmq
的环境搭建就不赘述了,网上教程比较多,有条件的用服务器,没条件的像我搞个Windows
版的也很快乐嘛。
1、启用 rabbitmq的mqtt协议
我们先开启 rabbitmq
的 mqtt
协议,因为默认安装下是关闭的,命令如下:
rabbitmq-plugins enable rabbitmq_mqtt
2、mqtt 客户端依赖包
上一步中安装rabbitmq
环境并开启 mqtt
协议后,实际上mqtt
消息代理服务就搭建好了,接下来要做的就是实现客户端消息的推送和订阅。
这里使用spring-integration-mqtt
、org.eclipse.paho.client.mqttv3
两个工具包实现。
<!--mqtt依赖包-->
<dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId>
</dependency>
<dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.0</version>
</dependency>
3、消息发送者
消息的发送比较简单,主要是应用到@ServiceActivator
注解,需要注意messageHandler.setAsync
属性,如果设置成false
,关闭异步模式发送消息时可能会阻塞。
@Configuration
public class IotMqttProducerConfig {@Autowiredprivate MqttConfig mqttConfig;@Beanpublic MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();factory.setServerURIs(mqttConfig.getServers());return factory;}@Beanpublic MessageChannel mqttOutboundChannel() {return new DirectChannel();}@Bean@ServiceActivator(inputChannel = "iotMqttInputChannel")public MessageHandler mqttOutbound() {MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getServerClientId(), mqttClientFactory());messageHandler.setAsync(false);messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());return messageHandler;}
}
MQTT
对外提供发送消息的API
时,需要使用@MessagingGateway
注解,去提供一个消息网关代理,参数defaultRequestChannel
指定发送消息绑定的channel
。
可以实现三种API
接口,payload
为发送的消息,topic
发送消息的主题,qos
消息质量。
@MessagingGateway(defaultRequestChannel = "iotMqttInputChannel")
public interface IotMqttGateway {// 向默认的 topic 发送消息void sendMessage2Mqtt(String payload);// 向指定的 topic 发送消息void sendMessage2Mqtt(String payload,@Header(MqttHeaders.TOPIC) String topic);// 向指定的 topic 发送消息,并指定服务质量参数void sendMessage2Mqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
}
4、消息订阅
消息订阅和我们平时用的MQ消息监听实现思路基本相似,@ServiceActivator
注解表明当前方法用于处理MQTT
消息,inputChannel
参数指定了用于接收消息的channel
。
/*** @Author: xiaofu* @Description: 消息订阅配置* @date 2020/6/8 18:24*/
@Configuration
public class IotMqttSubscriberConfig {@Autowiredprivate MqttConfig mqttConfig;@Beanpublic MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();factory.setServerURIs(mqttConfig.getServers());return factory;}@Beanpublic MessageChannel iotMqttInputChannel() {return new DirectChannel();}@Beanpublic MessageProducer inbound() {MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientFactory(), mqttConfig.getDefaultTopic());adapter.setCompletionTimeout(5000);adapter.setConverter(new DefaultPahoMessageConverter());adapter.setQos(1);adapter.setOutputChannel(iotMqttInputChannel());return adapter;}/*** @author xiaofu* @description 消息订阅* @date 2020/6/8 18:20*/@Bean@ServiceActivator(inputChannel = "iotMqttInputChannel")public MessageHandler handlerTest() {return message -> {try {String string = message.getPayload().toString();System.out.println("接收到消息:" + string);} catch (MessagingException ex) {//logger.info(ex.getMessage());}};}
}
六、测试消息
额~ 由于本渣渣对硬件一窍不通,为了模拟硬件的发送消息,只能借助一下工具,其实硬件端实现MQTT
协议,跟我们前边的基本没什么区别,只不过换种语言嵌入到硬件中而已。
这里选的测试工具为mqttbox
,下载地址:http://workswithweb.com/mqttbox.html
1、测试消息发送
我们用先用mqttbox
模拟向主题mqtt_test_topic
发送消息,看后台是否能成功接收到。
看到后台成功拿到了向主题mqtt_test_topic
发送的消息。
2、测试消息订阅
用mqttbox
模拟订阅主题mqtt_test_topic
,在后台向主题mqtt_test_topic
发送一条消息,这里我简单的写了个controller
调用API发送消息。
http://127.0.0.1:8080/fun/testMqtt?topic=mqtt_test_topic&message=我是后台向主题 mqtt_test_topic 发送的消息
我们看mqttbox
的订阅消息,已经成功的接收到了后台的消息,到此我们的MQTT
通信环境就算搭建成功了。如果把mqttbox
工具换成具体硬件设备,整个流程就是我们常说的智能家居了,其实真的没那么难。
七、应用注意事项
在我们实际的生产环境中遇到过的问题,这里分享一下让大家少踩坑。
clientId 要唯一
在客户端connect
连接的时,会有一个clientId
参数,需要每个客户端都保持唯一的。但我们在开发测试阶段clientId
直接在代码中写死了,而且服务都是单实例部署,并没有暴露出什么问题。
MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientFactory(), mqttConfig.getDefaultTopic());
然而在生产环境内侧的时候,由于服务是多实例集群部署,结果出现了下边的奇怪问题。同一时间内只能有一个客户端能拿到消息,其他客户端不但不能消费消息,而且还在不断的掉线重连:Lost connection: 已断开连接; retrying...
。
这就是由于clientId
相同导致客户端间相互竞争消费,最后将clientId
获取方式换成从发号器中拿,问题就好了,所以这个地方是需要特别注意的。
平时程序在开发环境没问题,可偏偏到了生产环境就一大堆问题,很多都是因为服务部署方式不同导致的。所以多学习分布式还是很有必要的。
八、其他中间件
MQTT
它只是一种协议,支持MQTT
协议的消息中间件产品非常多,下边的也只是其中的一部分
- Mosquitto
- Eclipse Paho
- RabbitMQ
- Apache ActiveMQ
- HiveMQ
- JoramMQ
- ThingMQ
- VerneMQ
- Apache Apollo
- emqttd Xively
- IBM Websphere
…
总结
我也是第一次做和硬件相关的项目,之前听到智能家居都会觉得好高大上,但实际上手开发后发现,技术嘛万变不离其宗,也只是换种用法而已。
双手奉上项目 demo 的github
地址 :https://github.com/chengxy-nds/springboot-rabbitmq-mqtt.git
感兴趣的小伙伴可以下载跑一跑,实现起来非常的简单。
原创不易,燃烧秀发输出内容,希望你能有一丢丢收获!
整理了几百本各类技术电子书,送给小伙伴们,关注公号回复【666】自行领取。和一些小伙伴们建了一个技术交流群,一起探讨技术、分享技术资料,旨在共同学习进步,如果感兴趣就加入我们吧!
springboot + rabbitmq 做智能家居,我也没想到会这么简单相关推荐
- 我也没想到 springboot + rabbitmq 做智能家居,会这么简单
前一段有幸参与到一个智能家居项目的开发,由于之前都没有过这方面的开发经验,所以对智能硬件的开发模式和技术栈都颇为好奇. 智能可燃气体报警器 产品是一款可燃气体报警器,如果家中燃气泄露浓度到达一定阈值, ...
- SpringBoot +RabbitMQ 做智能家居,居然如此简单!
前一段有幸参与到一个智能家居项目的开发,由于之前都没有过这方面的开发经验,所以对智能硬件的开发模式和技术栈都颇为好奇. 智能可燃气体报警器 产品是一款可燃气体报警器,如果家中燃气泄露浓度到达一定阈值, ...
- 华为鸿蒙全能家居,能兑现多少?华为智慧屏十年不过时,用鸿蒙理念做智能家居...
原标题:能兑现多少?华为智慧屏"十年不过时",用鸿蒙理念做智能家居 12月21日的华为全屋智能及智慧屏新品发布会上,华为正式公开全屋智能解决方案,并发布了6款全新的华为智慧屏S系列 ...
- 个人入行做智能家居,应该从哪几方面入手?
理想总是与现实相悖,在经济持续不景气的今天,豪情壮志之后便是对生活及职业的不满,大伙儿都在感叹如今钱越来越难挣,生意越来越不好做,大家都在戾气中虎视眈眈地瞄着任何一个可能会爆发的机会. 智能家居便是被 ...
- 海尔智家业绩尚可,但其智能家居“全场景”没做起来
数据智能产业创新服务媒体 --聚焦数智 · 改变商业 近日,海尔智家发布了其2022年年报,数据显示,2022年海尔智家实现收入2435.14亿元,同比增长7.2%:扣非归母净利润139.63亿元, ...
- python 网关控制家居_Python竟然可以做智能家居温湿度报警系统!赶紧给自己寝室做一个...
物联网技术开创了一个智慧城市的新时代,从智能摄像头到部署各种传感器,以此对城市各种数据进行收集,并经云端AI技术处理后,有助于提高对交通和街道等城市公共管理能力:物联网的发展也为智能家居带来了新的玩法 ...
- python能开发智能家居吗_厉害了!Python竟然可以做智能家居温湿度报警系统
*本文作者:小柠檬Emm,本文属 FreeBuf 原创奖励计划,未经许可禁止转载. 物联网技术开创了一个智慧城市的新时代,从智能摄像头到部署各种传感器,以此对城市各种数据进行收集,并经云端AI技术处理 ...
- 自己动手做智能家居之:智能空调控制
阅读本文需要 2 分钟. 夏天天气非常热的时候整晚吹空调经常会吹得整个人都不舒服,早上起来很疲惫,达不到轻松睡眠的效果.空调吹久了还是会感觉到冷,定时功能不能完全满足需求,半夜醒来开了关,关了开实在是 ...
- 智能家居项目开发(一):简单工厂设计模式引入
智能家居项目开发 一.智能家居功能细节拆分 二.设计模式概念的引入 三.类与对象 四.结构体新玩法 五.工厂模式 一.智能家居功能细节拆分 1.控制区 我们会用到之前学过的: 语音识别模块 socke ...
- python可以做特效吗_真没想到,Python还能实现5毛特效
来源 | ZackSock(ID:ZackSock) 图源 | 视觉中国 Python牛已经不是一天两天的事了,但是我开始也没想到,Python能这么牛.前段时间接触了一个批量抠图的模型库,而后在一些 ...
最新文章
- 系统安装重装与优化:chapter6:使用常用软件与电脑外设
- xp下添加linux启动,如何在windows xp系统下安装linux???
- 牛客题霸 [回文数字(palindrome-number)] C++题解/答案
- Xen虚拟机两大迁移方法详解
- 【Spring BootSpring Cloud系列】Spring Boot初识
- mysql自带客户端连接服务器,客户端连接mysql服务器的指令 mysql -u root -p 详细讲解及使用实例...
- java怎么将图片文件转流并在jsp前端显示_jsp已经被淘汰了吗?
- SpringCloud学习成长之路七 高可用配置中心
- 地点坐标拾取,经纬度精确到小数点后6位
- Linux XFS 文件系统文件的删除恢复
- 三星S5P6818 (ARM Cortex-A53架构)
- 人生无常,心安便是归处
- 5分钟使用Unity制作AR应用,结合EasyAR制作AR(转)
- lay和lied_lay和lie的区别
- mysql ssd 应用_MySQL Disk--SSD 特性
- Arithmetic Progression 题解(随机数使用)
- 【考研英语-基础-长难句分析】特殊结构_分裂结构【插入式_同位语 插入语 状从 非限定从-从句后移式】
- 移动端GPGPU 架构
- PHP打印调用堆栈信息,用于程序调试
- Lambda之Supplier
热门文章
- fastbin attack学习总结
- 【云栖大会】2016 杭州云栖大会随笔
- ie8安装java提示被阻止_IE8浏览器安装失败的解决方法
- ie10服务器运行失败,win7系统下无法安装ie10浏览器如何解决?win7系统下无法安装ie10浏览器三种解决方法...
- docker制作Nginx镜像
- Android11 亮度自动调节
- jxl 冻结单元格_如何在Word中冻结表格中单元格的大小
- Linux第一个正式版本,你安装的第一个Linux系统是什么版本的,成功没有啊?
- Riverbed发布业界最全面的数字体验管理解决方案
- 中山大学3D游戏设计读书笔记 unity3D Note9