关于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）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

1. 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
1
2
3
4
5

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

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

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

https://localhost:18083
1

默认账号：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
1
2
3
4
5
6
7
8

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
1
2
3
4
5
6
7

B:TCP stack

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

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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

D：EMQ客户端侧配置：

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

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    |
----------------------                                                                   -------------------
1
2
3
4
5

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

9）常用TCP端口：

四、使用

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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

B：配置位系统服务，权限调整：

chmod +x /etc/init.d/emqttd
chkconfig --add emqttd
chkconfig --list

C：service emqttd start ##如果“必须设置 HOME”错误，则必须设置 HOME 取消注释“#export HOME=/root”。