核心引擎使用者手冊

# 前言

本文檔為金蝶Apusic分佈式消息隊列for MQTT（Apusic Distributed Message Queue，簡稱：ADMQ for MQTT）消息引擎的用戶手冊，詳細介紹了ADMQ for MQTT消息引擎的功能使用、配置方法及管理操作等內容。

# 適用對象

本文檔適用於IT信息化業務負責人、研發經理、軟件項目經理、軟件架構師、運維工程師。

# 相關文檔

瞭解更多ADMQ for MQTT產品相關的信息，請參閱以下ADMQ for MQTT產品手冊文檔集：

序號 手冊文檔 說明
1 金蝶Apusic分佈式消息隊列for MQTT 快速使用手冊 簡單介紹瞭如何快速上手使用ADMQ for MQTT 。
2 金蝶Apusic分佈式消息隊列for MQTT 安裝手冊 詳細介紹如何在各操作系統上安裝ADMQ for MQTT，以及ADMQ for MQTT服務啟停等操作。
3 金蝶Apusic分佈式消息隊列for MQTT 消息引擎用戶手冊 詳細介紹 ADMQ for MQTT 消息引擎相關功能的使用、配置、管理及配套工具的使用方法。
4 金蝶Apusic分佈式消息隊列for MQTT 管控臺用戶手冊 詳細介紹ADMQ for MQTT管控臺相關功能的使用和操作說明。
5 金蝶Apusic分佈式消息隊列for MQTT 開發手冊 詳細介紹基於各開發語言進行ADMQ for MQTT客戶端應用開發的說明。
6 金蝶Apusic分佈式消息隊列for MQTT 遷移手冊 詳細介紹從MQTT Broker遷移到ADMQ for MQTT的說明。
7 金蝶Apusic分佈式消息隊列for MQTT 運維手冊 詳細介紹ADMQ for MQTT的監控、運維、安全加固等運維說明。
8 金蝶Apusic分佈式消息隊列for MQTT 性能優化手冊 詳細介紹ADMQ for MQTT性能調優的說明。

# 技術支持

ADMQ for MQTT產品提供全面的技術支持服務，您可以通過以下方式獲得技術支持：

  • 網址：www.apusic.com
  • 電話：400-855-5800
  • 郵箱：support@apusic.com
  • 金蝶雲社區：https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

您在取得技術支持時，請提供如下信息：

  1. 您的姓名

  2. 公司信息與聯繫方式

  3. 操作系統及其版本

  4. 產品版本號

  5. 出現異常及錯誤的日誌、截圖等詳細信息

# 核心概念

# MQTT 協議

MQTT（Message Queuing Telemetry Transport）是一種基於發佈/訂閱模式的輕量級消息協議，專為低帶寬、高延遲或不可靠網絡設計。ADMQ for MQTT 完全兼容 MQTT 3.1、3.1.1、5.0 協議。

# 消息模型

发布者 ──→ MQTT Broker
              ├── 主题匹配 ──→ 订阅者 A（QoS 0）
              ├── 主题匹配 ──→ 订阅者 B（QoS 1）
              └── 保留消息存储 ──→ 新订阅者立即收到
1
2
3
4
  1. 發佈者發送消息到指定 Topic
  2. Broker 根據 Topic 匹配規則將消息分發給訂閱者
  3. 訂閱者接收並處理消息

# 主題（Topic）

# 主題格式

主題是 MQTT 中消息的邏輯分類單元，使用 / 分隔層級：

sensor/living-room/temperature
sensor/bedroom/humidity
1
2

# 通配符

通配符 說明 示例
+ 匹配單個層級 sensor/+/temperature 匹配 sensor/living-room/temperature
# 匹配零個或多個層級，必須位於末尾 sensor/# 匹配 sensor/living-room/temperature 和 sensor/bedroom/humidity/data

# 保留消息

