Core Engine User Manual

# Preface

This document is the user manual for the Kingdee Apusic Distributed Message Queue for MQTT (Apusic Distributed Message Queue, ADMQ for MQTT) message engine, detailing the functional usage, configuration methods, and management operations of the ADMQ for MQTT message engine.

# Target Audience

This document is intended for IT business leaders, R&D managers, software project managers, software architects, and operations engineers.

To learn more about ADMQ for MQTT, please refer to the following ADMQ for MQTT product documentation set:

No.	Document	Description
1	ADMQ for MQTT Quick Start Guide	Briefly introduces how to quickly get started with ADMQ for MQTT.
2	ADMQ for MQTT Installation Manual	Details how to install ADMQ for MQTT on various operating systems and how to start/stop ADMQ for MQTT services.
3	ADMQ for MQTT Message Engine User Manual	Details the usage, configuration, management, and supporting tools of ADMQ for MQTT message engine features.
4	ADMQ for MQTT Console User Manual	Details the usage and operation instructions of the ADMQ for MQTT console.
5	ADMQ for MQTT Development Manual	Details ADMQ for MQTT client application development based on various programming languages.
6	ADMQ for MQTT Migration Manual	Details the migration from MQTT Broker to ADMQ for MQTT.
7	ADMQ for MQTT Operations Manual	Details ADMQ for MQTT monitoring, operations, security hardening, and other operational tasks.
8	ADMQ for MQTT Performance Tuning Manual	Details ADMQ for MQTT performance tuning guidelines.

# Technical Support

ADMQ for MQTT provides comprehensive technical support services. You can obtain technical support through the following channels:

Website: www.apusic.com
Phone: 400-855-5800
Email: support@apusic.com
Kingdee Cloud Community: https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

When contacting technical support, please provide the following information:

Your name
Company information and contact details
Operating system and its version
Product version number
Detailed information such as logs and screenshots of the exception or error

# Core Concepts

# MQTT Protocol

MQTT (Message Queuing Telemetry Transport) is a lightweight messaging protocol based on the publish/subscribe pattern, designed for low-bandwidth, high-latency, or unreliable networks. ADMQ for MQTT is fully compatible with MQTT 3.1, 3.1.1, and 5.0 protocols.

# Message Model

Publisher ──→ MQTT Broker
              ├── Topic Match ──→ Subscriber A (QoS 0)
              ├── Topic Match ──→ Subscriber B (QoS 1)
              └── Retained Message Store ──→ New subscriber receives immediately

1
2
3
4

The publisher sends a message to a specific Topic
The Broker distributes the message to subscribers based on Topic matching rules
The subscriber receives and processes the message

# Topic

# Topic Format

Topics are the logical classification units for messages in MQTT, using / to separate levels:

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

1
2

# Wildcards

Wildcard	Description	Example
`+`	Matches a single level	`sensor/+/temperature` matches `sensor/living-room/temperature`
`#`	Matches zero or more levels, must be at the end	`sensor/#` matches `sensor/living-room/temperature` and `sensor/bedroom/humidity/data`

# Retained Messages

Retained messages are a special mechanism in MQTT:

When a publisher sends a retained message, the Broker keeps the last retained message for that topic
When a new subscriber subscribes to that topic, it immediately receives this retained message
Commonly used for device status, configuration delivery, etc.
Sending an empty retained message clears the retained message for that topic

# QoS (Quality of Service)

QoS defines the reliability level of message delivery.

QoS Level	Name	Description	Typical Scenario
0	At most once	Message sent once, delivery not guaranteed	High-frequency, loss-tolerant Telemetry data
1	At least once	Message guaranteed delivery, may be duplicated	General business messages
2	Exactly once	Message guaranteed delivery without duplication	Scenarios requiring high consistency

# QoS Description

QoS is the result of negotiation between publisher and subscriber
Messages are delivered at the QoS level specified by the subscriber
QoS 1/2 requires client acknowledgment and consumes more resources

# Session and Will Message

# Clean Session

Clean Session = true: After the client disconnects, the Broker does not retain its subscription relationships and unacknowledged messages
Clean Session = false: The Broker retains subscription relationships, unacknowledged messages, and QoS 1/2 offline messages. The client can resume after reconnecting

