Apache RocketMQ开发者指南

消息轨迹

1. 消息轨迹数据关键属性

| Producer端 | Consumer端 | Broker端 | | —————- | —————– | ———— | | 生产实例信息 | 消费实例信息 | 消息的Topic | | 发送消息时间 | 投递时间,投递轮次 | 消息存储位置 | | 消息是否发送成功 | 消息是否消费成功 | 消息的Key值 | | 发送耗时 | 消费耗时 | 消息的Tag值 |

2. 支持消息轨迹集群部署

2.1 Broker端配置文件

这里贴出Broker端开启消息轨迹特性的properties配置文件内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
storePathRootDir=/data/rocketmq/rootdir-a-m
storePathCommitLog=/data/rocketmq/commitlog-a-m
autoCreateSubscriptionGroup=true
## if msg tracing is open,the flag will be true
traceTopicEnable=true
listenPort=10911
brokerIP1=XX.XX.XX.XX1
namesrvAddr=XX.XX.XX.XX:9876

2.2 普通模式

RocketMQ集群中每一个Broker节点均用于存储Client端收集并发送过来的消息轨迹数据。因此，对于RocketMQ集群中的Broker节点数量并无要求和限制。

2.3 物理IO隔离模式

对于消息轨迹数据量较大的场景，可以在RocketMQ集群中选择其中一个Broker节点专用于存储消息轨迹，使得用户普通的消息数据与消息轨迹数据的物理IO完全隔离，互不影响。在该模式下，RockeMQ集群中至少有两个Broker节点，其中一个Broker节点定义为存储消息轨迹数据的服务端。

2.4 启动开启消息轨迹的Broker

nohup sh mqbroker -c ../conf/2m-noslave/broker-a.properties &

3. 保存消息轨迹的Topic定义

RocketMQ的消息轨迹特性支持两种存储轨迹数据的方式：

3.1 系统级的TraceTopic

在默认情况下，消息轨迹数据是存储于系统级的TraceTopic中(其名称为：RMQ_SYS_TRACE_TOPIC)。该Topic在Broker节点启动时，会自动创建出来（如上所叙，需要在Broker端的配置文件中将traceTopicEnable的开关变量设置为true）。

3.2 用户自定义的TraceTopic

如果用户不准备将消息轨迹的数据存储于系统级的默认TraceTopic，也可以自己定义并创建用户级的Topic来保存轨迹（即为创建普通的Topic用于保存消息轨迹数据）。下面一节会介绍Client客户端的接口如何支持用户自定义的TraceTopic。

4. 支持消息轨迹的Client客户端实践

为了尽可能地减少用户业务系统使用RocketMQ消息轨迹特性的改造工作量，作者在设计时候采用对原来接口增加一个开关参数(enableMsgTrace)来实现消息轨迹是否开启；并新增一个自定义参数(customizedTraceTopic)来实现用户存储消息轨迹数据至自己创建的用户级Topic。

4.1 发送消息时开启消息轨迹

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true);
        producer.setNamesrvAddr("XX.XX.XX.XX1");
        producer.start();
            try {
                {
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                }

            } catch (Exception e) {
                e.printStackTrace();
            }

4.2 订阅消息时开启消息轨迹

1
2
3
4
5
6
7
8
9
10
11
12
13
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true);
        consumer.subscribe("TopicTest", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setConsumeTimestamp("20181109221800");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");

4.3 支持自定义存储消息轨迹Topic

在上面的发送和订阅消息时候分别将DefaultMQProducer和DefaultMQPushConsumer实例的初始化修改为如下即可支持自定义存储消息轨迹Topic。

1
2
3
4
5
6
        ##其中Topic_test11111需要用户自己预先创建，来保存消息轨迹；
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName",true,"Topic_test11111");
        ......

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true,"Topic_test11111");
        ......

权限控制

1.权限控制特性介绍

