消息引擎用户手册

# 前言

本文档为金蝶Apusic分布式消息队列for RocketMQ（Apusic Distributed Message Queue for RocketMQ，简称：ADMQ for RocketMQ）消息引擎的用户手册，详细介绍了ADMQ for RocketMQ消息引擎的功能使用、配置方法及管理操作等内容。

# 适用对象

本文档适用于IT信息化业务负责人、研发经理、软件项目经理、软件架构师、运维工程师。

# 相关文档

了解更多ADMQ for RocketMQ产品相关的信息，请参阅以下ADMQ for RocketMQ产品手册文档集：

序号	手册文档	说明
1	金蝶Apusic分布式消息队列for RocketMQ 快速使用手册	简单介绍了如何快速上手使用ADMQ for RocketMQ 。
2	金蝶Apusic分布式消息队列for RocketMQ 安装手册	详细介绍如何在各操作系统上安装ADMQ for RocketMQ，以及ADMQ for RocketMQ服务启停等操作。
3	金蝶Apusic分布式消息队列for RocketMQ 消息引擎用户手册	详细介绍 ADMQ for RocketMQ 消息引擎相关功能的使用、配置、管理及配套工具的使用方法。
4	金蝶Apusic分布式消息队列for RocketMQ 管控台用户手册	详细介绍ADMQ for RocketMQ管控台相关功能的使用和操作说明。
5	金蝶Apusic分布式消息队列for RocketMQ 开发手册	详细介绍基于各开发语言进行ADMQ for RocketMQ客户端应用开发的说明。
6	金蝶Apusic分布式消息队列for RocketMQ 迁移手册	详细介绍从RocketMQ迁移到ADMQ for RocketMQ的说明。
7	金蝶Apusic分布式消息队列for RocketMQ 运维手册	详细介绍ADMQ for RocketMQ的监控、运维、安全加固等运维说明。
8	金蝶Apusic分布式消息队列for RocketMQ 性能优化手册	详细介绍ADMQ for RocketMQ性能调优的说明。

# 技术支持

ADMQ for RocketMQ产品提供全面的技术支持服务，您可以通过以下方式获得技术支持：

网址：www.apusic.com
电话：400-855-5800
邮箱：support@apusic.com
金蝶云社区：https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

您在取得技术支持时，请提供如下信息：

您的姓名
公司信息与联系方式
操作系统及其版本
产品版本号
出现异常及错误的日志、截图等详细信息

# 核心引擎目录说明

目录名	说明
`bin`	存放启动、停止及管理脚本
`conf`	存放配置文件 (如 `broker.conf`,`nameserver.properties`)
`lib`	存放 RoekctMQ 依赖包
`logs`	存放运行日志 (Broker 日志、NameServer 日志等)

# Nameserver 配置说明

# 基础配置（Basic Configuration）

参数	说明	生产环境建议
listenPort	Nameserver的监听端口	默认使用9876
deployServerAddress	Nameserver的地址配置参数	默认使用127.0.0.1，可以更新成对应机器地址
clusterName	集群名称	默认defaultCluster
managerUrl	管控台的接入地址	根据实际部署的地址IP填写，如果没有管控台，默认参数可不修改

# TLS配置

核心作用：tls认证加密

参数	说明	生产环境建议
tls.enable	TLS认证开启参数	默认false
tls.server.keyPath	存放服务器证书密钥的地址	根据实际存放文件地址填写
tls.server.keyPassword	服务器证书密钥密码	根据实际设置的密码填写
tls.server.certPath	存放服务器证书的地址	根据实际存放文件地址填写
tls.server.trustCertPath	存放服务器信任证书的地址	根据实际存放文件地址填写
tls.client.keyPath	存放客户端密钥的地址	根据实际存放文件地址填写
tls.client.keyPassword	客户端证书密钥密码	根据实际设置的密码填写
tls.client.certPath	存放客户端证书的地址	根据实际存放文件地址填写
tls.client.trustCertPath	存放客户端信任证书的地址	根据实际存放文件地址填写

