搭建自己的MQTT服务器，实现设备上云(Ubuntu+EMQX)

一、EMQX介绍

这篇文章教大家在ECS云服务器上部署EMQX，搭建自己私有的MQTT服务器，配置EMQX实现设备上云，设备数据转发，存储；服务器我采用的华为云的ECS服务器，系统选择Ubuntu系统。

Windows版本的看这里：
https://blog.csdn.net/xiaolong1126626497/article/details/134280836

1.1 EMQX介绍

EMQX是一款大规模可弹性伸缩的云原生分布式物联网MQTT消息服务器。目前EMQX提供了：开源版和企业版。

作为全球最具扩展性的 MQTT 消息服务器，EMQX 提供了高效可靠海量物联网设备连接，能够高性能实时移动与处理消息和事件流数据，可以快速构建关键业务的物联网平台与应用。

开源版的特点： 大规模可弹性伸缩的云原生分布式物联网 MQTT 消息服务器，高效可靠连接海量物联网设备，高性能实时处理消息与事件流数据，助力构建关键业务的物联网平台与应用。

• 基于 APL 2.0 开放源码协议
• 完整 MQTT 3.1.0、3.1.1 和 5.0 规范，支持 MQTT-SN
• Masterless 高可用集群架构
• 高并发、低时延、高性能
• 可扩展的网关和插件体系

帮助文档：https://www.emqx.io/docs/zh/v5.2/getting-started/getting-started.html

EMQX 支持多种安装方式，比如容器化部署，通过 EMQX Kubernetes Operator 安装部署、或通过安装包的形式部署在物理服务器或虚拟机上，针对安装包部署形式，目前支持以下操作系统：

RedHat
CentOS
RockyLinux
AmazonLinux
Ubuntu
Debian
macOS
Linux
Windows

1.2 MQTT核心概念

MQTT（消息队列遥测传输）是物联网（Internet of Things）中最常用的轻量级消息协议。该协议基于发布/订阅（pub/sub）模式进行消息通信，允许设备和应用程序实时交换数据，使用简单有效的消息格式，最大限度地减少网络开销并降低功耗。

作为MQTT消息平台，EMQX Enterprise为完整的MQTT消息功能提供全面支持。

1.3 发布/订阅模式

该协议是事件驱动的，并使用发布/订阅模式连接设备。与传统的客户端/服务器模式不同，它是一种消息传递模式，其中发送方（发布者）不直接将消息发送给特定的接收方（订阅者）。相反，发布者将消息分类为不同的主题，订阅者订阅他们感兴趣的主题。当发布者向主题发送消息时，MQTT代理路由并过滤所有传入消息，然后将其传递给对该主题表达兴趣的所有订阅者。

发布者和订阅者彼此解耦，不需要知道对方的存在。他们之间的唯一连接基于预定的消息协议。发布/订阅模式支持灵活的消息通信，因为可以根据需要动态添加或删除订阅者和发布者。它还使消息广播、多播和单播的实现更加容易。

1.4 MQTT服务器

MQTT服务器充当发布客户端和订阅客户端之间的代理，将所有收到的消息转发给匹配的订阅客户端。因此，有时服务器被直接称为MQTT代理。

1.5 MQTT客户端

客户端是指可以使用MQTT协议连接到MQTT服务器的设备或应用程序。它们既可以充当发布者和订阅者，也可以单独担任这两个角色中的任何一个。在MQTT通信中，客户端可以向服务器发布消息，同时也可以订阅一个或多个主题以接收来自其他客户端的消息。通过这种方式，MQTT客户端能够在物联网环境中实现高效、实时的数据传输和交互。

1.6 主题与通配符

在MQTT中，主题是用于识别和区分不同消息的，构成了MQTT消息路由的基础。发布者在发布消息时可以指定消息的主题，而订阅者可以选择订阅感兴趣的主题以接收相关消息。

为了实现一次订阅多个主题的目标，订阅者在订阅主题时可以使用通配符。MQTT提供了两种类型的主题通配符，单级通配符和多级通配符，以满足不同的订阅需求。

• 单级通配符：用于匹配主题中的一个级别。
• 多级通配符：用于匹配主题中的多个级别。

