0 简介

去年玩了一阵NB-IOT，数据走的是电信、华为、阿里云等成熟平台，想利用自己的私有云平台，终端设备采集的数据直接通过NB-IOT的UDP、TCP/IP、COAP、MQTT的等方式透传到自己的云平台。开始想利用UDP，但是UDP不安全，随后尝试过COAP协议，安装完libcoap后不知如何下手，就没有完全深入。

随后开始研究通过S7协议将西门子PLC数据采集到树莓派中，存储到数据库中，实现了数据的采集。随后继续尝试数据的透传，因采用的树莓派，所以考虑采用MQTT的方式。

最开始打算采用MQTT是Apache Apollo，安装完成后，发现相关配套资料不多，并且版本很久没有更新，取而代之的是繁杂的Apache ActiveMQ。在分析各种客户端对比后，选择国产开源的Emqx。

Emqx 为百万级消息服务器，使用配置简单，还有管理平台、各种插件，中文文档。

1 软硬件

1.1 硬件

NBIOT模块：微雪SIM700C NB-IoT HAT ，中国移动的NB-IOT体验卡（据说电信、联通的不支持自己搭建平台的数据传输，有待核实）
树莓派：4B
云服务器：腾讯云 win2008 R2 SP1 64位，mysql server 5.1

1.2 软件

调试电脑：win 10
串口调试助手：sscom5.13.1
emqx:4.2.1
python：3.65
pycharm IDE编辑器
paho-mqtt 库：1.5.1

2 环境搭建

2.1 Emqx 的安装

2.1.1 下载Emqx的安装包

在Emqx官方网站下载最新版本的Emqx-4.2.1，选择windows系统版本。注意Emqx只支持64位系统。

2.1.2 解压

登录到云服务器后，上传安装包，并解压，将emqx放置于c:serveremqx目录下，

2.1.3 安装

在CMD终端下，进入Emqx的bin文件下，运行 emqx install。

2.1.4 启动等相关命令

在bin文件下输入emqx可以看到安装、卸载、启动、停止等相关命令

启动： emqx start

停止：emqx stop
重启：emqx restart

2.1.5 缺少MSVCR120.dll处理办法

在安装和启动过程中出现MSVCR120.dll缺少无法执行的错误，可以参考下面文章的终极方法——下载DirectX修复工具，选择工具-》选项打开-》实验室-》选中C++强力修复，完美解决。

2.2 Emqx的测试

2.2.1 Emqx的端口说明

控制台Dashboard的端口为：18083

2.2.2 登录控制台

在服务器IE浏览器中输入http://localhost:18083/，出现Dashboard的登录界面。

输入用户名称admin，密码public，进入控制台页面。

2.2.3 修改控制台界面语言

为了方便使用，可以在Settings，将语言修改为中文，方便使用。

2.3 串口调试器驱动的安装

在微雪官方网站上下载串口调试驱动和串口调试软件，具体安装就不再叙述。

3 SIM7000C MQTT连接测试

3.1 Emqx 新增加用户

微雪官方在SIM7020C 上给出了连接MQTT的方法，其方法是连接公开服务器，不用用户名称和密码的，登录实际的emqx需要配置用户名称和密码。

从官方手册上得知，添加用户可以在conf上添加，但是这样出现错误就无法恢复。可以采用命令行方式处理。

在bin目录下的运行emqx_ctl users add 用户名 密码，例如创建testapp1用户，密码为testapp1

3.2 NBIOT设备上网注册

3.2.1 启动SSCOM调试助手

给树莓派上电，模块数据先连接电脑，打开调试助手，其运行比较慢，容易死机，需要一点耐心。

3.2.2 连接测试

端口号选择comX SimTech HS-USB AT Port 9001，波特率选择115200，点击打开串口。
再点击右侧的“扩展”，在弹出AT运行对话框中选择“AT”进行测试，如果左侧反馈为OK标识串口运行连接成功。

3.2.3 设备NB-IOT连接AT指令