# Broker 配置说明

# 基础配置（Basic Configuration）

参数	说明	生产环境建议
brokerClusterName	集群名称	默认集群名称defaultCluster
brokerName	broker节点名称	默认节点名称broker-a
brokerRole	broker角色	默认ASYNC_MASTER
brokerId	broker id	根据实际部署的broker设置
brokerIP1	broker 映射IP	broker映射的IP
listenPort	监听端口	默认端口10911
haListenPort	主从同步的监听端口	默认端口9888
managerUrl	管控台的接入地址	根据实际部署的地址IP填写，如果没有管控台，默认参数可不修改

# NameServer相关配置

参数	说明	生产环境建议
namesrvAddr	nameserver连接地址	nameserver组件连接地址

# 存储配置

核心作用：存储管理

参数	说明	生产环境建议
storePathRootDir	存储路径	根据实际使用设置
storePathCommitLog	commitlog存储路径	根据实际使用设置
storePathConsumerQueue	消费队列存储路径	根据实际使用设置
storePathIndex	index存储路径	根据实际使用设置
log.dir	日志存储地址	根据实际使用设置

# Promethus配置

核心作用：监控运维

参数	说明	生产环境建议
enableMultiDispatch	是否开启多维度消息分发的参数	建议开启，设置成true
enablemq	是否开启多维度消息分发的参数	建议开启，设置成true
metricsPromExporterHost	promethus监控地址	默认0.0.0.0
metricsExporterType	监控类型	默认PROM

# 核心概念

# RocketMQ 架构

ADMQ for RocketMQ 完全兼容 RocketMQ 核心架构，主要包含以下组件：

组件	说明
NameServer	轻量级服务注册与发现中心，维护 Broker 路由信息
Broker	消息存储和转发服务节点，分 Master 和 Slave
Producer	生产者，负责发送消息到 Broker
Consumer	消费者，负责从 Broker 接收消息
Topic	消息主题，消息的分类标识
MessageQueue	消息队列，Topic 的分片单位

# 消息模型

Producer -> Topic -> MessageQueue -> Consumer

生产者发送消息到指定 Topic
Topic 被划分为多个 MessageQueue，分布在不同 Broker 上
消费者订阅 Topic，从 MessageQueue 拉取消息消费
消费进度以 Offset 形式记录

# 常见命令

# Topic 管理

# Topic 属性

属性	说明
读写队列数	每个 Broker 上的队列数量
权限	读写权限控制
顺序类型	是否顺序消息
消息类型	普通/事务/定时消息

# 创建 Topic

创建一个名为test-topic的主题，包含8个读队列，8个写队列

# 使用 mqadmin 创建 Topic
bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic -r 8 -w 8

# 参数说明
# -n: NameServer 地址
# -c: 集群名称
# -t: Topic 名称
# -r: 读队列数
# -w: 写队列数

1
2
3
4
5
6
7
8
9

# 查看 Topic

# 查看 Topic 列表
bin/mqadmin topicList -n localhost:9876

# 查看 Topic 路由信息
bin/mqadmin topicRoute -n localhost:9876 -t test-topic

# 查看 Topic 状态
bin/mqadmin topicStatus -n localhost:9876 -t test-topic

1
2
3
4
5
6
7
8

# 删除 Topic

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

# Group 管理

# 创建 Group

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

# 查看 Group

查看所有订阅组列表及进度

bin/mqadmin consumerProgress -n localhost:9876

查看订阅组的连接

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

# 删除 Group

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

# 消息查询

根据key查询消息

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

根据消息id查询消息

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

# 配置管理

获取broker配置

# 配置管理
bin/mqadmin getBrokerConfig -n localhost:9876 -b broker-a:10911

1
2

更新broker配置

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

# 消息轨迹

# 查看消息轨迹
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 是 Topic 的逻辑分片，每个 Topic 在 Broker 上划分为多个队列。

# 队列分配策略

