天天看点
返回

物联网网关开发：进程之间通信方式--MQTT消息总线

道哥的第 020 篇原创

目录

  • ​​一、Linux 系统中的进程之间通信(IPC)​​
  • ​二、基于 Socket 通信的优点​
  • ​​1. 跨主机，具有伸缩性​​
  • ​​2. 操作系统会自动回收资源​​
  • ​​3. 可记录、可重现​​
  • ​​4. 跨语言​​
  • ​三、MQTT 消息总线​
  • ​​1. MQTT 是一个通信的机制​​
  • ​​2. MQTT 的实现​​
  • ​​3. 在 MQTT 之上，设计自己的通信协议​​
  • ​四、嵌入式系统中如何利用 MQTT 消息总线​
  • ​​1. 一个嵌入式系统的通信框架​​
  • ​​2. 稍微复杂一点的通信模型​​
  • ​五、Mosquitto: 一个简单的测试代码​
  • ​​1. 直接通过 apt 来安装、测试​​
  • ​​2. 通过源码来手动编译、测试​​
  • ​​六、总结​​
  • ​七、资源下载​
  • ​​1. mosquitto-1.4.9.tgz​​
  • ​​2. Mosquitto Demo 示例代码​​

一、Linux 系统中的进程之间通信(IPC)

作为一名嵌入式软件开发人员来说，处理进程之间的通信是很常见的事情。从通信目的的角度来看，我们可以把进程之间的通信分成 3 种：

  1. 为了进程的调度: 可以通过信号来实现;
  2. 为了共享资源：可以通过互斥锁、信号量、读写锁、文件锁等来实现;
  3. 为了传递数据：可以通过共享内存、命名管道、消息队列、Socket来实现。

关于上面提到的这些、操作系统为我们提供的通信原语，网络上的各种资料、文章满天飞，在这里就不啰嗦了。在这些方法中应该如何选择呢？根据我个人的经验，贵精不贵多，认真挑选三四样东西就能完全满足日常的工作需要。

我们今天想讨论的问题主要是第 3 个：传递数据，在上面这几种传递数据的方法中，我最喜欢、最常用的就是 Socket 通信。

有些小伙伴可能会说：Socket 通信就是 TCP/IP 的那一套东西，还需要自己管理连接、对数据进行组包、分包，也是挺麻烦的。

没错，Socket 通信本身的确需要手动来处理这些底层的东西，但是我们可以给 Socket 穿上一层“外衣”：利用 MQTT 消息总线，在系统的各进程之间进行数据交互，下面我们就一一道来。

二、基于 Socket 通信的优点

这里我就不自己发挥了，直接引用陈硕老师的那本书《Linux 多线程服务端编程》这本书中的观点(第 65 页，3.4小节)：

1. 跨主机，具有伸缩性

反正都是多进程了，如果一台机器的处理能力不够，就能用多台主机来处理。把进程分散到同一台局域网的多台机器上，程序改改 Host:Port 配置就能继续用。相反，文章开头部分列出的那些进程之间通信方式都不能跨机器，这就限制了可扩展性。

2. 操作系统会自动回收资源

TCP port 由一个进程独占，当程序意外退出时，操作系统会自动回收资源，不会给系统留下垃圾，程序重启之后能比较容易地恢复。

3. 可记录、可重现

两个进程通过 TCP 通信，如果一个崩溃了，操作系统会关闭连接，另一个进程几乎立刻就能感受到，可以快速 failover。当然应用层的心跳是必不可少的。(补充：操作系统本身对于 TCP 连接有一个保活时间，默认是 2 个小时，而且是针对全局的。)

4. 跨语言

服务端和客户端不必使用同一种编程语言。

  1. 陈硕老师描述的是通用的 Socket 通信，因此客户端和服务端一般位于不同的物理机器上。
  2. 在嵌入式开发中，一般都是用同一种编程语言，因此，跨语言这个有点可以忽略不计了。

三、MQTT 消息总线

1. MQTT 是一个通信的机制

对物联网领域熟悉的小伙伴，对于 MQTT 消息总线一定非常熟悉，目前几大物联网云平台（亚马孙、阿里云、华为云）都提供了 MQTT 协议的接入方式。