AT #AT命令和串口运行正常测试
OK
AT+CNMP=38 #设置为LET模式
OK
AT+CMNB=2 #设置为NBIOT模式
OK
AT+NBSC=1  ##打开扰码设置
OK
AT+CSQ  ##网络信号质量状态，前面第一数值一般要12以上才能正常收发数据，信号质量太低可以将天线延长或者更换
+CSQ: 16,99
OK
AT+CGATT?  ## 查询网络是否附着
+CGATT: 1
OK
AT+CPSI? ##春夏网络注册信息
+CPSI: LTE NB-IOT,Online,460-00,0x5A21,145587201,320,EUTRAN-BAND8,3688,0,0,-8,-104,-96,9
OK
AT+CGNAPN ##查询APN码
+CGNAPN: 1,"cmnbiot"
OK
AT+CSTT="cmnbiot"  ###设置对应的APN码，必须开启树莓派才能连接成功
OK
AT+CIICR ## 激活移动场景
OK
AT+CIFSR ##获取本地IP地址
200.189.41.2××

3.3 连接Emqx服务器

AT+CNACT=1 ##开启无线连接，也可以输入AT+CNACT=1，“cmnboit”
OK
+APP PDP: ACTIVE
AT+CNACT? ##获取IP地址
+CNACT: 1,"200.189.41.2××"
OK
AT+SMCONF="URL",232.137.102.1××,1883 ##填入Emqx的外网IP地址和端口号，此外也可以填入网址加端口号
OK
AT+SMCONF="KEEPTIME",60 ## 心跳包，连接时间
OK
AT+SMCONF="USERNAME","testapp1" ##连接用户名，必填
OK
AT+SMCONF="PASSWORD","testapp1" ##连接密码，必填
OK
AT+SMCONN
OK

3.4 订阅消息

假设我们订阅test消息，消息等级为1。

AT+SMSUB="test",1  ##订阅消息test，消息等级为1
OK

备注一下消息等级：
qos = 0，最多一次送达。也就是消息发出去就直接删除掉，没有后面的事情了。
qos = 1，至少一次送达。发送方消息发出去之后会将消息保存，等待接收方带 messageId 的回应，如果没有回应，就会重新发送，直到回应到达后再把消息删除。
qos = 2，准确一次送达。

3.5 Emqx Dashbord 查看连接

3.5.1 客户端信息

登录Dashbord，在客户端可以检索到我们刚刚用testapp1用户名称的连接。

3.5.2 主题信息

点击主题，可以看到我们刚刚订阅的test主题。

订阅下也能找到test列表。

3.6 SIM7000C发送消息到云服务器

3.6.1 打开Websocket监控

在Dashboard的工具下选择“Websocket”，点击连接。

3.6.2 打开websocket的订阅

在Websocket界面下的订阅栏，输入我们订阅主题“test”，服务质量选择“1”

3.6.3 数据发送

假设我们要在test主题上发送“Hello,NB-IOT”这12个字，采用串口助手输入以下：

AT+SMPUB="test","12",1,1 ##test为主题，12为字节数量
> Hello,NB-IOT  ##输入的文字，注意数据发送比较慢，不用重复回车，重复回车回重复发
OK
+SMSUB: "test","Hello,NB-IOT"

3.6.4 数据服务器端查看

在Websocket界面下的消息链栏右侧，刷新后可以看到消息内容、主题服务质量和时间。
因上传反应比较慢，重复回车了几次，所以重复上传了4次。

3.7 云服务器发送消息到SIM7000C

3.7.1 服务器端发送消息

在Websocket端，消息栏中，将主题设定为“test”，消息内容为“{ “msg”: “Hello, SIM7000C!” }”，服务质量选1，点击发送。发送完成后回在发布消息列表和订阅消息列表均可以看到。

3.7.2 调试助手查看消息

调试助手上手动了这条信息。

4 paho-mqtt python库使用

各种语言对应MQTT客户端现均有，具体可以参考Emqx整理的资料。
对于python比较成熟的库为paho-mqtt，可以将其作为客户端，Emqx作为中间信息交换器，SIM7000C为终端。可以实现三者的相关信息的传递（发送和接收）。

4.1 安装paho-mqtt库

pah0-mqtt 现在为1.5.1版本

pip install paho-mqtt

4.2 编辑代码

参考凌晨点点的大作，python代码如下：

import paho.mqtt.client as mqtt
import json
import time
import random

#########################
##设置MQTT Broker连接参数
########################
# EMQX主机信息
HOST = "232.137.102.110"
PORT = 1883
# Broker的登录账号（EMQX上设定的可访问的账户密码）
client_id = f'py-mqtt-test-{random.randint(0,1000)}'    ##客户端ID，可以用随机函数产生
username = 'testapp1'
password = 'testapp1'
#话题topic
topic="test"


