關于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

更多安裝方式，請單擊查看！

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埠：

四、使用

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”，