保留消息是 MQTT 的特殊機制：

  • 發佈者發送保留消息時，Broker 會為該主題保留最後一條保留消息
  • 新的訂閱者訂閱該主題時，會立即收到這條保留消息
  • 常用於設備狀態、配置下發等場景
  • 發送一條空的保留消息可以清除該主題的保留消息

# QoS（服務質量等級）

QoS 定義了消息傳遞的可靠性等級。

QoS 等級 名稱 說明 典型場景
0 最多一次 消息只發一次，不保證到達 高頻、可容忍丟失的 Telemetry 數據
1 至少一次 消息保證到達，可能重複 一般業務消息
2 恰好一次 消息保證到達且不重複 對一致性要求極高的場景

# QoS 說明

  • QoS 是發佈者和訂閱者之間的協商結果
  • 消息會按訂閱者指定的 QoS 等級投遞
  • QoS 1/2 需要客戶端確認，會佔用更多資源

# 會話與遺囑

# Clean Session

  • Clean Session = true：客戶端斷開連接後，Broker 不保存其訂閱關係和未確認消息
  • Clean Session = false：Broker 保存訂閱關係、未確認消息和 QoS 1/2 的離線消息，客戶端重連後可恢復

# MQTT 5.0 會話

MQTT 5.0 中對應概念為 Clean Start 和 Session Expiry Interval：

  • Clean Start：連接時是否清理之前的會話
  • Session Expiry Interval：會話過期時間

# 遺囑消息

遺囑消息在客戶端異常斷開連接時由 Broker 自動發佈：

  • 客戶端連接時預設遺囑主題、遺囑內容、QoS、Retain
  • 當連接異常斷開且未發送 DISCONNECT 時，Broker 自動發佈遺囑消息
  • 常用於設備上下線狀態通知

# 認證與授權

# 認證器

認證器定義了客戶端連接時如何驗證身份：

認證方式 說明
Password-Based 基於用戶名密碼認證
JWT 基於 JSON Web Token 認證
SCRAM 基於 SCRAM 挑戰響應機制

# 數據源

數據源 說明
built_in_database 內置數據庫
MySQL MySQL 數據庫
PostgreSQL PostgreSQL 數據庫
HTTP HTTP 服務

# 授權源

授權源定義了 ACL 規則的數據來源：

  • 多個授權源按排序依次匹配
  • 同一授權源內的規則按順序匹配，匹配成功後停止

# ACL 規則

ACL 用於控制客戶端對 MQTT 資源的訪問權限：

屬性 說明
主體 全部、用戶名、客戶端 ID
動作 發佈（publish）、訂閱（subscribe）、全部（all）
權限 允許（allow）、拒絕（deny）
主題 支持通配符
QoS 限制 可限制允許的 QoS 等級，-1 表示不限制
Retain 限制 可限制 Retain 行為，-1 表示不限制

# 擴展能力

# 規則引擎

規則引擎用於基於 SQL 語法處理消息：

SELECT
  payload.deviceId AS deviceId,
  payload.temperature AS temperature,
  timestamp AS ts
FROM
  "sensor/+/temperature"
WHERE
  payload.temperature > 30
1
2
3
4
5
6
7
8

規則可以觸發以下動作：

  • 重新發布到其他 Topic
  • 轉發到數據橋接
  • 發送 HTTP 請求
  • 寫入數據庫

# 數據橋接

數據橋接用於將 MQTT 消息轉發到外部系統：

橋接類型 說明
Kafka 轉發到 Kafka Topic
RocketMQ 轉發到 RocketMQ Topic
RabbitMQ 轉發到 RabbitMQ Exchange/Queue
MySQL/PostgreSQL 寫入關係型數據庫
HTTP 調用 HTTP API
Webhook 發送 Webhook 通知

# 主題重寫

主題重寫規則用於在客戶端發佈或訂閱時，自動將源主題替換為目標主題：

字段 說明
動作 規則生效的動作：subscribe、publish、all
源主題 原始主題
目標主題 重寫後的主題
正則表達式 匹配源主題的正則
排序 規則優先級