权限控制（ACL）主要为RocketMQ提供Topic资源级别的用户访问控制。用户在使用RocketMQ权限控制时，可以在Client客户端通过 RPCHook注入AccessKey和SecretKey签名；同时，将对应的权限控制属性（包括Topic访问权限、IP白名单和AccessKey和SecretKey签名等）设置在distribution/conf/plain_acl.yml的配置文件中。Broker端对AccessKey所拥有的权限进行校验，校验不过，抛出异常； ACL客户端可以参考：org.apache.rocketmq.example.simple包下面的AclClient代码。

2. 权限控制的定义与属性值

2.1权限定义

对RocketMQ的Topic资源访问权限控制定义主要如下表所示，分为以下四种

2.2 权限定义的关键属性

| 字段 | 取值 | 含义 | | ————————– | —————————- | ———————– | | globalWhiteRemoteAddresses | *;192.168.*.*;192.168.0.1 | 全局IP白名单 | | accessKey | 字符串 | Access Key | | secretKey | 字符串 | Secret Key | | whiteRemoteAddress | *;192.168.*.*;192.168.0.1 | 用户IP白名单 | | admin | true;false | 是否管理员账户 | | defaultTopicPerm | DENY;PUB;SUB;PUB|SUB | 默认的Topic权限 | | defaultGroupPerm | DENY;PUB;SUB;PUB|SUB | 默认的ConsumerGroup权限 | | topicPerms | topic=权限 | 各个Topic的权限 | | groupPerms | group=权限 | 各个ConsumerGroup的权限 |

具体可以参考distribution/conf/plain_acl.yml配置文件

3. 支持权限控制的集群部署

在distribution/conf/plain_acl.yml配置文件中按照上述说明定义好权限属性后，打开aclEnable开关变量即可开启RocketMQ集群的ACL特性。这里贴出Broker端开启ACL特性的properties配置文件内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
storePathRootDir=/data/rocketmq/rootdir-a-m
storePathCommitLog=/data/rocketmq/commitlog-a-m
autoCreateSubscriptionGroup=true
## if acl is open,the flag will be true
aclEnable=true
listenPort=10911
brokerIP1=XX.XX.XX.XX1
namesrvAddr=XX.XX.XX.XX:9876

4. 权限控制主要流程

ACL主要流程分为两部分，主要包括权限解析和权限校验。

4.1 权限解析

Broker端对客户端的RequestCommand请求进行解析，拿到需要鉴权的属性字段。 主要包括： （1）AccessKey：类似于用户名，代指用户主体，权限数据与之对应； （2）Signature：客户根据 SecretKey 签名得到的串，服务端再用SecretKey进行签名验证；

4.2 权限校验

Broker端对权限的校验逻辑主要分为以下几步： （1）检查是否命中全局 IP 白名单；如果是，则认为校验通过；否则走 2； （2）检查是否命中用户 IP 白名单；如果是，则认为校验通过；否则走 3； （3）校验签名，校验不通过，抛出异常；校验通过，则走 4； （4）对用户请求所需的权限 和 用户所拥有的权限进行校验；不通过，抛出异常； 用户所需权限的校验需要注意已下内容： （1）特殊的请求例如 UPDATE_AND_CREATE_TOPIC 等，只能由 admin 账户进行操作； （2）对于某个资源，如果有显性配置权限，则采用配置的权限；如果没有显性配置权限，则采用默认的权限；

5. 热加载修改后权限控制定义

RocketMQ的权限控制存储的默认实现是基于yml配置文件。用户可以动态修改权限控制定义的属性，而不需重新启动Broker服务节点。

6. 权限控制的使用限制

(1)如果ACL与高可用部署(Master/Slave架构)同时启用，那么需要在Broker Master节点的distribution/conf/plain_acl.yml配置文件中 设置全局白名单信息，即为将Slave节点的ip地址设置至Master节点plain_acl.yml配置文件的全局白名单中。