策略	说明
轮询	消息依次发送到各队列
哈希	按 Key 哈希选择队列
顺序	按指定规则选择队列

# 生产者（Producer）

# 发送方式

发送方式	说明	适用场景
同步发送	发送后等待服务端响应	需要可靠性的场景
异步发送	发送后立即返回，通过回调获取结果	需要高吞吐的场景
单向发送	发送后不等待响应	可靠性要求低的场景
批量发送	批量发送多条消息	需要高吞吐的场景
事务发送	两阶段提交发送	需要事务一致性的场景

# 消息属性

属性	说明
Topic	消息主题
Tag	消息标签，用于过滤
Key	业务唯一标识
Body	消息体
Properties	扩展属性

# 消息路由

生产者发送消息时，根据以下规则选择 MessageQueue：

未指定 Key：轮询选择队列
指定 Key：按 Key 哈希选择队列（同一 Key 的消息进入同一队列）
顺序消息：按消息队列选择器选择队列

# 消费者（Consumer）

# 消费模式

# 集群消费（Clustering）

同一消费者组内消息仅被消费一次
消息被组内某个消费者消费
适用场景：大部分业务场景

# 广播消费（Broadcasting）

同一消费者组内每条消息都被所有消费者消费
适用场景：本地缓存更新、配置同步

# 消费方式

消费方式	说明
推模式（Push）	Broker 主动推送消息到消费者
拉模式（Pull）	消费者主动拉取消息

# 消费位点（Offset）

# 位点类型

类型	说明
集群位点	消费进度存储在 Broker 上，集群共享
广播位点	消费进度存储在消费者本地，各自维护

# 位点管理

# 查看消费者组消费进度
bin/mqadmin consumerProgress -n localhost:9876 -g test-group

# 重置消费位点
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

# 消息过滤

# Tag 过滤

消费者订阅时指定 Tag，只接收匹配的消息：

// 订阅所有 Tag
consumer.subscribe("test-topic", "*");

// 订阅指定 Tag
consumer.subscribe("test-topic", "tag-a");

// 订阅多个 Tag（用 || 分隔）
consumer.subscribe("test-topic", "tag-a || tag-b");

1
2
3
4
5
6
7
8

# SQL92 过滤

使用 SQL92 语法进行复杂过滤：

// 订阅满足 SQL 条件的消息
consumer.subscribe("test-topic", MessageSelector.bySql("age > 18 AND city = 'Beijing'"));

1
2

# 高级消息特性

# 顺序消息

# 全局顺序

所有消息按发送顺序消费，性能较低。

# 分区顺序

同一分区（MessageQueue）内的消息按顺序消费，不同分区间无序。

// 发送顺序消息
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) {
        // 按订单 ID 选择队列
        Long orderId = (Long) arg;
        long index = orderId % mqs.size();
        return mqs.get((int) index);
    }
}, orderId);

// 消费顺序消息
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        // 按顺序消费
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

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

# 事务消息

基于两阶段提交实现分布式事务：

