前一段有幸参与到一个智能家居项目的开发,由于之前都没有过这方面的开发经验,所以对智能硬件的开发模式和技术栈都颇为好奇。
智能可燃气体报警器
产品是一款可燃气体报警器,如果家中燃气泄露浓度到达一定阈值,报警器检测到并上传气体浓度值给后台,后台以电话、短信、微信等方式,提醒用户家中可能有气体泄漏。
用户还可能向报警器发一些关闭报警、调整音量的指令等。整体功能还是比较简单的,大致的逻辑如下图所示:
但当我真正的参与其中开发时,其实有一点小小的失望,因为在整个研发过程中,并没用到什么新的技术,还是常规的几种中间件,只不过换个用法而已。
技术选型用rabbitmq 来做核心的组件,主要考虑到运维成本低,组内成员使用的熟练度比较高。
下面和小伙伴分享一下如何用 springboot + rabbitmq 搭建物联网(IOT)平台,其实智能硬件也没想象的那么高不可攀!
很多小伙伴可能有点懵?rabbitmq 不是消息队列吗?怎么又能做智能硬件了?
其实rabbitmq有两种协议,我们平时接触的消息队列是用的AMQP协议,而用在智能硬件中的是MQTT协议。
MQTT 全称(Message Queue Telemetry Transport):一种基于发布/订阅(publish/subscribe)模式的轻量级通讯协议,通过订阅相应的主题来获取消息,是物联网(Internet of Thing)中的一个标准传输协议。
该协议将消息的发布者(publisher)与订阅者(subscriber)进行分离,因此可以在不可靠的网络环境中,为远程连接的设备提供可靠的消息服务,使用方式与传统的MQ有点类似。
TCP协议位于传输层,MQTT 协议位于应用层,MQTT 协议构建于TCP/IP协议上,也就是说只要支持TCP/IP协议栈的地方,都可以使用MQTT协议。
MQTT协议为什么在物联网(IOT)中如此受偏爱?而不是其它协议,比如我们更为熟悉的 HTTP协议呢?
前边说过MQTT是一种轻量级的协议,它只专注于发消息, 所以此协议的结构也非常简单。
在MQTT协议中,一个MQTT数据包由:固定头(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(持久化)
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这几种类型的消息:
消息质量(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 大致流程:
3、Qos 2:Exactly once(只有一次),相对于QoS 1,QoS 2升级实现了仅接受一次message,publisher 和 broker 同样对消息进行持久化,其中 publisher 缓存了message和 对应的msgID,而 broker 缓存了 msgID,可以保证消息不重复,由于又增加了一个confirm 机制,整个流程变得复杂很多。
publisher -> broker 大致流程:
图片源于网络,如有侵权联系删除
LWT 全称为 Last Will and Testament,其实遗嘱是一个由客户端预先定义好的主题和对应消息,附加在CONNECT的数据包中,包括遗愿主题、遗愿 QoS、遗愿消息等。
当MQTT代理 Broker 检测到有客户端client非正常断开连接时,再由服务器主动发布此消息,然后相关的订阅者会收到消息。
举个栗子:聊天室中所有人都订阅一个叫talk的主题 ,但小富由于网络抖动突然断开了链接,这时聊天室中所有订阅主题 talk的客户端都会收到一个 “小富离开聊天室” 的遗愿消息。
遗嘱的相关参数:
那客户端Client 有哪些场景是非正常断开连接呢?
注意:当客户端通过发布 DISCONNECT 数据包断开连接时,属于正常断开连接,并不会触发 LWT 的机制,与此同时Broker 还会丢弃掉当前客户端在连接时指定的相关 LWT 参数。
MQTT协议广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等领域。使用的场景也是非常非常多,下面列举一些:
具体 rabbitmq 的环境搭建就不赘述了,网上教程比较多,有条件的用服务器,没条件的像我搞个Windows版的也很快乐嘛。
在这里插入图片描述
我们先开启 rabbitmq 的 mqtt协议,因为默认安装下是关闭的,命令如下:
rabbitmq-plugins enable rabbitmq_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>
消息的发送比较简单,主要是应用到@ServiceActivator注解,需要注意messageHandler.setAsync属性,如果设置成false,关闭异步模式发送消息时可能会阻塞。
@Configuration
public class IotMqttProducerConfig {
@Autowired
private MqttConfig mqttConfig;
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setServerURIs(mqttConfig.getServers()); return factory;
} @Bean
public 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);}
消息订阅和我们平时用的MQ消息监听实现思路基本相似,@ServiceActivator注解表明当前方法用于处理MQTT消息,inputChannel 参数指定了用于接收消息的channel。
/**
* @Author: xiaofu
* @Description: 消息订阅配置
* @date 2020/6/8 18:24
*/
@Configuration
public class IotMqttSubscriberConfig {
@Autowired
private MqttConfig mqttConfig;
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setServerURIs(mqttConfig.getServers()); return factory;
} @Bean
public MessageChannel iotMqttInputChannel() {
return new DirectChannel();
} @Bean
public 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
我们用先用mqttbox模拟向主题mqtt_test_topic发送消息,看后台是否能成功接收到。
看到后台成功拿到了向主题mqtt_test_topic发送的消息。
用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工具换成具体硬件设备,整个流程就是我们常说的智能家居了,其实真的没那么难。
在我们实际的生产环境中遇到过的问题,这里分享一下让大家少踩坑。
在客户端connect连接的时,会有一个clientId 参数,需要每个客户端都保持唯一的。但我们在开发测试阶段clientId直接在代码中写死了,而且服务都是单实例部署,并没有暴露出什么问题。
MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientFactory(), mqttConfig.getDefaultTopic());
然而在生产环境内侧的时候,由于服务是多实例集群部署,结果出现了下边的奇怪问题。同一时间内只能有一个客户端能拿到消息,其他客户端不但不能消费消息,而且还在不断的掉线重连:Lost connection: 已断开连接; retrying...。
这就是由于clientId相同导致客户端间相互竞争消费,最后将clientId获取方式换成从发号器中拿,问题就好了,所以这个地方是需要特别注意的。
平时程序在开发环境没问题,可偏偏到了生产环境就一大堆问题,很多都是因为服务部署方式不同导致的。所以多学习分布式还是很有必要的。
MQTT它只是一种协议,支持MQTT协议的消息中间件产品非常多,下边的也只是其中的一部分
我也是第一次做和硬件相关的项目,之前听到智能家居都会觉得好高大上,但实际上手开发后发现,技术嘛万变不离其宗,也只是换种用法而已。
双手奉上项目 demo 的github地址 :https://github.com/chengxy-nds/springboot-rabbitmq-mqtt.git
感兴趣的小伙伴可以下载跑一跑,实现起来非常的简单。
整理了几百本各类技术电子书相送 ,嘘~,「免费」 送给小伙伴们,私信或者评论【666】自行领取。和一些小伙伴们建了一个技术交流群,一起探讨技术、分享技术资料,旨在共同学习进步。
推荐阅读:中华新金融网