(2)如果ACL与高可用部署(多副本Dledger架构)同时启用，由于出现节点宕机时，Dledger Group组内会自动选主，那么就需要将Dledger Group组 内所有Broker节点的plain_acl.yml配置文件的白名单设置所有Broker节点的ip地址。

7. ACL mqadmin配置管理命令

7.1 更新ACL配置文件中“account”的属性值

该命令的示例如下：

sh mqadmin updateAclConfig -n 192.168.1.2:9876 -b 192.168.12.134:10911 -a RocketMQ -s 1234567809123 -t topicA=DENY,topicD=SUB -g groupD=DENY,groupB=SUB

说明：如果不存在则会在ACL Config YAML配置文件中创建；若存在，则会更新对应的“accounts”的属性值; 如果指定的是集群名称，则会在集群中各个broker节点执行该命令；否则会在单个broker节点执行该命令。

7.2 删除ACL配置文件里面的对应“account”

该命令的示例如下：

sh mqadmin deleteAccessConfig -n 192.168.1.2:9876 -c DefaultCluster -a RocketMQ

说明：如果指定的是集群名称，则会在集群中各个broker节点执行该命令；否则会在单个broker节点执行该命令。 其中，参数”a”为Access Key的值，用以标识唯一账户id，因此该命令的参数中指定账户id即可。

7.3 更新ACL配置文件里面中的全局白名单

该命令的示例如下：

sh mqadmin updateGlobalWhiteAddr -n 192.168.1.2:9876 -b 192.168.12.134:10911 -g 10.10.154.1,10.10.154.2

说明：如果指定的是集群名称，则会在集群中各个broker节点执行该命令；否则会在单个broker节点执行该命令。 其中，参数”g”为全局IP白名的值，用以更新ACL配置文件中的“globalWhiteRemoteAddresses”字段的属性值。

7.4 查询集群/Broker的ACL配置文件版本信息

该命令的示例如下：

sh mqadmin clusterAclConfigVersion -n 192.168.1.2:9876 -c DefaultCluster

说明：如果指定的是集群名称，则会在集群中各个broker节点执行该命令；否则会在单个broker节点执行该命令。

7.5 查询集群/Broker的ACL配置文件全部内容

该命令的示例如下：

sh mqadmin getAccessConfigSubCommand -n 192.168.1.2:9876 -c DefaultCluster

说明：如果指定的是集群名称，则会在集群中各个broker节点执行该命令；否则会在单个broker节点执行该命令。

特别注意开启Acl鉴权认证后导致Master/Slave和Dledger模式下Broker同步数据异常的问题， 在社区[4.5.1]版本中已经修复，具体的PR链接为：https://github.com/apache/rocketmq/pull/1149；

Dledger快速搭建

前言

该文档主要介绍如何快速构建和部署基于 DLedger 的可以自动容灾切换的 RocketMQ 集群。

详细的新集群部署和旧集群升级指南请参考 部署指南。

1. 源码构建

构建分为两个部分，需要先构建 DLedger，然后 构建 RocketMQ

1.1 构建 DLedger

git clone https://github.com/openmessaging/openmessaging-storage-dledger.git

cd openmessaging-storage-dledger

mvn clean install -DskipTests

1.2 构建 RocketMQ

git clone https://github.com/apache/rocketmq.git

cd rocketmq

git checkout -b store_with_dledger origin/store_with_dledger

mvn -Prelease-all -DskipTests clean install -U

2. 快速部署

在构建成功后

cd distribution/target/apache-rocketmq

sh bin/dledger/fast-try.sh start

如果上面的步骤执行成功，可以通过 mqadmin 运维命令查看集群状态。

sh bin/mqadmin clusterList -n 127.0.0.1:9876

顺利的话，会看到如下内容：

（BID 为 0 的表示 Master，其余都是 Follower）

启动成功，现在可以向集群收发消息，并进行容灾切换测试了。