# MQTT 5.0 Session

The corresponding concepts in MQTT 5.0 are Clean Start and Session Expiry Interval:

Clean Start: Whether to clean up the previous session upon connection
Session Expiry Interval: Session expiration time

# Will Message

The will message is automatically published by the Broker when a client disconnects abnormally:

The client presets the will topic, will payload, QoS, and Retain when connecting
When the connection is abnormally closed without sending a DISCONNECT, the Broker automatically publishes the will message
Commonly used for device online/offline status notifications

# Authentication and Authorization

# Authenticators

Authenticators define how client identities are verified upon connection:

Authentication Method	Description
Password-Based	Username/password based authentication
JWT	JSON Web Token based authentication
SCRAM	SCRAM challenge-response mechanism

# Data Sources

Data Source	Description
built_in_database	Built-in database
MySQL	MySQL database
PostgreSQL	PostgreSQL database
HTTP	HTTP service

# Authorization Sources

Authorization sources define the data source for ACL rules:

Multiple authorization sources are matched in order of priority
Rules within the same authorization source are matched sequentially; matching stops after the first successful match

# ACL Rules

ACL is used to control client access permissions to MQTT resources:

Attribute	Description
Subject	All, Username, Client ID
Action	Publish, Subscribe, All
Permission	Allow, Deny
Topic	Supports wildcards
QoS Limit	Can restrict allowed QoS levels, `-1` means no restriction
Retain Limit	Can restrict Retain behavior, `-1` means no restriction

# Extension Capabilities

# Rule Engine

The rule engine is used to process messages based on SQL syntax:

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

Rules can trigger the following actions:

Republish to another Topic
Forward to data bridge
Send HTTP request
Write to database

# Data Bridge

Data bridges are used to forward MQTT messages to external systems:

Bridge Type	Description
Kafka	Forward to Kafka Topic
RocketMQ	Forward to RocketMQ Topic
RabbitMQ	Forward to RabbitMQ Exchange/Queue
MySQL/PostgreSQL	Write to relational database
HTTP	Call HTTP API
Webhook	Send Webhook notification

# Topic Rewrite

Topic rewrite rules are used to automatically replace the source topic with a target topic when a client publishes or subscribes:

Field	Description
Action	The action the rule applies to: subscribe, publish, all
Source Topic	Original topic
Destination Topic	Rewritten topic
Regex	Regular expression to match the source topic
Order	Rule priority

Auto-subscribe rules are used to automatically subscribe a client to specified topics upon successful connection:

Field	Description
Topic	Topic to auto-subscribe
QoS	Subscription QoS level
No Local	MQTT 5.0 feature: whether to receive messages published by itself
Retain As Published	MQTT 5.0 feature: whether to retain the Retain flag from publication
Retain Handling	MQTT 5.0 feature: how to handle retained messages upon subscription

# Shared Subscription

Shared subscription is an MQTT 5.0 feature used to distribute messages load-balanced among multiple subscribers:

$share/group1/sensor/+/temperature

Subscribers within the same shared group receive messages in a round-robin fashion.

# Listener

# Listener Types

Type	Port	Description
TCP	1883	Standard MQTT TCP access
SSL	8883	Encrypted MQTT TCP access
WebSocket	8083	WebSocket access
WebSocket Secure	8084	Encrypted WebSocket access
QUIC	14567	QUIC-based access

# Listener Configuration

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

# System Configuration

# Broker Configuration

Parameter	Description	Default
max_packet_size	Maximum packet size	1MB
max_clientid_len	Maximum ClientID length	65535
max_topic_levels	Maximum topic levels	128
max_topic_alias	Maximum topic aliases	65535
retain_available	Whether retained messages are supported	true
wildcard_subscription	Whether wildcard subscriptions are supported	true
shared_subscription	Whether shared subscriptions are supported	true

# Session Configuration

Parameter	Description	Default
session_expiry_interval	Session expiration time	2h
max_subscriptions	Maximum subscriptions per client	infinity
max_inflight	In-flight window size	32
max_mqueue_len	Message queue length	1000

# Log Configuration

