Core Engine User Manual

# Preface

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

# Intended Audience

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

For more information about ADMQ for RocketMQ products, please refer to the following ADMQ for RocketMQ product manual documentation set:

No. Manual Document Description
1 Kingdee Apusic Distributed Message Queue for RocketMQ Quick Start Guide Briefly introduces how to quickly get started with ADMQ for RocketMQ.
2 Kingdee Apusic Distributed Message Queue for RocketMQ Installation Guide Details how to install ADMQ for RocketMQ on various operating systems, as well as starting and stopping ADMQ for RocketMQ services.
3 Kingdee Apusic Distributed Message Queue for RocketMQ Message Engine User Manual Details the usage, configuration, management, and supporting tools related to the ADMQ for RocketMQ message engine.
4 Kingdee Apusic Distributed Message Queue for RocketMQ Console User Manual Details the usage and operation instructions for the ADMQ for RocketMQ console features.
5 Kingdee Apusic Distributed Message Queue for RocketMQ Developer Guide Details the instructions for developing ADMQ for RocketMQ client applications based on various development languages.
6 Kingdee Apusic Distributed Message Queue for RocketMQ Migration Guide Details the instructions for migrating from RocketMQ to ADMQ for RocketMQ.
7 Kingdee Apusic Distributed Message Queue for RocketMQ Operations Guide Details the monitoring, operations, security hardening, and other operational instructions for ADMQ for RocketMQ.
8 Kingdee Apusic Distributed Message Queue for RocketMQ Performance Optimization Guide Details the performance tuning instructions for ADMQ for RocketMQ.

# Technical Support

ADMQ for RocketMQ provides comprehensive technical support services. You can obtain technical support through the following methods:

  • 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:

  1. Your name

  2. Company information and contact details

  3. Operating system and its version

  4. Product version number

  5. Detailed information such as logs and screenshots of any exceptions or errors

# Core Engine Directory Description

Directory Description
bin Stores startup, shutdown, and management scripts
conf Stores configuration files (e.g., broker.conf, nameserver.properties)
lib Stores RocketMQ dependency packages
logs Stores runtime logs (Broker logs, NameServer logs, etc.)

# NameServer Configuration

# Basic Configuration

Parameter Description Production Environment Recommendation
listenPort NameServer listening port Default 9876
deployServerAddress NameServer address configuration parameter Default 127.0.0.1, can be updated to the corresponding machine address
clusterName Cluster name Default defaultCluster
managerUrl Console access address Fill in according to the actual deployment IP. If there is no console, the default parameter can be left unchanged

# TLS Configuration

Core Function: TLS authentication encryption

Parameter Description Production Environment Recommendation
tls.enable TLS enable parameter Default false
tls.server.keyPath Path to the server certificate key Fill in according to the actual file storage path
tls.server.keyPassword Server certificate key password Fill in according to the actual set password
tls.server.certPath Path to the server certificate Fill in according to the actual file storage path
tls.server.trustCertPath Path to the server trust certificate Fill in according to the actual file storage path
tls.client.keyPath Path to the client key Fill in according to the actual file storage path
tls.client.keyPassword Client certificate key password Fill in according to the actual set password
tls.client.certPath Path to the client certificate Fill in according to the actual file storage path
tls.client.trustCertPath Path to the client trust certificate Fill in according to the actual file storage path

# Broker Configuration

# Basic Configuration

Parameter Description Production Environment Recommendation
brokerClusterName Cluster name Default cluster name defaultCluster
brokerName Broker node name Default node name broker-a
brokerRole Broker role Default ASYNC_MASTER
brokerId Broker ID Set according to the actual deployed broker
brokerIP1 Broker mapped IP Broker mapped IP
listenPort Listening port Default port 10911
haListenPort Master-slave sync listening port Default port 9888
managerUrl Console access address Fill in according to the actual deployment IP. If there is no console, the default parameter can be left unchanged
Parameter Description Production Environment Recommendation
namesrvAddr NameServer connection address NameServer component connection address

# Storage Configuration

Core Function: Storage management