# 订阅信息的回调函数
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("data/receive")

##打印订阅消息
def on_message(client, userdata, msg):
    print("主题:"+msg.topic+" 消息:"+str(msg.payload.decode('utf-8')))

##订阅消息
def on_subscribe(client, userdata, mid, granted_qos):
    print("On Subscribed: qos = %d" % granted_qos)


def on_disconnect(client, userdata, rc):
    if rc != 0:
        print("Unexpected disconnection %s" % rc)
##需要发送消息
data = {
    "type":2,
    "timestamp": time.time(),
    "messageId":"9fcda359-89f5-4933-432",
    "command":"233/recommend",
    "data":{
        "openId":"12345",
        "appId":568789,
        "recommendType":"testing"
    }
}
param = json.dumps(data)


client = mqtt.Client(client_id)
client.username_pw_set(username, password)
client.on_connect = on_connect
client.on_message = on_message
client.on_subscribe = on_subscribe
client.on_disconnect = on_disconnect
client.connect(HOST, PORT, 60) ##链接
client.publish("test", payload=param, qos=1)     # 发送消息
client.loop_start()

4.3 运行结果

webshoct上能查询到发送消息调试助手上收到的信息

5 Emqx插件mysql数据库认证

5.1 MySQL数据库配置

数据库配置文件路径和文件为：C:serveremqxetcplugins6emqx_backend_mysql.conf

在这里插入代码片
##--------------------------------------------------------------------
## MySQL Auth/ACL Plugin
##--------------------------------------------------------------------
## MySQL server address.
##
## Value: Port | IP:Port
##
## Examples: 3306, 127.0.0.1:3306, localhost:3306
##auth.mysql.server = 127.0.0.1:3306
auth.mysql.server = localhost:3305 ###数据库地址和端口号
## MySQL pool size.
##
## Value: Number
auth.mysql.pool = 8
## MySQL username.
##
## Value: String
auth.mysql.username = Datauser ##数据库登录用户名称
## MySQL password.
##
## Value: String
auth.mysql.password = password ### 数据库登录密码
## MySQL database.
##
## Value: String
auth.mysql.database = mqtt  ##数控库名称
## MySQL query timeout
##
## Value: Duration
auth.mysql.query_timeout = 5s
## Variables: %u = username, %c = clientid
## Authentication query.
##
## Note that column names should be 'password' and 'salt' (if used).
## In case column names differ in your DB - please use aliases,
## e.g. "my_column_name as password".
##
## Value: SQL
##
## Variables:
##  - %u: username
##  - %c: clientid
##  - %C: common name of client TLS cert
##  - %d: subject of client TLS cert
##
auth.mysql.auth_query = select password from mqtt_user where username = '%u' limit 1
## auth.mysql.auth_query = select password_hash as password from mqtt_user where username = '%u' limit 1
## Password hash.
##
## Value: plain | md5 | sha | sha256 | bcrypt
##auth.mysql.password_hash = sha256
auth.mysql.password_hash = plain ##设置加密方式，为了方便改成不加密，生产应用更加实际情况处理
## sha256 with salt prefix
## auth.mysql.password_hash = salt,sha256
## bcrypt with salt only prefix
## auth.mysql.password_hash = salt,bcrypt
## sha256 with salt suffix
## auth.mysql.password_hash = sha256,salt
## pbkdf2 with macfun iterations dklen
## macfun: md4, md5, ripemd160, sha, sha224, sha256, sha384, sha512
## auth.mysql.password_hash = pbkdf2,sha256,1000,20
## Superuser query.
##
## Value: SQL
##
## Variables:
##  - %u: username
##  - %c: clientid
##  - %C: common name of client TLS cert
##  - %d: subject of client TLS cert
##
auth.mysql.super_query = select is_superuser from mqtt_user where username = '%u' limit 1
## ACL query.
##
## Value: SQL
##
## Variables:
##  - %a: ipaddr
##  - %u: username
##  - %c: clientid
##
## Note: You can add the 'ORDER BY' statement to control the rules match order
auth.mysql.acl_query = select allow, ipaddr, username, clientid, access, topic from mqtt_acl where ipaddr = '%a' or username = '%u' or username = '$all' or clientid = '%c' ##认证与权限的查询与匹配
## Mysql ssl configuration.
##
## Value: on | off
auth.mysql.ssl = off
## CA certificate.
##
## Value: File
## auth.mysql.ssl.cafile  = path to your ca file
## Client ssl certificate.
##
## Value: File
## auth.mysql.ssl.certfile = path to your clientcert file
## Client ssl keyfile.
##
## Value: File
## auth.mysql.ssl.keyfile = path to your clientkey file