关闭快速集群，可以执行：

sh bin/dledger/fast-try.sh stop

快速部署，默认配置在 conf/dledger 里面，默认的存储路径在 /tmp/rmqstore。

3. 容灾切换

部署成功，杀掉 Leader 之后（在上面的例子中，杀掉端口 30931 所在的进程），等待约 10s 左右，用 clusterList 命令查看集群，就会发现 Leader 切换到另一个节点了。

Dledger集群搭建

前言

该文档主要介绍如何部署自动容灾切换的 RocketMQ-on-DLedger Group。

RocketMQ-on-DLedger Group 是指一组相同名称的 Broker，至少需要 3 个节点，通过 Raft 自动选举出一个 Leader，其余节点 作为 Follower，并在 Leader 和 Follower 之间复制数据以保证高可用。
RocketMQ-on-DLedger Group 能自动容灾切换，并保证数据一致。
RocketMQ-on-DLedger Group 是可以水平扩展的，也即可以部署任意多个 RocketMQ-on-DLedger Group 同时对外提供服务。

1. 新集群部署

1.1 编写配置

每个 RocketMQ-on-DLedger Group 至少准备三台机器（本文假设为 3）。
编写 3 个配置文件，建议参考 conf/dledger 目录下的配置文件样例。
关键配置介绍：

这里贴出 conf/dledger/broker-n0.conf 的配置举例。

1
2
3
4
5
6
7
8
9
10
11
12
brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=30911
namesrvAddr=127.0.0.1:9876
storePathRootDir=/tmp/rmqstore/node00
storePathCommitLog=/tmp/rmqstore/node00/commitlog
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-127.0.0.1:40911;n1-127.0.0.1:40912;n2-127.0.0.1:40913
## must be unique
dLegerSelfId=n0
sendMessageThreadPoolNums=16

1.2 启动 Broker

与老版本的启动方式一致。

nohup sh bin/mqbroker -c conf/dledger/xxx-n0.conf &
nohup sh bin/mqbroker -c conf/dledger/xxx-n1.conf &
nohup sh bin/mqbroker -c conf/dledger/xxx-n2.conf &

2. 旧集群升级

如果旧集群采用 Master 方式部署，则每个 Master 都需要转换成一个 RocketMQ-on-DLedger Group。
如果旧集群采用 Master-Slave 方式部署，则每个 Master-Slave 组都需要转换成一个 RocketMQ-on-DLedger Group。

2.1 杀掉旧的 Broker

可以通过 kill 命令来完成，也可以调用 bin/mqshutdown broker。

2.2 检查旧的 Commitlog

RocketMQ-on-DLedger 组中的每个节点，可以兼容旧的 Commitlog ，但其 Raft 复制过程，只能针对新增加的消息。因此，为了避免出现异常，需要保证 旧的 Commitlog 是一致的。
如果旧的集群是采用 Master-Slave 方式部署，有可能在shutdown时，其数据并不是一致的，建议通过md5sum 的方式，检查最近的最少 2 个 Commmitlog 文件，如果发现不一致，则通过拷贝的方式进行对齐。

虽然 RocketMQ-on-DLedger Group 也可以以 2 节点方式部署，但其会丧失容灾切换能力（2n + 1 原则，至少需要3个节点才能容忍其中 1 个宕机）。
所以在对齐了 Master 和 Slave 的 Commitlog 之后，还需要准备第 3 台机器，并把旧的 Commitlog 从 Master 拷贝到 第 3 台机器（记得同时拷贝一下 config 文件夹）。

在 3 台机器准备好了之后，旧 Commitlog 文件也保证一致之后，就可以开始走下一步修改配置了。

2.3 修改配置

参考新集群部署。

2.4 重新启动 Broker

参考新集群部署。

• Previous
  RocketMQ-00丨Apache RocketMQ开发者指南_03
• Next
  RocketMQ-00丨Apache RocketMQ开发者指南_05