目前，学习 MQTT 最好的文档是 IBM 的在线手册：​​https://developer.ibm.com/zh/technologies/messaging/articles/iot-mqtt-why-good-for-iot/。​​

这里，我直接把一些重点信息列出来：

  1. MQTT协议轻量、简单、开放和易于实现；
  2. MQTT 是基于发布 (Publish)/订阅 (Subscribe)范式的消息协议；
  3. MQTT 工作在 TCP/IP协议族上；
  4. 有三种消息发布服务质量；
  5. 小型传输，开销很小（固定长度的头部是 2 字节），协议交换最小化，以降低网络流量；

MQTT 消息传输需要一个中间件，称为：Broker，其实也就是一个 Server。通信模型如下：

物联网网关开发：进程之间通信方式--MQTT消息总线
  1. MQTT Broker 需要首先启动；
  2. ClientA 和 ClientB 需要连接到 Broker;
  3. ClientA 订阅主题 topic_1，ClientB 订阅主题 topic_2;
  4. ClientA 往 topic_2 这个主题发送消息，就会被 ClientB 接收到；
  5. ClientB 往 topic_1 这个主题发送消息，就会被 ClientA 接收到；

基于 topic 主题的通信方式有一个很大的好处就是解耦，一个客户端可以订阅多个 topic，任何接入到总线的其他客户端都可以往这些 topic 中发送信息(一个客户端发送消息给自己也是可以的)。

2. MQTT 的实现

MQTT 只是一个协议而已，在 IBM 的在线文档中可以看到，有很多语言都实现了 MQTT 协议，包括：C/C++、Java、Python、C#、JavaScript、Go、Objective-C等等。那么对于嵌入式开发来说，使用比较多的是这几个实现：

Mosquitto;

Paho MQTT;

wolfMQTT;

MQTTRoute。

在下面，我们会重点介绍 Mosquitto 这个开源实现的编译和使用方式，这也是我在项目中使用最多的。

3. 在 MQTT 之上，设计自己的通信协议

从上面的描述中可以看出，MQTT 消息总线就是一个通信机制，为通信主体提供了一个传递数据的通道而已。

在这个通道之上，我们可以根据实际项目的需要，发送任何格式、编码的数据。在项目中，我们最常用的就是 json 格式的纯文本，这也是各家物联网云平台所推荐的方式。如果在文本数据中需要包含二进制数据，那就转成 BASE64 编码之后再发送。

四、嵌入式系统中如何利用 MQTT 消息总线

从上面的描述中可以看到，只要在服务端运行着一个 MQTT Broker 服务，每个连接到总线的客户端都可以灵活地相互收发数据。

我们可以把这个机制应用在嵌入式应用程序的设计中：MQTT Broker 作为一个独立的服务运行在嵌入式系统本地，其他需要交互的进程，只要连接到本地的这个 Broker，就可以相互发送数据了。运行模型如下：

物联网网关开发：进程之间通信方式--MQTT消息总线

每一个进程只需要订阅一个固定的 topic（比如：自己的 client Id），那么其他进程如果想要发送数据给它，就直接发送到这个 topic 即可。

1. 一个嵌入式系统的通信框架

我之前开发过一个环境监测系统，采集大气中的 PM2.5、PM10等污染物参数，在 Contex A8 平台下开发，需要实现数据记录(数据库)、UI 监控界面等功能。

污染物的数据采样硬件模块是第三方公司提供的，我们只需要通过该模块提供的串口协议去控制采样设备、接收采样数据即可。最终设计的通信模型如下：

物联网网关开发：进程之间通信方式--MQTT消息总线
  1. UI 进程通过消息总线，发送控制指令给采样控制进程，采样控制进程接收到后通过串口发送控制指令给采样模块；
  2. 采样控制进程从串口接收采样模块发来的PM2.5等数据后，把所有的数据发送到消息总线上指定的 topic 中；
  3. UI 进程程订阅该 topic，接收到数据后，显示在屏幕上；
  4. 数据库进程也订阅该 topic，接收到数据后，把数据存储在 SQLite 数据库中；

在这个产品中，核心进程是采样控制进程，负责与采样模块的交互。通过把 UI 处理、数据库处理设计成独立的进程，降低了系统的复杂性，即使这 2 个进程崩溃了，也不会影响到核心的采样控制进程。