5.2 新建数据库和表

因MySQL 5.1数据库配置中没有开启InnoDB，需要再配置文件中开启，重启MySQL即可。
新建mqtt数据库，创建如下两张表。
mqtt_user用户表

CREATE TABLE `mqtt_user` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `username` varchar(100) DEFAULT NULL,
  `password` varchar(100) DEFAULT NULL,
  `salt` varchar(35) DEFAULT NULL,
  `is_superuser` tinyint(1) DEFAULT 0,
  `created` datetime DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `mqtt_username` (`username`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

mqtt_acl用户权限表

CREATE TABLE `mqtt_acl` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `allow` int(1) DEFAULT NULL COMMENT '0: deny, 1: allow',
  `ipaddr` varchar(60) DEFAULT NULL COMMENT 'IpAddress',
  `username` varchar(100) DEFAULT NULL COMMENT 'Username',
  `clientid` varchar(100) DEFAULT NULL COMMENT 'ClientId',
  `access` int(2) NOT NULL COMMENT '1: subscribe, 2: publish, 3: pubsub',
  `topic` varchar(100) NOT NULL DEFAULT '' COMMENT 'Topic Filter',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

在mqtt_user数据表中添加用户名和密码（testapp1，testapp2）。

在mqtt_acl权限表中添加如下相关信息

5.3 加载插件

进入Dashboard界面，在插件中，将emqx_auth_mysql启动。

启动成功回显示状态为运行中。若出现报错，检查配置文件中设置，例如数据库IP，端口号，登录用户名和密码等。

5.4 认证测试

利用SIM7000C的用户testapp1登录测试，先去CMD终端emqx_ctl users del 删除记录。（当然也可以新建一个用户和密码测试）

在调试助手中输入下列AT指令

AT+SMCONF="URL",232.137.102.1××,1883
OK
AT+SMCONF="KEEPTIME",60
OK
AT+SMCONF="USERNAME","testapp1"
OK
AT+SMCONF="PASSWORD","testapp1"
OK
AT+SMCONN
OK
AT+SMCONF="QOS",1 
OK
AT+SMSUB="test",1
OK

在客户端能查询到testapp1登录信息。

说明配置成功。

6 总结

利用Emqx作为中间插件，能很方便的实现终端设备通过MQTT协议与自己云服务器通信。
Emqx的企业版本支持根据规则插件功能，可以直接将数据进行处理，并存储到相应的数据库或者后续单元中。

7 参考文献

[1] MQTT学习（六）–各类MQTT代理服务器特性对比（译文）
[2]Emqx 开发文档
[3]阿里云、腾讯云服务器Windows 2008 R2系统上搭建EMQ X(emqttd ) MQTT服务器
[4]解决msvcr120.dll文件丢失问题（搞了半天，简直奔溃，最后完美解决）
[5] paho-mqtt python库官方
[6] EMQ_Mysql基础配置和操作
[6] EMQ X 规则引擎系列（二）存储消息到 MySQL 数据库
[7] EMQ X MQTT Broker 规则引擎实战——在线状态记录、消息 MySQL 存储
[8] EMQ X 插件持久化系列 （五）MySQL MQTT 数据存储

最后

以上就是忧伤悟空为你收集整理的微雪电子SIM7000C NB-IOT Emqx MQTT 通信的详细测试过程0 简介1 软硬件2 环境搭建3 SIM7000C MQTT连接测试4 paho-mqtt python库使用5 Emqx插件mysql数据库认证6 总结7 参考文献的全部内容，希望文章能够帮你解决微雪电子SIM7000C NB-IOT Emqx MQTT 通信的详细测试过程0 简介1 软硬件2 环境搭建3 SIM7000C MQTT连接测试4 paho-mqtt python库使用5 Emqx插件mysql数据库认证6 总结7 参考文献所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。