大话 IOT 技术(3) -- MQTT篇

文章目录

- 前言
- 前情提要
- MQTT介绍
- 组成
- 万恶的app
- mqtt服务端
- 伪代码实现
- 开源的力量
- 后话

当你迷茫的时候，请点击物联网目录大纲快速查看前面的技术文章，相信你总能找到前行的方向

前言

本篇将开始讲述IOT技术的一个重点，mqtt协议。

我发现有一个物联网的准则，想分享给大家，名字我都想好了，就叫张氏定理：

手机app端用http协议与服务端（eg.bypass）进行通信
设备端用mqtt协议与mqtt服务端进行通信
服务端（eg.bypass）还要能用mqtt协议与mqtt服务端进行通信

这样，手机到设备的通信链路就算是通了起来。

http协议大家早就司空见惯了，它就像是大货车，能装的东西种类很多，本身体积也是比较庞大的，适用于手机这种要实现丰富的功能的大款。

mqtt协议是本文的重点讲述的对象，它就像小车，轻便简捷，我们通常用json作为消息体进行传递mqtt信息。

下面我们就进入正文吧。

前情提要

上篇(大话 IOT 技术(2) – 配网篇)我们讲了配网的流程，不过与本文关系不大，所以，我们还是先回顾一下经典 IOT 整体架构图

MQTT介绍

本文我还是沿用我们大话系列的精神，用生动的故事形式，来介绍mqtt技术，所以朋友们不用担心太枯燥和无聊。

先引用官方的一段介绍作为开场白，详见 https://mqtt.org/mqtt-specification/：

MQTT is an OASIS standard. The specification is managed by the OASIS MQTT Technical Committee.

MQTT（Message Queuing Telemetry Transport）是一种轻量级的消息传输协议，它被设计为易于实现，支持所有消息方向的通信，并为远程通信环境提供必要的网络优化。

前面讲到过，mqtt协议和http协议很像，它们都是物联网中的基石。

一般所说xx协议直观理解都是以xx开头，比如，http协议以http/https 开头，例如，url 为 http://test.domain.com/aaa

同理，mqtt协议也是以 mqtt/mqtts开头，例如，mqtt://test.domain.com/bbb，协议具体怎么实现不是本文关注的（偏底层），有兴趣的朋友可以自行百度。

我们主要关注mqtt 应用层，主要里面的几个关键术语，mqtt客户端，mqtt服务端（又叫broker），连接，mqtt消息，主题（topic）,发布，订阅。

因为这些和我们工作息息相关，而且已经有开源库帮我们封装好了，我们只需要会调用就能实现 IOT 的所有功能。

组成

和众多C/S架构一样，由一个 mqtt服务端（server，又叫broker），还有一个或多个 mqtt客户端（client）组成。

mqtt服务端就像是妈妈，每个客户端就像是她的孩子，孩子们有心事不会跟其他兄弟姐妹们直接说，都会向妈妈倾诉，由妈妈代为转述。

妈妈是他们的共同点，在第一篇文章中，我们也去找了手机与设备之间的共同点，只要有共同点，那就有通信的可能，这里也是一样的。

妈妈只是一个中间商，目的还是要实现各客户端之间的通信。

比如：

bypass服务 发送 mqtt消息给 mqtt服务端，消息是想查询设备1的状态（bypass —> mqtt 服务端）
设备1 从 mqtt服务端获取消息(订阅消息)，将自己的状态参数也写成一个 mqtt消息，发给 mqtt服务端（mqtt 服务端--->设备1， 设备1 ---> mqtt 服务端）
bypass服务 从 mqtt服务端获取到了设备1的消息(订阅消息)，再进行业务处理，更新在手机端显示（mqtt 服务端 ---> bypass）

这样，就完成了 bypass服务与设备1 之间的通信。

同理，bypass服务与设备2 之间的通信也是类似。

那么问题又来了，你怎么保证 bypass服务发送的 mqtt消息一定能被设备1 收到呢？设备2 会不会偷看消息？

真是个 good question，这也就是 mqtt服务端 的职责所在。

万恶的app

首先，我们来讲个题外话。

相信大家手机中肯定装了很多种app，很多万恶的app在你安装后就会流氓地要你关注订阅一些感兴趣的频道，比如，科技，财经，军事，娱乐…

而它的万恶是我都感兴趣，却不让我勾选了，最多只能选几个，这真的让人火大，那我一个也不选吧，还不行，你看这整的我只想说一句粗口：……

好吧，我忍了！

于是我手机上多了一个万恶的app，大概长这样

有没有同款的朋友们，请扣1哈

每个关注订阅的频道，都会有消息的通知，就像上面那样，用消息数量显示，我知道体育频道没有消息，我就不会点进去看，我最喜欢的娱乐频道消息满天飞，我就会点击进去浏览各种八卦……

虽然万恶，但其思想与本文要讲的 mqtt服务很像，所以我才忍你很久了，曾子曰：“我的忍耐是有限度的……"

上面不经意还是引入了一个mqtt专业术语：订阅（subscribe），我们再加一个，主题（topic）。

订阅想必大家很好理解，毕竟烂大街的订阅加关注，一键三连……

主题就是上面的兴趣频道

mqtt 中的订阅主题，就等同于所说的关注订阅某个频道，然后巴巴地等着新消息查看。比如汪苏泷的演唱会要来深圳了，什么时候开始抢票，他好帅啊……

这个万恶的app其实就是充当了服务端的身份，管理这些主题，消息动态，而我就是那个可怜巴巴没有抢到票的弱小客户端，悲伤不禁逆流成河……