通过使用这两种通配符，订阅者可以灵活地订阅一系列相关的主题，而不仅仅是一个具体的主题。

1.7 服务质量（QoS）

MQTT定义了三个级别的QoS，以提供不同级别的消息可靠性。每个消息在发布时可以独立设置自己的QoS。

• QoS 0：最多传递一次消息，可能会丢失；
• QoS 1：至少传递一次消息，保证到达，但可能会重复；
• QoS 2：精确传递一次消息，保证到达且不重复。

随着QoS级别的增加，消息传输的复杂性也增加。需要根据实际情况选择合适的QoS级别。

在选择QoS级别时，需权衡消息传递的可靠性和网络资源的消耗。例如，在网络资源有限但对消息可靠性要求不高的情况下，可以选择QoS 0；而在对消息可靠性要求较高，且能容忍一定网络资源消耗的场景中，可以选择QoS 1或QoS 2。

二、购买ECS云服务器

2.1 登录官网

https://www.huaweicloud.com/

2.2 购买ECS服务器

【1】选择ECS弹性服务器

【2】选择ECS服务器的区域、配置信息、操作系统（我选择的Ubuntu18.04 64位）。

【3】购买弹性公网IP，配置带宽。

【4】配置密码

【5】选择购买时长，我这里选择了1个月时长

【6】确认付费付款

收到邮件提醒，服务器创建成功。 （为了写教程，花费320元，买了一个月服务器）

【7】返回弹性服务器的控制台

【8】点击服务器名字，可以进入到详情页面。

2.3 配置安全组

要确保MQTT服务器常用的几个端口已经开放出出来。

2.4 安装FinalShell

Windows下安装 FinalShell 终端，方便使用SSH协议远程登录到云服务器。 （当然，使用其他方式登录也是一样的）

2.5 远程登录到云服务器终端

【1】新建连接，选择SSH连接。

【2】填入IP地址、用户名、密码

这里的主机就是填服务器的公网IP地址，密码就是创建服务器输入的密码，用户名直接用root。

【3】点击连接服务器

【4】第一次登录会弹出提示框，选择接受并保存

【5】接下来可以看到服务器已经登录成功了。

二、Linux下安装EMQX

本章节将介绍如何在 Ubuntu 系统中下载安装并启动 EMQX。

支持的 Ubuntu 版本：

• Ubuntu 22.04
• Ubuntu 20.04
• Ubuntu 18.04

2.1 官网地址

链接：https://www.emqx.io/docs/zh/v5.2/deploy/install-ubuntu.html

2.2 通过Apt源安装

EMQX 支持通过 Apt 源安装，免除了用户需要手动处理依赖关系和更新软件包等的困扰，具有更加方便、安全和易用等优点。

在命令行终端，复制下面的命令过去，按下回车键。

【1】通过以下命令配置 EMQX Apt 源：

curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash
1

【2】运行以下命令安装 EMQX：

sudo apt-get install emqx
1

【3】运行以下命令启动 EMQX：

sudo systemctl start emqx
1

过程如下：

2.3 EMQX常用的命令

sudo systemctl emqx start    启动
sudo systemctl emqx stop     停止
sudo systemctl emqx restart  重启 
1
2
3

三、配置EMQX服务器

3.1 登录EMQX内置管理控制台

EMQX 提供了一个内置的管理控制台，即 EMQX Dashboard。方便用户通过 Web 页面就能轻松管理和监控 EMQX 集群，并配置和使用所需的各项功能。

在浏览器里输入: http://122.112.225.194:18083 就可以访问EMQX的后台管理页面。可以管理以连接的客户端或检查运行状态。

这里面的IP地址，就是自己ECS云服务器的公网IP地址。

打开浏览器后，输入地址后打开的效果：

默认用户名和密码：

用户名：admin
密码：public
1
2

第一次登录会提示你修改新密码，如果不想设置，也可以选择跳过（公网服务器部署，还是要修改密码安全些）。

下面修改新密码：

登录成功的页面显示如下：

3.2 MQTT配置

这里可以配置MQTT的一些参数，根据自己的需求进行配置。

3.3 测试MQTT通信

新建一个客户端，点击连接。