Parameter	Description	Default
log.level	Log level	warning
log.file	Log file path	log/admq-mqtt.log
log.rotation_size	Log rotation size	10MB
log.rotation_count	Number of retained log files	5

# Common Commands

# Node Start/Stop

# Start
bin/admq-daemon start mqtt server
# Stop
bin/admq-daemon stop mqtt server

1
2
3
4

# Node Management

# Check node status
bin/admq mqtt admin ctl status
bin/admq mqtt admin ctl status

# Check version and uptime
bin/admq mqtt admin ctl broker
# Output: version / sysdescr / uptime / datetime

# View Broker statistics (connections, sessions, topics, etc.)
bin/admq mqtt admin ctl broker stats

# View Broker metrics (message send/receive, authentication, authorization, etc.)
bin/admq mqtt admin ctl broker metrics

# View node Erlang information
bin/admq mqtt admin ctl vm
bin/admq mqtt admin ctl vm ports       # Port usage
bin/admq mqtt admin ctl vm io          # IO information
bin/admq mqtt admin ctl vm memory      # Memory usage

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# Cluster Management

# Check cluster status
bin/admq mqtt admin ctl cluster status
bin/admq mqtt admin ctl cluster status --json    # JSON format output

# Join cluster (current node actively requests to join the target node's cluster)
bin/admq mqtt admin ctl cluster join emqx@192.168.1.101

# Leave cluster (current node actively leaves)
bin/admq mqtt admin ctl cluster leave

# Force remove a node (use with caution, may cause state inconsistency)
bin/admq mqtt admin ctl cluster force-leave emqx@192.168.1.102

# Enable automatic cluster discovery (discovery strategy must be set in configuration)
bin/admq mqtt admin ctl cluster discovery enable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# Client Management

# List all connected clients
bin/admq mqtt admin ctl clients list

# View details of a specific client
bin/admq mqtt admin ctl clients show <ClientId>

# Kick out a specific client
bin/admq mqtt admin ctl clients kick <ClientId>

# Query by username
bin/admq mqtt admin ctl clients list --username admin

# Query by 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

# Topics and Subscriptions

# List all current topics
bin/admq mqtt admin ctl topics list

# View information about a specific topic
bin/admq mqtt admin ctl topics show <Topic>

# List all subscriptions
bin/admq mqtt admin ctl subscriptions list

# List shared subscriptions
bin/admq mqtt admin ctl subscriptions list --shared

# View subscriptions of a specific client
bin/admq mqtt admin ctl subscriptions show <ClientId>

# Add a subscription (as a specific client)
bin/admq mqtt admin ctl subscriptions add <ClientId> <Topic> [QoS]

# Delete a subscription
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

# Listener Management

# List all listeners
bin/admq mqtt admin ctl listeners
# or
bin/admq mqtt admin ctl listeners list

# View details of a specific listener
bin/admq mqtt admin ctl listeners show mqtt:tcp:0.0.0.0:1883

# Stop a listener
bin/admq mqtt admin ctl listeners stop mqtt:tcp 0.0.0.0:1883

# Start a listener
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

Default Port Quick Reference:

Protocol	Default Port
MQTT TCP	1883
MQTT SSL	8883
MQTT WebSocket	8083
MQTT WSS	8084
Dashboard	18083

# Log Management

# View log level
bin/admq mqtt admin ctl log level
bin/admq mqtt admin ctl log primary-level

# Set log level
bin/admq mqtt admin ctl log set-level debug

# List all log handlers
bin/admq mqtt admin ctl log handlers list

# Set log level for a specific handler
bin/admq mqtt admin ctl log handlers set-level console debug

1
2
3
4
5
6
7
8
9
10
11
12

# Message Tracing

# First lower the log level
bin/admq mqtt admin ctl log primary-level debug

# Trace a specific client
bin/admq mqtt admin ctl trace start client <ClientId> trace.log debug

# Trace a specific topic
bin/admq mqtt admin ctl trace start topic <Topic> trace.log info

# List all traces
bin/admq mqtt admin ctl trace list

# Stop a trace
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

# Retained Messages

# View retained message count
bin/admq mqtt admin ctl retainer info

# List all topics with retained messages
bin/admq mqtt admin ctl retainer topics