Parameter Description Production Environment Recommendation
storePathRootDir Storage path Set according to actual usage
storePathCommitLog CommitLog storage path Set according to actual usage
storePathConsumerQueue Consumer queue storage path Set according to actual usage
storePathIndex Index storage path Set according to actual usage
log.dir Log storage path Set according to actual usage

# Prometheus Configuration

Core Function: Monitoring and operations

Parameter Description Production Environment Recommendation
enableMultiDispatch Parameter to enable multi-dimensional message dispatch Recommended to enable, set to true
enablemq Parameter to enable multi-dimensional message dispatch Recommended to enable, set to true
metricsPromExporterHost Prometheus monitoring address Default 0.0.0.0
metricsExporterType Monitoring type Default PROM

# Core Concepts

# RocketMQ Architecture

ADMQ for RocketMQ is fully compatible with the RocketMQ core architecture, mainly including the following components:

Component Description
NameServer Lightweight service registration and discovery center, maintains Broker routing information
Broker Message storage and forwarding service node, includes Master and Slave
Producer Producer, responsible for sending messages to Broker
Consumer Consumer, responsible for receiving messages from Broker
Topic Message topic, the classification identifier of messages
MessageQueue Message queue, the shard unit of a Topic

# Message Model

Producer -> Topic -> MessageQueue -> Consumer
1
  1. Producer sends messages to a specified Topic
  2. Topic is divided into multiple MessageQueues distributed across different Brokers
  3. Consumers subscribe to a Topic and pull messages from MessageQueues for consumption
  4. Consumption progress is recorded in the form of Offset

# Common Commands

# Topic Management

# Topic Attributes

Attribute Description
Read/Write Queue Count Number of queues on each Broker
Permission Read/Write permission control
Order Type Whether it is an ordered message
Message Type Normal/Transactional/Scheduled message

# Creating a Topic

Create a Topic named test-topic with 8 read queues and 8 write queues:

# Use mqadmin to create a Topic
bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic -r 8 -w 8

# Parameter description
# -n: NameServer address
# -c: Cluster name
# -t: Topic name
# -r: Number of read queues
# -w: Number of write queues
1
2
3
4
5
6
7
8
9

# Viewing a Topic

# View Topic list
bin/mqadmin topicList -n localhost:9876

# View Topic routing information
bin/mqadmin topicRoute -n localhost:9876 -t test-topic

# View Topic status
bin/mqadmin topicStatus -n localhost:9876 -t test-topic
1
2
3
4
5
6
7
8

# Deleting a Topic

bin/mqadmin deleteTopic -n localhost:9876 -c DefaultCluster -t test-topic
1

# Group Management

# Creating a Group

bin/mqadmin updateSubGroup -n localhost:9876 -b localhost:10911 -g groupName
1

# Viewing a Group

View all subscription group lists and progress:

bin/mqadmin consumerProgress -n localhost:9876
1

View subscription group connections:

bin/mqadmin consumerConnection -n localhost:9876 -g groupName
1

# Deleting a Group

bin/mqadmin deleteSubGroup -n localhost:9876 -b localhost:10911 -g groupName
1

# Message Query

Query messages by key:

bin/mqadmin queryMsgByKey -n localhost:9876 -t test-topic -k test-key
1

Query messages by message ID:

bin/mqadmin queryMsgById -n localhost:9876 -i message-id
1

# Configuration Management

Get broker configuration:

# Configuration management
bin/mqadmin getBrokerConfig -n localhost:9876 -b broker-a:10911
1
2

Update broker configuration:

bin/mqadmin updateBrokerConfig -n localhost:9876 -b broker-a:10911 -k key -v value
1

# Message Trace

# View message trace
bin/mqadmin printMsg -n localhost:9876 -t test-topic -s "2026-01-01#00:00:00:000" -e "2026-01-02#00:00:00:000"
1
2

# MessageQueue

MessageQueue is the logical shard of a Topic. Each Topic is divided into multiple queues on Brokers.

# Queue Allocation Strategy

Strategy Description
Round Robin Messages are sent to queues in sequence
Hash Selects a queue by Key hash
Ordered Selects a queue according to specified rules

# Producer

# Send Methods