连接之后，然后点击订阅，和发布，如果下面消息能正常的接收。说明MQTT服务器通信是已经正常，没问题了。

并且在这个页面也可以看到主题发布和主题订阅的格式。

3.4 MQTT客户端登录服务器测试

接下来就打开我们自己的MQTT客户端登录MQTT服务器进行测试数据的通信。

端口选择: 1883

根据软件参数填入参数，登录，进行主题的发布和订阅。

说明： 目前还没有配置客户端认证，现在只要IP和端口输入正确，MQTT三元组可以随便输入，都可以登录上服务器的，服务器没有对三元组做校验。

EMQ X 默认配置中启用了匿名认证，任何客户端都能接入 EMQX。没有启用认证插件或认证插件没有显式允许/拒绝（ignore）连接请求时，EMQX 将根据匿名认证启用情况决定是否允许客户端连接。

然后打开EMQX的管理后台，可以看到我们的设备已经登录服务器了，名字为test1。

在订阅主题的页面也可以看到我们客户端设备订阅的主题。

3.5 客户端认证配置

EMQX 默认配置中启用了匿名认证，任何客户端都能接入 EMQX。没有启用认证插件或认证插件没有显式允许/拒绝（ignore）连接请求时，EMQX 将根据匿名认证启用情况决定是否允许客户端连接。

在正式产品里肯定是要启用认证的，不然任何设备都能接入。

下面就介绍如何配置 客户端认证。

【1】打开客户端认证页面

【2】选择密码认证

【3】选择内置数据库

【4】设置认证方式（都可以默认，不用改），直接点击创建。

【5】创建成功后，点击用户管理

【6】添加用户

【7】添加成功

【8】添加完毕之后，打开MQTT客户端可以进行测试。

登录的时候，MQTT用户名和密码必须输入正确，按照上一步添加的信息进行如实填写，否则是无法登录服务器的。

3.6 客户端授权配置

客户端授权页面可以配置每个客户端(设备)的主题发布，订阅权限。限制它是否可以发布主题，订阅主题。 如果有需要就可以进行配置。

http://127.0.0.1:18083/#/authorization/detail/built_in_database?tab=users

【1】创建数据源

【2】选择内置数据库

【3】完成创建

【4】点击权限管理

【5】选择客户端ID，点击添加

【6】配置权限

3.7 数据转发(集成)

在集成选项里，可以对设备数据处理。 比如：转发到自己的HTTP服务器，转发到自己其他的MQTT服务器，创建规则，某些事件触发某些动作等等。

选择数据桥接。

可以把数据发送端自己的HTTP服务器，或者发送到其他的MQTT服务器。

选择HTTP服务 （如果自己有HTTP服务器，可以将数据转发给自己的HTTP服务器）。

四、MQTT客户端消息互发测试

4.1 添加2个设备

为了方便测试设备间互相订阅主题，数据收发，在客户端认证页面至少添加2个设备。我这里分别添加了test1和test2。

4.2 设备间测试

设备A订阅设备B的主题，设备B订阅设备A的主题，实现数据互发。

设备A的MQTT信息：

MQTT服务器地址：122.112.225.194
MQTT服务器端口号：1883
MQTT客户端ID：AAA
MQTT用户名：test1
MQTT登录密码：12345678

订阅主题：BBB/#
发布主题：AAA/1
发布的消息：{ "msg": "我是AAA设备" }
1
2
3
4
5
6
7
8
9

设备B的MQTT信息：

MQTT服务器地址：122.112.225.194
MQTT服务器端口号：1883
MQTT客户端ID：BBB
MQTT用户名：test2
MQTT登录密码：12345678

订阅主题：AAA/#
发布主题：BBB/1
发布的消息：{ "msg": "我是BBB设备" }
1
2
3
4
5
6
7
8
9

4.3 MQTT实时图传测试

下面演示，利用MQTT协议进行图像传输。

这是打包好的APP，0积分可以下载，下载下来就可以测试图传：
https://download.csdn.net/download/xiaolong1126626497/88512661

这是MQTT协议图传上位机的Qt源码，如果需要源码的可以到这里下载：
https://download.csdn.net/download/xiaolong1126626497/87614468