比如：如果 UI 进程出现错误崩溃了，会立刻重启，启动之后通过缓存信息知道此刻正在执行采样工作，于是 UI 进程立刻连接到消息总线、进入采样数据显示界面，继续接收、显示采样控制进程发出的PM2.5等数据。

这个通信模型还有另外一个有点：可扩展性。

在项目开发的后期，甲方说需要集成一个第三方的气体模块，用来采集大气中NO、SO2等参数，通信方式是 RS485。

此时扩展这个功能模块就异常简单了，直接写一个独立的气体参数进程，接入到消息总线上。这个进程通过 RS485，从第三方气体模块接收到NO、SO2等气体参数时，直接往消息总线上的某个 topic 一丢，UI进程、数据库进程订阅这个 topic，就可以立刻接收到气体相关的数据了。

此外，这个设计模型还有其他一些优点：

  1. 并行开发：每个进程可以由不同的人员并行开发，只要相互之间定义好通信协议即可；
  2. 调试方便：由于发送的数据都是 manual readable，在开发阶段，可以在 PC 机上专门写一个监控程序，接入到嵌入式系统中的 MQTT Broker 之后，这样就可以接收到所有进程发出的消息；
  3. 通信安全：在产品 release 之后，为了防止其他人偷听数据(比如 2 中的调试进程)，可以为 MQTT Broker 指定一个配置文件，只能允许本地进程(127.0.0.1)连接到消息总线上。
物联网网关开发：进程之间通信方式--MQTT消息总线

2. 稍微复杂一点的通信模型

在刚才描述的嵌入式系框架设计中，每一个进程都是运行在本地的，所有的消息也都是在系统内进行收发。那么，如果需要把数据传输到云端、或者需要从云端接收一些控制指令，又该如何设计呢？

加入一个 MQTT Bridge 桥接模块即可！也就是再增加一个进程，这个进程同时连接到云端的 MQTT Broker 和本地的 MQTT Broker，通信模型如下：

物联网网关开发：进程之间通信方式--MQTT消息总线
  1. MQTT Bridge 接收到云端发来的指令时，转发到本地的消息总线上；
  2. MQTT Bridge 接收到本地的消息时，转发到云端的消息总线上。

五、Mosquitto: 一个简单的测试代码

上面的内容主要讨论的是设计的思想，具体到代码层面，我一般使用的是 Mosquitto 这个开源的实现。

在 Linux 系统中安装、测试都非常方便，下面就简单说明一下。

1. 直接通过 apt 来安装、测试