# 自動訂閱

自動訂閱規則用於在客戶端連接成功後，自動為其訂閱指定的主題：

字段 說明
主題 自動訂閱的主題
QoS 訂閱 QoS 等級
No Local MQTT 5.0 特性，是否接收自己發佈的消息
Retain As Published MQTT 5.0 特性，是否保留髮布時的 Retain 標誌
Retain Handling MQTT 5.0 特性，訂閱時保留消息的處理方式

# 共享訂閱

共享訂閱是 MQTT 5.0 特性，用於將消息負載均衡地分發給多個訂閱者：

$share/group1/sensor/+/temperature
1

同一共享組內的訂閱者輪流接收消息。

# 監聽器

# 監聽器類型

類型 端口 說明
TCP 1883 標準 MQTT TCP 接入
SSL 8883 加密 MQTT TCP 接入
WebSocket 8083 WebSocket 接入
WebSocket Secure 8084 加密 WebSocket 接入
QUIC 14567 基於 QUIC 的接入

# 監聽器配置

listeners.tcp.default {
  bind = "0.0.0.0:1883"
  max_connections = 1024000
}

listeners.ssl.default {
  bind = "0.0.0.0:8883"
  ssl_options {
    certfile = "/path/to/server.crt"
    keyfile = "/path/to/server.key"
    cacertfile = "/path/to/ca.crt"
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

# 系統配置

# Broker 配置

參數 說明 默認值
max_packet_size 最大報文大小 1MB
max_clientid_len 最大 ClientID 長度 65535
max_topic_levels 最大主題層級 128
max_topic_alias 最大主題別名數 65535
retain_available 是否支持保留消息 true
wildcard_subscription 是否支持通配符訂閱 true
shared_subscription 是否支持共享訂閱 true

# 會話配置

參數 說明 默認值
session_expiry_interval 會話過期時間 2h
max_subscriptions 單個客戶端最大訂閱數 infinity
max_inflight 飛行窗口大小 32
max_mqueue_len 消息隊列長度 1000

# 日誌配置

參數 說明 默認值
log.level 日誌級別 warning
log.file 日誌文件路徑 log/admq-mqtt.log
log.rotation_size 日誌輪轉大小 10MB
log.rotation_count 日誌保留數量 5

# 常用命令

# 節點啟停

# 启动
bin/admq-daemon start mqtt server
# 停止
bin/admq-daemon stop mqtt server
1
2
3
4

# 節點管理

# 查看节点状态
bin/admq mqtt admin ctl status
bin/admq mqtt admin ctl status

# 查看版本与运行时长
bin/admq mqtt admin ctl broker
# 输出：version / sysdescr / uptime / datetime

# 查看 Broker 统计（连接数、会话数、主题数等）
bin/admq mqtt admin ctl broker stats

# 查看 Broker 指标（消息收发、认证、授权等）
bin/admq mqtt admin ctl broker metrics

# 查看节点 Erlang 信息
bin/admq mqtt admin ctl vm
bin/admq mqtt admin ctl vm ports       # 端口占用
bin/admq mqtt admin ctl vm io          # IO 信息
bin/admq mqtt admin ctl vm memory      # 内存使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 集群管理

# 查看集群状态
bin/admq mqtt admin ctl cluster status
bin/admq mqtt admin ctl cluster status --json    # JSON 格式输出

# 加入集群（当前节点主动请求加入目标节点所在集群）
bin/admq mqtt admin ctl cluster join emqx@192.168.1.101

# 离开集群（当前节点主动退出）
bin/admq mqtt admin ctl cluster leave

# 强制移除某节点（慎用，可能导致状态不一致）
bin/admq mqtt admin ctl cluster force-leave emqx@192.168.1.102

# 启用自动集群发现（需已在配置中设定发现策略）
bin/admq mqtt admin ctl cluster discovery enable
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 客戶端管理

# 列出所有已连接客户端
bin/admq mqtt admin ctl clients list

# 查看指定客户端详情
bin/admq mqtt admin ctl clients show <ClientId>

# 踢出指定客户端
bin/admq mqtt admin ctl clients kick <ClientId>

# 按用户名查询
bin/admq mqtt admin ctl clients list --username admin

# 按 IP 查询
bin/admq mqtt admin ctl clients list --peerhost 192.168.1.100
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 主題與訂閱

# 列出当前所有主题
bin/admq mqtt admin ctl topics list

# 查看指定主题信息
bin/admq mqtt admin ctl topics show <Topic>

# 列出所有订阅关系
bin/admq mqtt admin ctl subscriptions list

# 列出共享订阅
bin/admq mqtt admin ctl subscriptions list --shared

# 查看某客户端的订阅
bin/admq mqtt admin ctl subscriptions show <ClientId>

# 添加订阅（以某客户端身份）
bin/admq mqtt admin ctl subscriptions add <ClientId> <Topic> [QoS]

# 删除订阅
bin/admq mqtt admin ctl subscriptions del <ClientId> <Topic>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 監聽器管理

# 列出所有监听器
bin/admq mqtt admin ctl listeners
# 或
bin/admq mqtt admin ctl listeners list

# 查看某监听器详情
bin/admq mqtt admin ctl listeners show mqtt:tcp:0.0.0.0:1883

# 停止某监听器
bin/admq mqtt admin ctl listeners stop mqtt:tcp 0.0.0.0:1883

# 启动某监听器
bin/admq mqtt admin ctl listeners start mqtt:tcp 0.0.0.0:1883
1
2
3
4
5
6
7
8
9
10
11
12
13

默認端口速查：

協議 默認端口
MQTT TCP 1883
MQTT SSL 8883
MQTT WebSocket 8083
MQTT WSS 8084
Dashboard 18083

# 日誌管理

# 查看日志级别
bin/admq mqtt admin ctl log level
bin/admq mqtt admin ctl log primary-level

# 设置日志级别
bin/admq mqtt admin ctl log set-level debug

# 列出所有日志处理器
bin/admq mqtt admin ctl log handlers list

# 设置某个 handler 的日志级别
bin/admq mqtt admin ctl log handlers set-level console debug
1
2
3
4
5
6
7
8
9
10
11
12

# 消息追蹤

# 先调低日志级别
bin/admq mqtt admin ctl log primary-level debug

# 追踪指定客户端
bin/admq mqtt admin ctl trace start client <ClientId> trace.log debug

# 追踪指定主题
bin/admq mqtt admin ctl trace start topic <Topic> trace.log info

# 列出所有追踪
bin/admq mqtt admin ctl trace list

# 停止追踪
bin/admq mqtt admin ctl trace stop client <ClientId>
bin/admq mqtt admin ctl trace stop topic <Topic>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 保留消息

# 查看保留消息数量
bin/admq mqtt admin ctl retainer info

# 列出所有含保留消息的主题
bin/admq mqtt admin ctl retainer topics

# 查看某主题的保留消息
bin/admq mqtt admin ctl retainer show <Topic>

# 清除某主题的保留消息
bin/admq mqtt admin ctl retainer clean <Topic>

# 清除所有保留消息
bin/admq mqtt admin ctl retainer clean

# 重建保留消息索引
bin/admq mqtt admin ctl retainer reindex
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 排它訂閱

# 列出所有排它订阅
bin/admq mqtt admin ctl exclusive list

# 删除排它订阅
bin/admq mqtt admin ctl exclusive delete <Topic>
1
2
3
4
5

# 數據導入導出

# 导出全量数据
bin/admq mqtt admin ctl data export --dir /tmp

# 导出指定配置
bin/admq mqtt admin ctl data export --root-keys listeners,connectors,actions,rule_engine --dir /tmp

# 从备份文件导入
bin/admq mqtt admin ctl data import /tmp/emqx-export-2026-07-01-12-00-00.tar.gz
1
2
3
4
5
6
7
8

導出內容包括：集群配置、內置數據庫、數據目錄附加文件（如 SSL 證書）。

# 節點疏散與負載均衡

# 启动节点疏散（迁移连接到其他节点）
bin/admq mqtt admin ctl rebalance start --evacuation \
    --wait-health-check 60 \
    --redirect-to "Host1:Port1 Host2:Port2" \
    --conn-evict-rate 500 \
    --migrate-to "emqx2@192.168.1.102 emqx3@192.168.1.103" \
    --wait-takeover 200 \
    --sess-evict-rate 500

# 查看疏散状态
bin/admq mqtt admin ctl rebalance status

# 停止疏散
bin/admq mqtt admin ctl rebalance stop

# 启动集群负载均衡
bin/admq mqtt admin ctl rebalance start \
    --conn-evict-rate 100 \
    --sess-evict-rate 100

# 查看持久存储状态
bin/admq mqtt admin ctl ds info

# 设置副本站点
bin/admq mqtt admin ctl ds set-replicas <storage> <site1> <site2> ...
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

# 觀察器

# 启动观察器（实时监控节点活动）
bin/admq mqtt admin ctl observer status
1
2

# 常用 REST API 補充

CLI 不方便批量操作時，可配合 REST API：

# 基础信息
# Base URL: http://localhost:18083/api/v5
# 认证: Basic Auth 或 API Key

# 列出客户端
curl -s -u admin:public http://localhost:18083/api/v5/clients | jq .

# 踢出客户端
curl -X DELETE -u admin:public http://localhost:18083/api/v5/clients/<ClientId>

# 发布消息
curl -X POST -u admin:public \
  -H "Content-Type: application/json" \
  http://localhost:18083/api/v5/publish \
  -d '{"topic":"test/topic","payload":"Hello","qos":1,"retain":false}'

# 列出主题
curl -s -u admin:public http://localhost:18083/api/v5/topics | jq .

# 获取统计信息
curl -s -u admin:public http://localhost:18083/api/v5/stats | jq .

# Prometheus 指标
curl http://localhost:18083/api/v5/prometheus/stats
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 排障速查

問題 排查命令
節點未啟動 bin/admq mqtt admin ctl status
集群狀態異常 bin/admq mqtt admin ctl cluster status
連接不上 bin/admq mqtt admin ctl listeners 檢查端口
客戶端異常 bin/admq mqtt admin ctl clients show <ClientId>
消息丟失 bin/admq mqtt admin ctl trace start client <ClientId> trace.log debug
內存飆升 bin/admq mqtt admin ctl vm memory
端口衝突 bin/admq mqtt admin ctl vm ports / ss -tlnp \| grep 1883
日誌排查 tail -f /var/log/emqx/emqx.log.*

# 附錄：命令速查表

bin/admq-daemon start/stop mqtt server
bin/admq mqtt admin ctl status
bin/admq mqtt admin ctl broker / broker stats / broker metrics
bin/admq mqtt admin ctl vm / vm memory / vm ports / vm io
bin/admq mqtt admin ctl cluster status / join / leave / force-leave
bin/admq mqtt admin ctl clients list / show / kick
bin/admq mqtt admin ctl topics list / show
bin/admq mqtt admin ctl subscriptions list / show / add / del
bin/admq mqtt admin ctl listeners / list / show / start / stop
bin/admq mqtt admin ctl log level / set-level / handlers list / handlers set-level
bin/admq mqtt admin ctl trace start client|topic / stop / list
bin/admq mqtt admin ctl retainer info / topics / show / clean / reindex
bin/admq mqtt admin ctl exclusive list / delete
bin/admq mqtt admin ctl data export / import
bin/admq mqtt admin ctl license info / update
bin/admq mqtt admin ctl rebalance start / stop / status
bin/admq mqtt admin ctl ds info / set-replicas / join / leave / forget
bin/admq mqtt admin ctl observer status
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
编辑页面 (opens new window)
#消息引擎用戶手冊

← 開發手冊→