我们丰富一下流程吧

服务端是那个万恶的app后端服务，客户端有某科技博主和我。

发布消息过程：某科技博主写了一篇文章，标注了科技的标签，提交后，服务端后台会处理，会在科技频道新增一篇科技的文章，并返回成功的响应。

订阅消息过程：我点击了娱乐消息想要查看，服务端接收到了我的请求，从数据库中查询出标签为娱乐的消息并返回。

mqtt服务端

类似上面的万恶app，mqtt服务端也是类似，我们来转变一下上图：

服务端变成了mqtt 服务端，客户端有bypass服务和设备。

发布消息过程：bypass 发布了一个主题为topic1 的消息，mqtt服务端处理发布消息，在topic1新增一条消息，并返回成功的响应。

订阅消息过程：设备订阅主题为topic4 的消息，mqtt服务端处理订阅请求，从主题topic4中获取其消息并发回给设备，并且客户端的 on_message方法处理消息。

一般地，主题topic可以设置为含有设备的cid的字符串，更简单可以用cid 作为主题，既能保证主题唯一，又能方便设备订阅，这样每个设备都只需要订阅与自己 cid 相关的主题。

那么，bypass 可以发送消息到 cid的主题，设备订阅自己cid 的主题，得到消息后设备再自行处理。

伪代码实现

上面我们主要关注发布和订阅的处理，因为这也是mqtt设计的核心，其他的我们可以暂时忽略，如果让我们自己来设计实现，那我们会怎么做呢？

我认为主要是服务端和客户端设计实现，首先对它们进行梳理一下

mqtt 服务端：

对客户端连接的验证（do_connect），主要校验账号密码或者证书公私钥是否正确
管理 topic 对应的消息，用队列保存/读取，先入先出
处理客户端的发布请求（do_publish）
处理客户端的订阅请求（do_subscribe）

mqtt 客户端：

连接服务端（connect），连接结果处理（on_connect）
发布消息（publish）
订阅消息（subscribe），对消息处理（on_message）

在客户端有 on_connect，on_message 是扩展方法，是留给用户去实现自定义业务逻辑。

那我们很自然地可以定义 mqtt 的两个类

下面是我自己实现的mqtt完整的代码，仅仅代表思路的验证，不可用于实际生产哦。

import queueclass MqttBroker:def __init__(self) -> None:self.topic_queue_map = {}@staticmethoddef do_connect(**args):print("broker: do_connect 处理客户端连接请求")# 做一些账密或证书公私钥的验证，这里简化一下，直接返回truereturn Truedef do_publish(self, msg: dict):print("broker: do_publish 处理客户端发布请求")topic=msg.get('topic','')q:queue.Queue=self.topic_queue_map.get(topic,queue.Queue())q.put(msg)self.topic_queue_map[topic]=qreturn Truedef do_subscribe(self, topic: str):print("broker: do_subscribe 处理客户端订阅请求")q:queue.Queue=self.topic_queue_map.get(topic,queue.Queue())return qclass MqttClient:def __init__(self) -> None:self.conn: MqttBroker = Nonedef connect(self, user_name: str, password: str):print("client: 发起 mqtt connect 请求...")res=MqttBroker.do_connect(user_name=user_name, password=password)if res:self.conn = MqttBroker()self.on_connect(res=res)def publish(self, msg: dict):print(f"client: publish 发布消息{msg}")res=self.conn.do_publish(msg=msg)print("client: publish success !") if res else print("client: publish failed")def subscribe(self, topic):print(f"client: subscribe 订阅 topic为{topic}的消息")q:queue.Queue=self.conn.do_subscribe(topic=topic)while not q.empty():self.on_message(q.get())def on_connect(self, res):r= '成功' if res else '失败'print(f"client: on_connect ,连接{r}!")def on_message(self, msg):print(f"client: on_message ,自定义处理消息: {msg}")if __name__ == '__main__':client = MqttClient()# 模拟连接print("模拟连接:")client.connect('user', 'password')# 模拟发布print("\n\n\n模拟发布:")for i in range(3):msg = {'topic': '1001', 'msg': {'playload': f'msg{i}: hello mqtt! '}}client.publish(msg=msg)# 模拟订阅print("\n\n\n模拟订阅:")client.subscribe(topic="1001")

执行结果：

模拟连接:
client: 发起 mqtt connect 请求...
broker: do_connect 处理客户端连接请求
client: on_connect ,连接成功!模拟发布:
client: publish 发布消息{'topic': '1001', 'msg': {'playload': 'msg0: hello mqtt! '}}
broker: do_publish 处理客户端发布请求
client: publish success !
client: publish 发布消息{'topic': '1001', 'msg': {'playload': 'msg1: hello mqtt! '}}
broker: do_publish 处理客户端发布请求
client: publish success !
client: publish 发布消息{'topic': '1001', 'msg': {'playload': 'msg2: hello mqtt! '}}
broker: do_publish 处理客户端发布请求
client: publish success !模拟订阅:
client: subscribe 订阅 topic为1001的消息
broker: do_subscribe 处理客户端订阅请求
client: on_message ,自定义处理消息: {'topic': '1001', 'msg': {'playload': 'msg0: hello mqtt! '}}
client: on_message ,自定义处理消息: {'topic': '1001', 'msg': {'playload': 'msg1: hello mqtt! '}}
client: on_message ,自定义处理消息: {'topic': '1001', 'msg': {'playload': 'msg2: hello mqtt! '}}