消息引擎用户手册

# 前言

本文档为金蝶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)
#消息引擎用户手册

← 开发手册→