可以参考这个文档(​​https://www.vultr.com/docs/how-to-install-mosquitto-mqtt-broker-server-on-ubuntu-16-04​​)来安装测试。

(1) 安装

sudo apt-add-repository ppa:mosquitto-dev/mosquitto-ppa
sudo apt-get update
sudo apt-get install mosquitto
sudo apt-get install mosquitto-clients

(2) 测试

mosquitto broker 在安装之后会自动启动，可以用 ​

​netstat​

​ 查看 1883 端口来确认一下。

接收端：连接到 broker 之后，订阅 "test" 这个 topic。

mosquitto_sub -t "test"

发送端：连接到 broker 之后，往 "test" 这个 topic 发送字符串 “hello”。

mosquitto_pub -m "hello" -t "test"

当发送端执行 mosquitto_pub 时，在接收端的终端窗口中，就可以接收到 “hello” 这个字符串。

2. 通过源码来手动编译、测试

通过 apt 来安装主要是用来简单的学习和测试，如果要在项目开发中使用 Mosquitto，肯定需要手动编译，得到头文件和库文件，然后复制到应用程序中使用。

(1) 手动编译、安装 Mosquitto

我的开发环境是：

  1. 编译器：gcc (Ubuntu 5.4.0-6ubuntu1~16.04.12) 5.4.0 20160609
  2. Mosquitto 版本： mosquitto-1.4.9

mosquitto-1.4.9 可以到官方网站下载，也可以从文末的网盘中下载，你也可以尝试更高的版本。

编译、安装指令：

make
make install prefix=$PWD/install

成功安装之后，可以在当前目录的 install 文件夹下看到输出文件：

物联网网关开发：进程之间通信方式--MQTT消息总线
  1. bin：mqtt 客户端程序；
  2. include：应用程序需要 include 的头文件；
  3. lib：应用程序需要链接的库文件；
  4. sbin：mqtt broker 服务程序。

在编译过程中，如果遇到一些诸如：ares.h、uuid.h 等依赖文件找不到的错误，只需要通过 apt 指令安装响应的开发包即可。

(2) 最简单的 mosquitto 客户端代码

在 mosquitto 源码中，提供了丰富的 Sample 示例。如果你不乐意去探索，可以直接下载文末的这个网盘中的 Demo 示例程序，这个程序连接到消息总线上之后，订阅 “topic_01” 这个主题。当然，你也可以修改代码去发送消息(调用：mosquitto_publish 这个函数)。

进入 c_mqtt 示例代码目录之后，可以看到已经包含了 bin、include 和 lib 目录，它们就是从上面（1）中安装目录 install 中复制过来的。

执行 ​

​make​

​ 指令之后，即可编译成功，得到可执行文件： mqtt_client。

测试过程如下：

Step1: 启动 MQTT Broker

在第 1 个终端窗口中，启动 sbin/mosquitto 这个 Broker 程序。如果你在上面测试中已经启动了一个 broker，需要先 kill 掉之前的那个 broker，因为它们默认都使用 1883 这个端口，无法共存。

Step2: 启动接收端程序 mqtt_client

在第 2 个终端窗口中，启动 ​

​mqtt_client​

​ 也就是我们的示例代码编译得到的可执行程序，它订阅的 topic 是 “topic_01”。

./mqtt_client 127.0.0.1 1883

参数 1: Broker 服务的 IP 地址，因为都是在本地系统中，所以是 127.0.0.1；

参数 2: 端口号，一般默认是1883。

Step3: 启动发送端程序 bin/mosquitto_pub

在第 3 个终端窗口中，启动 bin/mosquitto_pub，命令如下：

./mosquitto_pub -h 127.0.0.1 -p 1883 -m "hello123" -t "topic_01"

参数 -h：Broker 服务的 IP 地址，因为都是在本地系统中，所以是 127.0.0.1；

参数 -p：端口号 1883；

参数 -m：发送的消息内容；

参数 -t：发送的主题 topic。

此时，可以在第 2 个终端窗口（mqtt_client）中打印出接收到的消息。

六、总结

这篇文章主要介绍了嵌入式系统中的一个设计模式：通过消息总线来实现进程之间的通信，并介绍了 Mosquitto 这个开源实现。

在实际的项目中，还需要更加严格的权限控制，比如：在接入消息总线时提供用户名、密码、设备证书，客户端的名称必须满足指定的格式，订阅的 topic 必须符合一定的格式等等。

在下一篇文章中，我们继续讨论这个话题，给出一个更具体、更实用的 Demo 例程。

七、资源下载

1. mosquitto-1.4.9.tgz

链接:​​https://pan.baidu.com/s/1izQ3dAlGbHiHwDvKnOSfyg​​

密码:dozt

2. Mosquitto Demo 示例代码

链接:​​https://pan.baidu.com/s/1M-dU3xapNbKyk2w07MtDyw​​

密码:aup3

物联网网关开发：进程之间通信方式--MQTT消息总线

物联网网关开发：进程之间通信方式--MQTT消息总线

物联网网关开发：进程之间通信方式--MQTT消息总线
物联网网关开发：进程之间通信方式--MQTT消息总线

物联网网关开发：进程之间通信方式--MQTT消息总线

sudo apt-add-repository ppa:mosquitto-dev/mosquitto-ppa
sudo apt-get update
sudo apt-get install mosquitto
sudo apt-get install mosquitto-clients

​netstat​

 
mosquitto_sub -t "test"
      
mosquitto_pub -m "hello" -t "test"
      

make
make install prefix=$PWD/install
物联网网关开发：进程之间通信方式--MQTT消息总线

​make​

​mqtt_client​

 
./mqtt_client 127.0.0.1 1883
      
./mosquitto_pub -h 127.0.0.1 -p 1883 -m "hello123" -t "topic_01"

消息总线 数据 客户端 嵌入式系统 采样控制
上一篇: 以友盟+U-Push为例，深度解读消息推送的筛选架构解决方案应用与实践
下一篇: 微服务架构学习与思考(06)：如何构建微服务？

继续阅读