Send Method Description Applicable Scenario
Synchronous Send Sends and waits for the server response Scenarios requiring reliability
Asynchronous Send Returns immediately after sending, gets results via callback Scenarios requiring high throughput
One-Way Send Sends without waiting for a response Scenarios with low reliability requirements
Batch Send Sends multiple messages in a batch Scenarios requiring high throughput
Transactional Send Two-phase commit send Scenarios requiring transactional consistency

# Message Properties

Property Description
Topic Message topic
Tag Message tag, used for filtering
Key Business unique identifier
Body Message body
Properties Extended properties

# Message Routing

When sending a message, the producer selects a MessageQueue based on the following rules:

  1. No Key specified: Round robin queue selection
  2. Key specified: Select queue by Key hash (messages with the same Key go to the same queue)
  3. Ordered message: Select queue according to the message queue selector

# Consumer

# Consumption Mode

# Clustering

  • Messages in the same consumer group are consumed only once
  • Messages are consumed by one consumer in the group
  • Applicable scenario: Most business scenarios

# Broadcasting

  • Each message in the same consumer group is consumed by all consumers
  • Applicable scenario: Local cache updates, configuration synchronization

# Consumption Method

Consumption Method Description
Push Mode Broker actively pushes messages to consumers
Pull Mode Consumers actively pull messages

# Consumer Offset

# Offset Types

Type Description
Cluster Offset Consumption progress is stored on the Broker, shared by the cluster
Broadcast Offset Consumption progress is stored locally on the consumer, maintained individually

# Offset Management

# View consumer group consumption progress
bin/mqadmin consumerProgress -n localhost:9876 -g test-group

# Reset consumer offset
bin/mqadmin resetOffsetByTime -n localhost:9876 -g test-group -t test-topic -s "2026-01-01#00:00:00:000"
1
2
3
4
5

# Message Filtering

# Tag Filtering

Consumers specify Tags when subscribing, and only receive matching messages:

// Subscribe to all Tags
consumer.subscribe("test-topic", "*");

// Subscribe to a specific Tag
consumer.subscribe("test-topic", "tag-a");

// Subscribe to multiple Tags (separated by ||)
consumer.subscribe("test-topic", "tag-a || tag-b");
1
2
3
4
5
6
7
8

# SQL92 Filtering

Use SQL92 syntax for complex filtering:

// Subscribe to messages that meet SQL conditions
consumer.subscribe("test-topic", MessageSelector.bySql("age > 18 AND city = 'Beijing'"));
1
2

# Advanced Message Features

# Ordered Messages

# Global Order

All messages are consumed in the order they are sent. Lower performance.

# Partition Order

Messages within the same partition (MessageQueue) are consumed in order. Different partitions are unordered.

// Send an ordered message
Message msg = new Message("order-topic", "tag-a", "order-1", "content".getBytes());
producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        // Select queue by order ID
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);

// Consume ordered messages
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        // Consume in order
        return ConsumeOrderlyStatus.SUCCESS;
    }
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# Transactional Messages

Implement distributed transactions based on two-phase commit:

