一、概要
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
B：配置位系统服务，权限调整：
chmod +x /etc/init.d/emqttd
chkconfig --add emqttd
chkconfig --list
C：service emqttd start ##如果“必须设置 HOME”错误，则必须设置 HOME 取消注释“#export HOME=/root”。