[嵌入式] 关于EMQ（emqttd）你应该了解的

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 嵌入式 -> 关于EMQ（emqttd）你应该了解的 -> 正文阅读

[嵌入式]关于EMQ（emqttd）你应该了解的

一、概要

EMQ (Erlang/Enterprise/Elastic MQTT Broker) 是基于 Erlang/OTP 平台开发的开源物联网 MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议）消息代理服务。

Erlang/OTP 是出色的软实时(Soft-Realtime)、低延时(Low-Latency)、分布式(Distributed) 的语言平台。MQTT 是轻量的(Lightweight)、发布订阅模式(PubSub) 的**物联网消息协议。**EMQ 是完全开源的，并在 Apache 2.0 版下获得许可。 EMQ 同时实现了 MQTT V3.1 和 V3.1.1 协议规范，同时支持 MQTT-SN、CoAP、WebSocket、STOMP 和 SockJS。

EMQ 项目设计目标是承载移动终端或物联网终端海量 MQTT 连接，并实现在海量物联网设备间快速低延时消息路由，EMQ 为物联网、M2M、智能硬件、移动消息和 HTML5 网络消息应用程序提供了一个可扩展的、企业级的、可扩展的开源 MQTT 代理，传感器、手机、Web 浏览器和应用服务器可以通过 EMQ 代理与异步 PUB/SUB MQTT 消息连接:

1> 稳定承载大规模的 MQTT 客户端连接，单服务器节点支持50万到100万连接。

2> 分布式节点集群，快速低延时的消息路由，单集群支持1000万规模的路由。

3> 消息服务器内扩展，支持定制多种认证方式、高效存储消息到后端数据库。

4> 完整物联网协议支持，MQTT、MQTT-SN、CoAP、WebSocket 或私有协议支持

EMQ 基于标准 MQTT 消息、每秒百万级的高性能、低时延、高可靠解决方案，EMQ 代理的 1.0 版本已在 12 核、32G CentOS 服务器上扩展到 130 万个并发 MQTT 连接。它可保障双向实时数据顺畅移动，灵活分发至其他企业系统，满足物联网业务中各类数据需求。更多查看EMQ官网。

官方文档。

中移某拨测项目采用了该技术，支持百万级拨测并发。

二、架构原理

EMQ拥抱 Erlang/OTP！是高并发、软实时、低延迟和容错平台。通过主题树匹配和路由表查找在集群节点之间路由 MQTT 消息。

在这里插入图片描述
EMQ代理可接入各平台，集成使用：

【MQTT 回顾】

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。
在这里插入图片描述

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

MQTT协议实现方式

实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

（1）Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）；
（2）payload，可以理解为消息的内容，是指订阅者具体要使用的内容。

2）网络传输与应用消息
MQTT会构建底层网络传输：它将建立客户端到服务器的连接，提供两者之间的一个有序的、无损的、基于字节流的双向传输。

当应用数据通过MQTT网络发送时，MQTT会把与之相关的服务质量（QoS）和主题名（Topic）相关连。

3）MQTT客户端

一个使用MQTT协议的应用程序或者设备，它总是建立到服务器的网络连接。客户端可以：

（1）发布其他客户端可能会订阅的信息；
（2）订阅其它客户端发布的消息；
（3）退订或删除应用程序的消息；
（4）断开与服务器连接。
4.4 MQTT服务器
MQTT服务器以称为"消息代理"（Broker），可以是一个应用程序或一台设备。它是位于消息发布者和订阅者之间，它可以：

（1）接受来自客户的网络连接；
（2）接受客户发布的应用信息；
（3）处理来自客户端的订阅和退订请求；
（4）向订阅的客户转发应用程序消息。

4） MQTT协议中的订阅、主题、会话

一、订阅（Subscription）

订阅包含主题筛选器（Topic Filter）和最大服务质量（QoS）。订阅会与一个会话（Session）关联。一个会话可以包含多个订阅。每一个会话中的每个订阅都有一个不同的主题筛选器。

二、会话（Session）

每个客户端与服务器建立连接后就是一个会话，客户端和服务器之间有状态交互。会话存在于一个网络之间，也可能在客户端和服务器之间跨越多个连续的网络连接。

三、主题名（Topic Name）

连接到一个应用程序消息的标签，该标签与服务器的订阅相匹配。服务器会将消息发送给订阅所匹配标签的每个客户端。

四、主题筛选器（Topic Filter）

一个对主题名通配符筛选器，在订阅表达式中使用，表示订阅所匹配到的多个主题。

五、负载（Payload）

消息订阅者所具体接收的内容。

5） MQTT协议中的方法

MQTT协议中定义了一些方法（也被称为动作），来于表示对确定资源所进行操作。这个资源可以代表预先存在的数据或动态生成数据，这取决于服务器的实现。通常来说，资源指服务器上的文件或输出。主要方法有：

（1）Connect。等待与服务器建立连接。
（2）Disconnect。等待MQTT客户端完成所做的工作，并与服务器断开TCP/IP会话。
（3）Subscribe。等待完成订阅。
（4）UnSubscribe。等待服务器取消客户端的一个或多个topics订阅。
（5）Publish。MQTT客户端发送消息请求，发送完成后返回应用程序线程。