这是Qt的工程，利用Qt开发的图传软件，分别为发送端和接收端。

下面演示，利用MQTT协议进行图像传输。

这是Qt的工程，利用Qt开发的图传软件，分别为发送端和接收端。

运行效果： 进行图传效果是非常的好。 报文最大一次可以发送 256MB

五、单片机设备上云

只要是MQTT客户端能正常上云通信了，那么单片机也是一样的。

上位机也可以采用MQTT协议接入服务器，订阅设备的主题，就可以实时接收设备的消息（当然，也可以采用HTTP协议接入）。

六、数据桥接

EMQX支持将设备上传的数据转发到其他地方，比如，自己的HTTP服务器。方便自己服务器进行其他的处理。

通过数据桥接，用户可以实时地将消息从 EMQX 发送到外部数据系统，或者从外部数据系统拉取数据并发送到 EMQX 的某个主题。而 EMQX Dashboard 提供了可视化创建数据桥接的能力，只需在页面中配置相关资源即可。

本章节就介绍如何搭建自己的HTTP服务器。配置EMQX转发数据到自己的HTTP服务器，保存处理数据。

6.1 搭建HTTP服务器

我这里直接使用python写代码搭建一个HTTP服务器。 ECS服务器上默认没有安装python3，需要先安装一下。

【1】安装python3

root@emqx:~/emqx# apt install python3 
Reading package lists... Done
Building dependency tree       
Reading state information... Done
python3 is already the newest version (3.6.7-1~18.04).
0 upgraded, 0 newly installed, 0 to remove and 1 not upgraded.
1
2
3
4
5
6

【2】编写代码

from flask import Flask, json, request

app = Flask(__name__)

@app.route('/', methods=['POST'])
def print_messages():
  reply= {"result": "ok", "message": "success"}
  print("got post request: ", request.get_data())
  return json.dumps(reply), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8000)
1
2
3
4
5
6
7
8
9
10
11
12

将以上代码保存到一个名为 server.py 的文件中。

这段代码创建了一个使用 Flask 框架的 Web 服务器，可以接收根路径的 POST 请求。当接收到 POST 请求时，服务器会将请求的数据打印到终端，并返回一个 JSON 格式的响应给客户端。服务器将在本地运行，并监听默认的 8000 端口。

【3】运行程序

# 安装 flask 依赖
pip install flask
pip3 install flask
    
# 启动服务
python3 server.py
1
2
3
4
5
6

在命令行中执行 python3 server.py，就可以启动一个简单的HTTP服务器，可以接收并处理POST请求。当有POST请求发生时，服务器将返回收到的POST数据。可以根据自己的需要，进一步扩展处理POST请求的逻辑。

**运行示例： ** （代码可以在本地写好上传到服务器，也可以直接 vim server.py 打开编辑器直接编写 ）

可以通过发送 POST 请求到 http://your-server-ip:8000/ 的方式来测试这个服务器。

比如：

http://122.112.225.194:8000/
1

6.2 数据转发配置

【1】在集成选项里，可以对设备数据处理，将数据转发到自己的HTTP服务器。

【2】选择Webhook。

Webhook，使用 Webhook 来转发数据到 HTTP 服务；

使用 Webhook 其实就是将 EMQX 接收并处理后的数据发送到一个 HTTP 服务上，再根据预设好的 HTTP 服务来处理和集成业务数据。

同样用户需要有一个预先搭建好的 HTTP 服务，需要在配置信息页面填写 HTTP 请求的服务地址，选择一个请求方法 POST、GET、PUT 或 DELETE，配置请求头，将需要发送的数据使用模板语法填写到请求体（body）中即可。

【3】选配置Webhook

触发器选择所有消息和事件，URL里填自己的服务器地址。

【4】点击测试。 测试服务器是否OK。

【6】没问题就直接点击保存

【7】创建成功

【8】保存之后。 会自动创建规则和数据桥接。 非常方便。

6.3 测试转发效果

【1】打开MQTT客户端，发送数据测试。

【2】看python服务器的终端，可以看到收到了EMQX服务器转发过来的数据。

如果自己接下来想要进行其他的操作，服务器写代码进行对应的处理即可。

【3】 点击这个服务可以看到已经触发转发的详情。