# View retained message for a specific topic
bin/admq mqtt admin ctl retainer show <Topic>

# Clear retained message for a specific topic
bin/admq mqtt admin ctl retainer clean <Topic>

# Clear all retained messages
bin/admq mqtt admin ctl retainer clean

# Rebuild retained message index
bin/admq mqtt admin ctl retainer reindex

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# Exclusive Subscriptions

# List all exclusive subscriptions
bin/admq mqtt admin ctl exclusive list

# Delete an exclusive subscription
bin/admq mqtt admin ctl exclusive delete <Topic>

1
2
3
4
5

# Data Import/Export

# Export all data
bin/admq mqtt admin ctl data export --dir /tmp

# Export specific configuration
bin/admq mqtt admin ctl data export --root-keys listeners,connectors,actions,rule_engine --dir /tmp

# Import from backup file
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

Exported content includes: cluster configuration, built-in database, data directory additional files (such as SSL certificates).

# Node Evacuation and Load Balancing

# Start node evacuation (migrate connections to other nodes)
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

# View evacuation status
bin/admq mqtt admin ctl rebalance status

# Stop evacuation
bin/admq mqtt admin ctl rebalance stop

# Start cluster load balancing
bin/admq mqtt admin ctl rebalance start \
    --conn-evict-rate 100 \
    --sess-evict-rate 100

# View persistent storage status
bin/admq mqtt admin ctl ds info

# Set replica sites
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

# Observer

# Start observer (real-time monitoring of node activity)
bin/admq mqtt admin ctl observer status

1
2

# Common REST API Supplement

When CLI is inconvenient for batch operations, REST API can be used:

# Basic Information
# Base URL: http://localhost:18083/api/v5
# Authentication: Basic Auth or API Key

# List clients
curl -s -u admin:public http://localhost:18083/api/v5/clients | jq .

# Kick out a client
curl -X DELETE -u admin:public http://localhost:18083/api/v5/clients/<ClientId>

# Publish a message
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}'

# List topics
curl -s -u admin:public http://localhost:18083/api/v5/topics | jq .

# Get statistics
curl -s -u admin:public http://localhost:18083/api/v5/stats | jq .

# Prometheus metrics
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

# Troubleshooting Quick Reference

Issue	Troubleshooting Command
Node not started	`bin/admq mqtt admin ctl status`
Cluster status abnormal	`bin/admq mqtt admin ctl cluster status`
Cannot connect	`bin/admq mqtt admin ctl listeners` to check ports
Client abnormal	`bin/admq mqtt admin ctl clients show <ClientId>`
Message loss	`bin/admq mqtt admin ctl trace start client <ClientId> trace.log debug`
High memory usage	`bin/admq mqtt admin ctl vm memory`
Port conflict	`bin/admq mqtt admin ctl vm ports` / `ss -tlnp \| grep 1883`
Log inspection	`tail -f /var/log/emqx/emqx.log.*`

# Appendix: Command Quick Reference

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

编辑页面

#Message Engine User Manual

← Console User Manual Development Manual→

Core Engine User Manual

# Preface

# Target Audience

# Related Documents

# Technical Support

# Core Concepts

# MQTT Protocol

# Message Model

# Topic

# Topic Format

# Wildcards

# Retained Messages

# QoS (Quality of Service)

# QoS Description

# Session and Will Message

# Clean Session

# MQTT 5.0 Session

# Will Message

# Authentication and Authorization

# Authenticators

# Data Sources

# Authorization Sources

# ACL Rules

# Extension Capabilities

# Rule Engine

# Data Bridge

# Topic Rewrite

# Auto Subscribe

# Shared Subscription

# Listener

# Listener Types

# Listener Configuration

# System Configuration

# Broker Configuration

# Session Configuration

# Log Configuration

# Common Commands

# Node Start/Stop

# Node Management

# Cluster Management

# Client Management

# Topics and Subscriptions

# Listener Management

# Log Management

# Message Tracing

# Retained Messages

# Exclusive Subscriptions

# Data Import/Export

# Node Evacuation and Load Balancing

# Observer

# Common REST API Supplement

# Troubleshooting Quick Reference

# Appendix: Command Quick Reference