三、安装部署

EMQ 代理是跨平台的，可以部署在 Linux、FreeBSD、Mac、Windows 甚至 Raspberry Pi 上。

1）下载emqttd-2.2-rc.2 : 适用centos7 64位的RPM包。

2）rpm -ivh emqttd-centos7-v2.3.9-1.el7.centos.x86_64.rpm

3）源码安装：

git clone https://github.com/emqtt/emqttd.git

cd emqttd && make rel

cd rel/emqttd && ./bin/emqttd console

更多安装方式，请单击查看！

4）安装依赖：yum install lksctp-tools ##Erlang/OTP R19 依赖 lksctp-tools 库，EMQ 代理从 1.1 版本开始需要 Erlang R18+来构建。

5）当 EMQ 代理成功启动时，将加载一个 Web Dashboard。Dashboard 里可检查代理的运行状态，监控 MQTT 数据包的统计和指标，查询客户端、会话、主题和订阅。访问：

https://localhost:18083

默认账号：admin/public

6）服务启停：systemctl start|stop|restart emqttd.service

# Start emqttd
./bin/emqttd start

# Check Status
./bin/emqttd_ctl status

# Stop emqttd
./bin/emqttd stop

7）EMQ插件：

EMQ 代理可以通过插件扩展。插件是一个 Erlang 应用程序，它为 EMQ 代理添加了额外的功能：
在这里插入图片描述
详情见官网。可以通过命令：bin/emqttd_ctl plugins load来加载指定的插件，如：

./bin/emqttd_ctl plugins load emq_auth_pgsql

8）EMQ配置：

最新版本的 EMQ 代理正在 12 核、32G CentOS 服务器上扩展到 130 万个 MQTT 连接。默认情况下，emqttd 代理仅允许 512 个并发连接，因为“ulimit -n”在大多数平台上限制为 1024。这样，就需要我们调整操作系统内核、TCP 堆栈、Erlang VM 和 emqttd 代理以实现一百万个连接基准测试。

A：内核参数：

# 2M:
sysctl -w fs.file-max=2097152
sysctl -w fs.nr_open=2097152
echo 2097152 > /proc/sys/fs/nr_open

# 1M:
ulimit -n 1048576

B:TCP stack

#backlog
sysctl -w net.core.somaxconn=65536

C：配置文件：emqttd/etc/emq.conf:

##	Max number of Erlang proccesses. A MQTT client consumes two proccesses. The value should be larger than max_clients * 2
node.process_limit = 2097152  

## Max number of Erlang Ports.  Sets the maximum number of simultaneously existing ports for this system；A MQTT client consumes one port. The value should be larger than max_clients
node.max_ports = 1048576

## node.process_limit > 最大允许并发客户端数 * 2 node.max_ports > 最大允许并发客户端数

## Size of acceptor pool连接数
listener.tcp.acceptors = 64

## Maximum number of concurrent clients
listener.tcp.max_clients = 1000000

##端口配置
## TCP Listener: 1883, 127.0.0.1:1883, ::1:1883
listener.tcp.external = 0.0.0.0:1883

## SSL Listener: 8883, 127.0.0.1:8883, ::1:8883
listener.ssl.external = 8883

## External MQTT/WebSocket Listener
listener.ws.external = 8083

## HTTP Management API Listener
listener.api.mgmt = 127.0.0.1:8080

D：EMQ客户端侧配置：

sysctl -w net.ipv4.ip_local_port_range="500 65535"
echo 1000000 > /proc/sys/fs/nr_open
ulimit -n 100000

E：常见文件路径：

EMQ 配置文件: /etc/emqttd/emq.conf
插件配置文件: /etc/emqttd/plugins/*.conf。

注意：在 EMQ 代理启动之前，配置文件将被预处理并转换为 Erlang app.config：

----------------------                                          2.0/schema/*.schema      -------------------
| etc/emq.conf       |                   -----------------              \|/              | data/app.config |
|       +            | --> mergeconf --> | data/app.conf | -->  cuttlefish generate  --> |                 |
| etc/plugins/*.conf |                   -----------------                               | data/vm.args    |
----------------------                                                                   -------------------

日志文件目录: /var/log/emqttd
数据文件目录：/var/lib/emqttd/

9）常用TCP端口：

端口	用途
1883	MQTT Port
8883	MQTT/SSL Port
8083	MQTT/WebSocket Port
8084	MQTT/WebSocket/SSL Port
8080	HTTP Management API Port
18083	Web Dashboard Port

四、使用

1）EMQ服务构建

A：常见启动服务：/etc/init.d/emqttd

#
# emqttd       Startup script for emqttd.
#
# chkconfig: 2345 90 10
# description: emqttd is mqtt broker.

# source function library
. /etc/rc.d/init.d/functions

# export HOME=/root

start() {
    echo "starting emqttd..."
    cd /opt/emqttd && ./bin/emqttd start
}

stop() {
    echo "stopping emqttd..."
    cd /opt/emqttd && ./bin/emqttd stop
}

restart() {
    stop
    start
}

case "$1" in
    start)
        start
        ;;
    stop)
        stop
        ;;
    restart)
        restart
        ;;
    *)
        echo $"Usage: $0 {start|stop}"
        RETVAL=2
esac