// 创建事务生产者
TransactionMQProducer producer = new TransactionMQProducer("tx-group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        try {
            // 执行业务逻辑
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查本地事务状态
        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

# 定时/延时消息

支持指定时间投递消息：

// 发送延时消息（延时级别）
Message msg = new Message("delay-topic", "tag-a", "content".getBytes());
// 设置延时级别（1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h）
msg.setDelayTimeLevel(3);  // 延时 10 秒
producer.send(msg);

// RocketMQ 5.x 支持任意时间
msg.setDelayTimeMs(3600000);  // 延时 1 小时

1
2
3
4
5
6
7
8

# 批量消息

批量发送消息提高吞吐：

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

# 集群与高可用

# 集群状态

# 查看 Broker 状态
bin/mqadmin brokerStatus -n localhost:9876 -b broker-a:10911

1
2

# 集群运维命令

# 查看集群状态
bin/mqadmin clusterList -n localhost:9876

# 查看 Broker 状态
bin/mqadmin brokerStatus -n localhost:9876 -b broker-a:10911

# 查看 Topic 列表
bin/mqadmin topicList -n localhost:9876

# 查看消费者组连接
bin/mqadmin consumerConnection -n localhost:9876 -g test-group

# 查看生产者连接
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

# 安全配置

# ACL（1.0-适用4.x版本）配置

# 配置文件修改

# 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

# 启用 ACL

# 启用 ACL 插件
bin/mqadmin updateBrokerConfig -n localhost:9876 -b broker-a:10911 -k aclEnable -v true

1
2

# ACL （2.0-适用5.x版本）配置

# 配置文件修改

修改broker.conf

# 启用认证
authenticationEnabled = true
authenticationMetadataProvider = org.apache.rocketmq.auth.authentication.provider.LocalAuthenticationMetadataProvider

# 启用授权
authorizationEnabled = true
authorizationMetadataProvider = org.apache.rocketmq.auth.authorization.provider.LocalAuthorizationMetadataProvider

# 初始化管理员用户（首次启动自动创建）
initAuthenticationUser = {"username":"rocketmq","password":"12345678"}

# 组件间认证凭证（用于Broker主从同步、集群内部通信等）
innerClientAuthenticationCredentials = {"accessKey":"rocketmq","secretKey":"12345678"}

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

通过修改broker.conf，需要重启集群

# 配置mqadmin工具

启用ACL后，需要配置mqadmin工具的认证凭证才能执行管理命令。

编辑 conf/tools.yml 文件：

# 使用初始化的管理员用户凭证
accessKey: rocketmq
secretKey: 12345678

1
2
3

# 创建业务用户和授权

生产者

# 创建生产者用户
bin/mqadmin createUser -n 127.0.0.1:9876 -c DefaultCluster \
  -u producer_user \
  -p producer123 \
  -t Normal

# 授予Topic发送权限
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

消费者

# 创建消费者用户
bin/mqadmin createUser -n 127.0.0.1:9876 -c DefaultCluster \
  -u consumer_user -p consumer123 -t Normal

# 授予消费者消费权限
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

# 验证配置

使用Java客户端发送一条测试消息：

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();
// ... 创建Producer并发送消息

1
2
3
4
5
6
7
8
9

# TLS 配置

# 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

编辑页面

#消息引擎用户手册

← 管控台用户手册开发手册→

消息引擎用户手册

# 前言

# 适用对象

# 相关文档

# 技术支持

# 核心引擎目录说明

# Nameserver 配置说明

# 基础配置（Basic Configuration）

# TLS配置

# Broker 配置说明

# 基础配置（Basic Configuration）

# NameServer相关配置

# 存储配置

# Promethus配置

# 核心概念

# RocketMQ 架构

# 消息模型

# 常见命令

# Topic 管理

# Topic 属性

# 创建 Topic

# 查看 Topic

# 删除 Topic

# Group 管理

# 创建 Group

# 查看 Group

# 删除 Group

# 消息查询

# 配置管理

# 消息轨迹

# MessageQueue

# 队列分配策略

# 生产者（Producer）

# 发送方式

# 消息属性

# 消息路由

# 消费者（Consumer）

# 消费模式

# 集群消费（Clustering）

# 广播消费（Broadcasting）

# 消费方式

# 消费位点（Offset）

# 位点类型

# 位点管理

# 消息过滤

# Tag 过滤

# SQL92 过滤

# 高级消息特性

# 顺序消息

# 全局顺序

# 分区顺序

# 事务消息

# 定时/延时消息

# 批量消息

# 集群与高可用

# 集群状态

# 集群运维命令

# 安全配置

# ACL（1.0-适用4.x版本） 配置

# 配置文件修改

# 启用 ACL

# ACL （2.0-适用5.x版本）配置

# 配置文件修改

# 配置mqadmin工具

# 创建业务用户和授权

# 验证配置

# TLS 配置

# ACL（1.0-适用4.x版本）配置