// Create a transactional producer
TransactionMQProducer producer = new TransactionMQProducer("tx-group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // Execute local transaction
        try {
            // Execute business logic
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // Check back local transaction status
        if (isTransactionSuccess(msg)) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# Scheduled/Delayed Messages

Supports delivering messages at a specified time:

// Send a delayed message (delay level)
Message msg = new Message("delay-topic", "tag-a", "content".getBytes());
// Set delay level (1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h)
msg.setDelayTimeLevel(3);  // Delay 10 seconds
producer.send(msg);

// RocketMQ 5.x supports arbitrary time
msg.setDelayTimeMs(3600000);  // Delay 1 hour
1
2
3
4
5
6
7
8

# Batch Messages

Send messages in batches to improve throughput:

List<Message> messages = new ArrayList<>();
for (int i = 0; i < 100; i++) {
    messages.add(new Message("batch-topic", "tag-a", ("msg-" + i).getBytes()));
}
producer.send(messages);
1
2
3
4
5

# Cluster and High Availability

# Cluster Status

# View Broker status
bin/mqadmin brokerStatus -n localhost:9876 -b broker-a:10911
1
2

# Cluster Operations Commands

# View cluster status
bin/mqadmin clusterList -n localhost:9876

# View Broker status
bin/mqadmin brokerStatus -n localhost:9876 -b broker-a:10911

# View Topic list
bin/mqadmin topicList -n localhost:9876

# View consumer group connections
bin/mqadmin consumerConnection -n localhost:9876 -g test-group

# View producer connections
bin/mqadmin producerConnection -n localhost:9876 -t test-topic -g test-group
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# Security Configuration

# ACL (v1.0 - applicable to 4.x) Configuration

# Configuration File Modification

# conf/plain_acl.yml
globalWhiteRemoteAddresses:
  - 10.*.*.*
  - 192.168.*.*

accounts:
  - accessKey: test-user
    secretKey: test-password
    whiteRemoteAddress:
    admin: false
    defaultTopicPerm: DENY
    defaultGroupPerm: SUB
    topicPerms:
      - test-topic=CREATE+WRITE+READ
    groupPerms:
      - test-group=READ
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# Enabling ACL

# Enable ACL plugin
bin/mqadmin updateBrokerConfig -n localhost:9876 -b broker-a:10911 -k aclEnable -v true
1
2

# ACL (v2.0 - applicable to 5.x) Configuration

# Configuration File Modification

Modify broker.conf:

# Enable authentication
authenticationEnabled = true
authenticationMetadataProvider = org.apache.rocketmq.auth.authentication.provider.LocalAuthenticationMetadataProvider

# Enable authorization
authorizationEnabled = true
authorizationMetadataProvider = org.apache.rocketmq.auth.authorization.provider.LocalAuthorizationMetadataProvider

# Initialize admin user (auto-created on first startup)
initAuthenticationUser = {"username":"rocketmq","password":"12345678"}

# Inter-component authentication credentials (used for Broker master-slave sync, cluster internal communication, etc.)
innerClientAuthenticationCredentials = {"accessKey":"rocketmq","secretKey":"12345678"}
1
2
3
4
5
6
7
8
9
10
11
12
13

After modifying broker.conf, the cluster needs to be restarted.

# Configuring the mqadmin Tool

After enabling ACL, you need to configure the authentication credentials for the mqadmin tool to execute management commands.

Edit the conf/tools.yml file:

# Use initialized admin user credentials
accessKey: rocketmq
secretKey: 12345678
1
2
3

# Creating Business Users and Authorization

Producer:

# Create a producer user
bin/mqadmin createUser -n 127.0.0.1:9876 -c DefaultCluster \
  -u producer_user \
  -p producer123 \
  -t Normal

# Grant Topic send permission
bin/mqadmin createAcl -n 127.0.0.1:9876 -c DefaultCluster \
  -s User:producer_user \
  -r Topic:TestTopic \
  -a Pub \
  -d Allow
1
2
3
4
5
6
7
8
9
10
11
12

Consumer:

# Create a consumer user
bin/mqadmin createUser -n 127.0.0.1:9876 -c DefaultCluster \
  -u consumer_user -p consumer123 -t Normal

# Grant consumer consumption permission
bin/mqadmin createAcl -n 127.0.0.1:9876 -c DefaultCluster \
  -s User:consumer_user -r Topic:TestTopic,Group:TestGroup -a Sub -d Allow
1
2
3
4
5
6
7

# Verifying Configuration

Use a Java client to send a test message:

SessionCredentials credentials = new SessionCredentials("producer_user", "producer123");
StaticSessionCredentialsProvider credentialsProvider = 
    new StaticSessionCredentialsProvider(credentials);

ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
    .setEndpoints("127.0.0.1:10911")
    .setCredentialProvider(credentialsProvider)
    .build();
// ... Create a Producer and send a message
1
2
3
4
5
6
7
8
9

# TLS Configuration

# broker.conf
tls.server.mode=enforcing
tls.server.certPath=/path/to/server.crt
tls.server.keyPath=/path/to/server.key
tls.server.trustCertPath=/path/to/ca.crt

tls.client.keyPath=conf/auth/client.key
tls.client.keyPassword=123456
tls.client.certPath=conf/auth/client.pem
tls.client.trustCertPath=conf/auth/ca.pem
1
2
3
4
5
6
7
8
9
10
编辑页面 (opens new window)
#Message Engine User Manual

← Development Manual→