消息队列-rocketMQ

概念

RocketMQ 的基本概念

1 消息（Message）

消息是指，消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主
题。

2 主题（Topic）

Topic表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行
消息订阅的基本单位。 topic:message 1:n message:topic 1:1
一个生产者可以同时发送多种Topic的消息；而一个消费者只对某种特定的Topic感兴趣，即只可以订阅
和消费一种Topic的消息。 producer:topic 1:n consumer:topic 1:1

3 标签（Tag）

为消息设置的标签，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业
务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提
供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。
Topic是消息的一级分类，Tag是消息的二级分类。
Topic：货物
tag=上海
tag=江苏
tag=浙江
——- 消费者 —–
topic=货物 tag = 上海
topic=货物 tag = 上海|浙江
topic=货物 tag = *

4 队列（Queue）

存储消息的物理实体。一个Topic中可以包含多个Queue，每个Queue中存放的就是该Topic的消息。一
个Topic的Queue也被称为一个Topic中消息的分区（Partition）。
一个Topic的Queue中的消息只能被一个消费者组中的一个消费者消费。一个Queue中的消息不允许同
一个消费者组中的多个消费者同时消费。

在学习参考其它相关资料时，还会看到一个概念：分片（Sharding）。分片不同于分区。在RocketMQ
中，分片指的是存放相应Topic的Broker。每个分片中会创建出相应数量的分区，即Queue，每个
Queue的大小都是相同的。

5 消息标识（MessageId/Key）

RocketMQ中每个消息拥有唯一的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。
不过需要注意的是，MessageId有两个：在生产者send()消息时会自动生成一个MessageId（msgId)，
当消息到达Broker后，Broker也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId与key都
称为消息标识。

msgId：由producer端生成，其生成规则为：
producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode +
当前时间 + AutomicInteger自增计数器
offsetMsgId：由broker端生成，其生成规则为： brokerIp + 物理分区的offset（Queue中的
偏移量）
key：由用户指定的业务相关的唯一标识

系统架构

RocketMQ 的系统架构

RocketMQ架构上主要分为四部分构成：

1 Producer

消息生产者，负责生产消息。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投
递，投递的过程支持快速失败并且低延迟。

例如，业务系统产生的日志写入到MQ的过程，就是消息生产的过程
再如，电商平台中用户提交的秒杀请求写入到MQ的过程，就是消息生产的过程

RocketMQ中的消息生产者都是以生产者组（Producer Group）的形式出现的。生产者组是同一类生产
者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。

2 Consumer

消息消费者，负责消费消息。一个消息消费者会从Broker服务器中获取到消息，并对消息进行相关业务
处理。

例如，QoS系统从MQ中读取日志，并对日志进行解析处理的过程就是消息消费的过程。
再如，电商平台的业务系统从MQ中读取到秒杀请求，并对请求进行处理的过程就是消息消费的
过程。

RocketMQ中的消息消费者都是以消费者组（Consumer Group）的形式出现的。消费者组是同一类消
费者的集合，这类Consumer消费的是同一个Topic类型的消息。消费者组使得在消息消费方面，实现
负载均衡（将一个Topic中的不同的Queue平均分配给同一个Consumer Group的不同的Consumer，注
意，并不是将消息负载均衡）和容错（一个Consmer挂了，该Consumer Group中的其它Consumer可
以接着消费原Consumer消费的Queue）的目标变得非常容易。

消费者组中Consumer的数量应该小于等于订阅Topic的Queue数量。如果超出Queue数量，则多出的
Consumer将不能消费消息。

不过，一个Topic类型的消息可以被多个消费者组同时消费。

注意，
1）消费者组只能消费一个Topic的消息，不能同时消费多个Topic消息
2）一个消费者组中的消费者必须订阅完全相同的Topic

3 Name Server

功能介绍

NameServer是一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现。
RocketMQ的思想来自于Kafka，而Kafka是依赖了Zookeeper的。所以，在RocketMQ的早期版本，即在
MetaQ v1.0与v2.0版本中，也是依赖于Zookeeper的。从MetaQ v3.0，即RocketMQ开始去掉了
Zookeeper依赖，使用了自己的NameServer。

主要包括两个功能：

Broker管理：接受Broker集群的注册信息并且保存下来作为路由信息的基本数据；提供心跳检测
机制，检查Broker是否还存活
路由信息管理：每个NameServer中都保存着Broker集群的整个路由信息和用于客户端查询的队列
信息。Producer和Conumser通过NameServer可以获取整个Broker集群的路由信息，从而进行消
息的投递和消费

路由注册

NameServer通常也是以集群的方式部署，不过，NameServer是无状态的，即NameServer集群中的各
个节点间是无差异的，各节点间相互不进行信息通讯。那各节点中的数据是如何进行数据同步的呢？在
Broker节点启动时，轮询NameServer列表，与每个NameServer节点建立长连接，发起注册请求。在
NameServer内部维护着⼀个Broker列表，用来动态存储Broker的信息。

注意，这是与其它像zk、Eureka、Nacos等注册中心不同的地方。
这种NameServer的无状态方式，有什么优缺点：
优点：NameServer集群搭建简单，扩容简单。
缺点：对于Broker，必须明确指出所有NameServer地址。否则未指出的将不会去注册。也正因
为如此，NameServer并不能随便扩容。因为，若Broker不重新配置，新增的NameServer对于
Broker来说是不可见的，其不会向这个NameServer进行注册。

Broker节点为了证明自己是活着的，为了维护与NameServer间的长连接，会将最新的信息以心跳包的
方式上报给NameServer，每30秒发送一次心跳。心跳包中包含 BrokerId、Broker地址(IP+Port)、
Broker名称、Broker所属集群名称等等。NameServer在接收到心跳包后，会更新心跳时间戳，记录这
个Broker的最新存活时间。

路由剔除

由于Broker关机、宕机或网络抖动等原因，NameServer没有收到Broker的心跳，NameServer可能会将
其从Broker列表中剔除。

NameServer中有⼀个定时任务，每隔10秒就会扫描⼀次Broker表，查看每一个Broker的最新心跳时间
戳距离当前时间是否超过120秒，如果超过，则会判定Broker失效，然后将其从Broker列表中剔除。

扩展：对于RocketMQ日常运维工作，例如Broker升级，需要停掉Broker的工作。OP需要怎么
做？
OP需要将Broker的读写权限禁掉。一旦client(Consumer或Producer)向broker发送请求，都会收
到broker的NO_PERMISSION响应，然后client会进行对其它Broker的重试。
当OP观察到这个Broker没有流量后，再关闭它，实现Broker从NameServer的移除。
OP：运维工程师
SRE：Site Reliability Engineer，现场可靠性工程师

路由发现

RocketMQ的路由发现采用的是Pull模型。当Topic路由信息出现变化时，NameServer不会主动推送给
客户端，而是客户端定时拉取主题最新的路由。默认客户端每30秒会拉取一次最新的路由。

扩展：
1）Push模型：推送模型。其实时性较好，是一个“发布-订阅”模型，需要维护一个长连接。而
长连接的维护是需要资源成本的。该模型适合于的场景：
实时性要求较高
Client数量不多，Server数据变化较频繁
2）Pull模型：拉取模型。存在的问题是，实时性较差。
3）Long Polling模型：长轮询模型。其是对Push与Pull模型的整合，充分利用了这两种模型的优
势，屏蔽了它们的劣势。

客户端NameServer选择策略

这里的客户端指的是Producer与Consumer

客户端在配置时必须要写上NameServer集群的地址，那么客户端到底连接的是哪个NameServer节点
呢？客户端首先会生产一个随机数，然后再与NameServer节点数量取模，此时得到的就是所要连接的
节点索引，然后就会进行连接。如果连接失败，则会采用round-robin策略，逐个尝试着去连接其它节
点。

首先采用的是随机策略进行的选择，失败后采用的是轮询策略。

扩展：Zookeeper Client是如何选择Zookeeper Server的？
简单来说就是，经过两次Shuf􀃦e，然后选择第一台Zookeeper Server。
详细说就是，将配置文件中的zk server地址进行第一次shuf􀃦e，然后随机选择一个。这个选择出
的一般都是一个hostname。然后获取到该hostname对应的所有ip，再对这些ip进行第二次
shuf􀃦e，从shuf􀃦e过的结果中取第一个server地址进行连接。

4 Broker

功能介绍

Broker充当着消息中转角色，负责存储消息、转发消息。Broker在RocketMQ系统中负责接收并存储从
生产者发送来的消息，同时为消费者的拉取请求作准备。Broker同时也存储着消息相关的元数据，包括
消费者组消费进度偏移offset、主题、队列等。

Kafka 0.8版本之后，offset是存放在Broker中的，之前版本是存放在Zookeeper中的。

模块构成

下图为Broker Server的功能模块示意图。

Remoting Module：整个Broker的实体，负责处理来自clients端的请求。而这个Broker实体则由以下模
块构成。
Client Manager：客户端管理器。负责接收、解析客户端(Producer/Consumer)请求，管理客户端。例
如，维护Consumer的Topic订阅信息
Store Service：存储服务。提供方便简单的API接口，处理消息存储到物理硬盘和消息查询功能。
HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
Index Service：索引服务。根据特定的Message key，对投递到Broker的消息进行索引服务，同时也提
供根据Message Key对消息进行快速查询的功能。

集群部署

为了增强Broker性能与吞吐量，Broker一般都是以集群形式出现的。各集群节点中可能存放着相同
Topic的不同Queue。不过，这里有个问题，如果某Broker节点宕机，如何保证数据不丢失呢？其解决
方案是，将每个Broker集群节点进行横向扩展，即将Broker节点再建为一个HA集群，解决单点问题。

Broker节点集群是一个主从集群，即集群中具有Master与Slave两种角色。Master负责处理读写操作请
求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slave则会自动切换为Master去工作。所
以这个Broker集群是主备集群。一个Master可以包含多个Slave，但一个Slave只能隶属于一个Master。
Master与Slave 的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。BrokerId为0表
示Master，非0表示Slave。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信
息到所有NameServer。

5 工作流程

具体流程

1）启动NameServer，NameServer启动后开始监听端口，等待Broker、Producer、Consumer连接。
2）启动Broker时，Broker会与所有的NameServer建立并保持长连接，然后每30秒向NameServer定时
发送心跳包。
3）发送消息前，可以先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，当然，在创
建Topic时也会将Topic与Broker的关系写入到NameServer中。不过，这步是可选的，也可以在发送消
息时自动创建Topic。
4）Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获
取路由信息，即当前发送的Topic消息的Queue与Broker的地址（IP+Port）的映射关系。然后根据算法
策略从队选择一个Queue，与队列所在的Broker建立长连接从而向Broker发消息。当然，在获取到路由
信息后，Producer会首先将路由信息缓存到本地，再每30秒从NameServer更新一次路由信息。
5）Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取其所订阅Topic的路由信息，
然后根据算法策略从路由信息中获取到其所要消费的Queue，然后直接跟Broker建立长连接，开始消费
其中的消息。Consumer在获取到路由信息后，同样也会每30秒从NameServer更新一次路由信息。不过
不同于Producer的是，Consumer还会向Broker发送心跳，以确保Broker的存活状态。

Topic的创建模式

手动创建Topic时，有两种模式：

集群模式：该模式下创建的Topic在该集群中，所有Broker中的Queue数量是相同的。
Broker模式：该模式下创建的Topic在该集群中，每个Broker中的Queue数量可以不同。

自动创建Topic时，默认采用的是Broker模式，会为每个Broker默认创建4个Queue。

读/写队列

从物理上来讲，读/写队列是同一个队列。所以，不存在读/写队列数据同步问题。读/写队列是逻辑上进
行区分的概念。一般情况下，读/写队列数量是相同的。

例如，创建Topic时设置的写队列数量为8，读队列数量为4，此时系统会创建8个Queue，分别是0 1 2 3
456 7。Producer会将消息写入到这8个队列，但Consumer只会消费0 1 2 3这4个队列中的消息，4 5 6
7中的消息是不会被消费到的。

再如，创建Topic时设置的写队列数量为4，读队列数量为8，此时系统会创建8个Queue，分别是0 1 2 3
456 7。Producer会将消息写入到0 1 2 3 这4个队列，但Consumer只会消费0 1 2 3 4 5 6 7这8个队列中
的消息，但是4 5 6 7中是没有消息的。此时假设Consumer Group中包含两个Consuer，Consumer1消
费01 2 3，而Consumer2消费4 5 6 7。但实际情况是，Consumer2是没有消息可消费的。
也就是说，当读/写队列数量设置不同时，总是有问题的。那么，为什么要这样设计呢？
其这样设计的目的是为了，方便Topic的Queue的缩容。

例如，原来创建的Topic中包含16个Queue，如何能够使其Queue缩容为8个，还不会丢失消息？可以动
态修改写队列数量为8，读队列数量不变。此时新的消息只能写入到前8个队列，而消费都消费的却是
16个队列中的数据。当发现后8个Queue中的消息消费完毕后，就可以再将读队列数量动态设置为8。整
个缩容过程，没有丢失任何消息。

perm用于设置对当前创建Topic的操作权限：2表示只写，4表示只读，6表示读写。

代码集成

普通消息

1 消息发送分类

Producer对于消息的发送方式也有多种选择，不同的方式会产生不同的系统效果。

同步发送消息

同步发送消息是指，Producer发出⼀条消息后，会在收到MQ返回的ACK之后才发下⼀条消息。该方式的消息可靠性最高，但消息发送效率太低。

异步发送消息

异步发送消息是指，Producer发出消息后无需等待MQ返回ACK，直接发送下⼀条消息。该方式的消息可靠性可以得到保障，消息发送效率也可以。

单向发送消息

单向发送消息是指，Producer仅负责发送消息，不等待、不处理MQ的ACK。该发送方式时MQ也不返回ACK。该方式的消息发送效率最高，但消息可靠性较差。

2 代码举例

创建工程

创建一个Maven的Java工程rocketmq-test。

导入依赖

导入rocketmq的client依赖。

<properties>
<project.build.sourceEncoding>UTF-
8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.8.0</version>
</dependency>
</dependencies>

定义同步消息发送生产者

public class SyncProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
// 创建一个producer，参数为Producer Group名称
DefaultMQProducer producer = new DefaultMQProducer("pg");
// 指定nameServer地址
producer.setNamesrvAddr("rocketmqOS:9876");
// 设置当发送失败时重试发送的次数，默认为2次
producer.setRetryTimesWhenSendFailed(3);
// 设置发送超时时限为5s，默认3s
producer.setSendMsgTimeout(5000);
// 开启生产者
producer.start();
// 生产并发送100条消息
for (int i = 0; i < 100; i++) {
   
     
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("someTopic", "someTag", body);
// 为消息指定key
msg.setKeys("key-" + i);
// 发送消息
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
// 关闭producer
producer.shutdown();
}
}
// 消息发送的状态
public enum SendStatus {
   
     
SEND_OK, // 发送成功
FLUSH_DISK_TIMEOUT, // 刷盘超时。当Broker设置的刷盘策略为同步刷盘时才可能出
现这种异常状态。异步刷盘不会出现
FLUSH_SLAVE_TIMEOUT, // Slave同步超时。当Broker集群设置的Master-Slave的复
制方式为同步复制时才可能出现这种异常状态。异步复制不会出现
SLAVE_NOT_AVAILABLE, // 没有可用的Slave。当Broker集群设置为Master-Slave的
复制方式为同步复制时才可能出现这种异常状态。异步复制不会出现
}

定义异步消息发送生产者

public class AsyncProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
// 指定异步发送失败后不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed(0);
// 指定新创建的Topic的Queue数量为2，默认为4
producer.setDefaultTopicQueueNums(2);
producer.start();
for (int i = 0; i < 100; i++) {
   
     
byte[] body = ("Hi," + i).getBytes();
try {
   
     
Message msg = new Message("myTopicA", "myTag", body);
// 异步发送。指定回调
producer.send(msg, new SendCallback() {
   
     
// 当producer接收到MQ发送来的ACK后就会触发该回调方法的执行
@Override
public void onSuccess(SendResult sendResult) {
   
     
System.out.println(sendResult);
}
@Override
public void onException(Throwable e) {
   
     
e.printStackTrace();
}
});
} catch (Exception e) {
   
     
e.printStackTrace();
}
} // end-for
// sleep一会儿
// 由于采用的是异步发送，所以若这里不sleep，
// 则消息还未发送就会将producer给关闭，报错
TimeUnit.SECONDS.sleep(3);
producer.shutdown();
}
}

定义单向消息发送生产者

public class OnewayProducer {
   
     
public static void main(String[] args) throws Exception{
   
     
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
for (int i = 0; i < 10; i++) {
   
     
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("single", "someTag", body);
// 单向发送
producer.sendOneway(msg);
}
producer.shutdown();
System.out.println("producer shutdown");
}
}

定义消息消费者

public class SomeConsumer {
   
     
public static void main(String[] args) throws MQClientException {
   
     
// 定义一个pull消费者
// DefaultLitePullConsumer consumer = new
DefaultLitePullConsumer("cg");
// 定义一个push消费者
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("cg");
// 指定nameServer
consumer.setNamesrvAddr("rocketmqOS:9876");
// 指定从第一条消息开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
// 指定消费topic与tag
consumer.subscribe("someTopic", "*");
// 指定采用“广播模式”进行消费，默认为“集群模式”
// consumer.setMessageModel(MessageModel.BROADCASTING);
// 注册消息监听器
consumer.registerMessageListener(new
MessageListenerConcurrently() {
   
     
// 一旦broker中有了其订阅的消息就会触发该方法的执行，
// 其返回值为当前consumer消费的状态
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
   
     
// 逐条消费消息
for (MessageExt msg : msgs) {
   
     
System.out.println(msg);
}
// 返回消费状态：消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 开启消费者消费
consumer.start();
System.out.println("Consumer Started");
}
}

顺序消息

1 什么是顺序消息

顺序消息指的是，严格按照消息的发送顺序进行消费的消息(FIFO)。

默认情况下生产者会把消息以Round Robin轮询方式发送到不同的Queue分区队列；而消费消息时会从多个Queue上拉取消息，这种情况下的发送和消费是不能保证顺序的。如果将消息仅发送到同一个Queue中，消费时也只从这个Queue上拉取消息，就严格保证了消息的顺序性。

2 为什么需要顺序消息

例如，现在有TOPIC ORDER_STATUS (订单状态)，其下有4个Queue队列，该Topic中的不同消息用于描述当前订单的不同状态。假设订单有状态：未支付、已支付、发货中、发货成功、发货失败。

根据以上订单状态，生产者从时序上可以生成如下几个消息：
订单T0000001:未支付 –> 订单T0000001:已支付 –> 订单T0000001:发货中 –> 订单T0000001:发货失败

消息发送到MQ中之后，Queue的选择如果采用轮询策略，消息在MQ的存储可能如下：

这种情况下，我们希望Consumer消费消息的顺序和我们发送是一致的，然而上述MQ的投递和消费方式，我们无法保证顺序是正确的。对于顺序异常的消息，Consumer即使设置有一定的状态容错，也不能完全处理好这么多种随机出现组合情况。

基于上述的情况，可以设计如下方案：对于相同订单号的消息，通过一定的策略，将其放置在一个Queue中，然后消费者再采用一定的策略（例如，一个线程独立处理一个queue，保证处理消息的顺序性），能够保证消费的顺序性。

3 有序性分类

根据有序范围的不同，RocketMQ可以严格地保证两种消息的有序性：分区有序与全局有序。

全局有序

当发送和消费参与的Queue只有一个时所保证的有序是整个Topic中消息的顺序，称为全局有序。

在创建Topic时指定Queue的数量。有三种指定方式：
1）在代码中创建Producer时，可以指定其自动创建的Topic的Queue数量
2）在RocketMQ可视化控制台中手动创建Topic时指定Queue数量
3）使用mqadmin命令手动创建Topic时指定Queue数量

分区有序

如果有多个Queue参与，其仅可保证在该Queue分区队列上的消息顺序，则称为分区有序。

如何实现Queue的选择？在定义Producer时我们可以指定消息队列选择器，而这个选择器是我们自己实现了MessageQueueSelector接口定义的。
在定义选择器的选择算法时，一般需要使用选择key。这个选择key可以是消息key也可以是其它数据。但无论谁做选择key，都不能重复，都是唯一的。

一般性的选择算法是，让选择key（或其hash值）与该Topic所包含的Queue的数量取模，其结果即为选择出的Queue的QueueId。

取模算法存在一个问题：不同选择key与Queue数量取模结果可能会是相同的，即不同选择key的消息可能会出现在相同的Queue，即同一个Consuemr可能会消费到不同选择key的消息。这个问题如何解决？一般性的作法是，从消息中获取到选择key，对其进行判断。若是当前Consumer需要消费的消息，则直接消费，否则，什么也不做。这种做法要求选择key要能够随着消息一起被Consumer获取到。此时使用消息key作为选择key是比较好的做法。

以上做法会不会出现如下新的问题呢？不属于那个Consumer的消息被拉取走了，那么应该消费该消息的Consumer是否还能再消费该消息呢？同一个Queue中的消息不可能被同一个Group中的不同Consumer同时消费。所以，消费现一个Queue的不同选择key的消息的Consumer一定属于不同的Group。而不同的Group中的Consumer间的消费是相互隔离的，互不影响的。

4 代码举例

public class OrderedProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
for (int i = 0; i < 100; i++) {
   
     
Integer orderId = i;
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("TopicA", "TagA", body);
SendResult sendResult = producer.send(msg, new
MessageQueueSelector() {
   
     
@Override
public MessageQueue select(List<MessageQueue> mqs,
Message msg, Object arg) {
   
     
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
System.out.println(sendResult);
}
producer.shutdown();
}
}

延迟消息

1 什么是延时消息

当消息写入到Broker后，在指定的时长后才可被消费处理的消息，称为延时消息。

采用RocketMQ的延时消息可以实现定时任务的功能，而无需使用定时器。典型的应用场景是，电商交易中超时未支付关闭订单的场景，12306平台订票超时未支付取消订票的场景。

在电商平台中，订单创建时会发送一条延迟消息。这条消息将会在30分钟后投递给后台业务系统（Consumer），后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如果未完成，则取消订单，将商品再次放回到库存；如果完成支付，则忽略。
在12306平台中，车票预订成功后就会发送一条延迟消息。这条消息将会在45分钟后投递给后台业务系统（Consumer），后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如果未完成，则取消预订，将车票再次放回到票池；如果完成支付，则忽略。

2 延时等级

延时消息的延迟时长不支持随意时长的延迟，是通过特定的延迟等级来指定的。延时等级定义在RocketMQ服务端的MessageStoreConfig 类中的如下变量中：

即，若指定的延时等级为3，则表示延迟时长为10s，即延迟等级是从1开始计数的。

当然，如果需要自定义的延时等级，可以通过在broker加载的配置中新增如下配置（例如下面增加了1天这个等级1d）。配置文件在RocketMQ安装目录下的conf目录中。

1 2	messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d

3 延时消息实现原理

具体实现方案是：

修改消息

Producer将消息发送到Broker后，Broker会首先将消息写入到commitlog文件，然后需要将其分发到相应的consumequeue。不过，在分发之前，系统会先判断消息中是否带有延时等级。若没有，则直接正常分发；若有则需要经历一个复杂的过程：

修改消息的Topic为SCHEDULE_TOPIC_XXXX
根据延时等级，在consumequeue目录中SCHEDULE_TOPIC_XXXX主题下创建出相应的queueId目录与consumequeue文件（如果没有这些目录与文件的话）。

延迟等级delayLevel与queueId的对应关系为queueId = delayLevel -1
需要注意，在创建queueId目录时，并不是一次性地将所有延迟等级对应的目录全部创建完毕，而是用到哪个延迟等级创建哪个目录

修改消息索引单元内容。索引单元中的Message Tag HashCode部分原本存放的是消息的Tag的Hash值。现修改为消息的投递时间。投递时间是指该消息被重新修改为原Topic后再次被写入到commitlog中的时间。投递时间 = 消息存储时间 + 延时等级时间。消息存储时间指的是消息被发送到Broker时的时间戳。
将消息索引写入到SCHEDULE_TOPIC_XXXX主题下相应的consumequeue中

SCHEDULE_TOPIC_XXXX目录中各个延时等级Queue中的消息是如何排序的？
是按照消息投递时间排序的。一个Broker中同一等级的所有延时消息会被写入到consumequeue目录中SCHEDULE_TOPIC_XXXX目录下相同Queue中。即一个Queue中消息投递时间的延迟等级时间是相同的。那么投递时间就取决于于消息存储时间了。即按照消息被发送到Broker的时间进行排序的。

投递延时消息

Broker内部有⼀个延迟消息服务类ScheuleMessageService，其会消费SCHEDULE_TOPIC_XXXX中的消息，即按照每条消息的投递时间，将延时消息投递到⽬标Topic中。不过，在投递之前会从commitlog中将原来写入的消息再次读出，并将其原来的延时等级设置为0，即原消息变为了一条不延迟的普通消息。然后再次将消息投递到目标Topic中。

ScheuleMessageService在Broker启动时，会创建并启动一个定时器TImer，用于执行相应的定时任务。系统会根据延时等级的个数，定义相应数量的TimerTask，每个TimerTask负责一个延迟等级消息的消费与投递。每个TimerTask都会检测相应Queue队列的第一条消息是否到期。若第一条消息未到期，则后面的所有消息更不会到期（消息是按照投递时间排序的）；若第一条消息到期了，则将该消息投递到目标Topic，即消费该消息。

将消息重新写入commitlog

延迟消息服务类ScheuleMessageService将延迟消息再次发送给了commitlog，并再次形成新的消息索引条目，分发到相应Queue。

这其实就是一次普通消息发送。只不过这次的消息Producer是延迟消息服务类ScheuleMessageService

4 代码举例

定义DelayProducer类

public class DelayProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
for (int i = 0; i < 10; i++) {
   
     
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("TopicB", "someTag", body);
// 指定消息延迟等级为3级，即延迟10s
// msg.setDelayTimeLevel(3);
SendResult sendResult = producer.send(msg);
// 输出消息被发送的时间
System.out.print(new SimpleDateFormat("mm:ss").format(new
Date()));
System.out.println(" ," + sendResult);
}
producer.shutdown();
}
}

定义OtherConsumer类

public class OtherConsumer {
   
     
public static void main(String[] args) throws MQClientException {
   
     
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("cg");
consumer.setNamesrvAddr("rocketmqOS:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
consumer.subscribe("TopicB", "*");
consumer.registerMessageListener(new
MessageListenerConcurrently() {
   
     
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
   
     
for (MessageExt msg : msgs) {
   
     
// 输出消息被消费的时间
System.out.print(new
SimpleDateFormat("mm:ss").format(new Date()));
System.out.println(" ," + msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started");
}
}

事务消息

1 问题引入

这里的一个需求场景是：工行用户A向建行用户B转账1万元。
我们可以使用同步消息来处理该需求场景：

1、工行系统发送一个给B增款1万元的同步消息M给Broker；
2、消息被Broker成功接收后，向工行系统发送成功ACK；
3、工行系统收到成功ACK后从用户A中扣款1万元；
4、建行系统从Broker中获取到消息M；
5、建行系统消费消息M，即向用户B中增加1万元；

这其中是有问题的：若第3步中的扣款操作失败，但消息已经成功发送到了Broker。对于MQ来
说，只要消息写入成功，那么这个消息就可以被消费。此时建行系统中用户B增加了1万元。出
现了数据不一致问题。

2 解决思路

解决思路是，让第1、2、3步具有原子性，要么全部成功，要么全部失败。即消息发送成功后，必须要
保证扣款成功。如果扣款失败，则回滚发送成功的消息。而该思路即使用事务消息。这里要使用分布
式事务解决方案。

注：如果不知道TC/TM可以看下面的XA模式目录部分的内容

使用事务消息来处理该需求场景：

1、事务管理器TM向事务协调器TC发起指令，开启全局事务；
2、工行系统发一个给B增款1万元的事务消息M给TC；
3、 TC会向Broker发送半事务消息prepareHalf，将消息M预提交到Broker此时的建行系统是看；
不到Broker中的消息M的 4、 Broker会将预提交执行结果Report给TC；
5、如果预提交失败，则TC会向TM上报预提交失败的响应，全局事务结束；如果预提交成功，TC会；
调用工行系统的回调操作，去完成工行用户A的预扣款1万元的操作 6、工行系统会向TC发送预扣款执行结果，即本地事务的执行状态；
7、 TC收到预扣款执行结果后，会将结果上报给TM；
预扣款执行结果存在三种可能性：

// 描述本地事务执行状态
public enum LocalTransactionState {
   
     
COMMIT_MESSAGE, // 本地事务执行成功
ROLLBACK_MESSAGE, // 本地事务执行失败
UNKNOW, // 不确定，表示需要进行回查以确定本地事务的执行结果
}

1、 TM会根据上报结果向TC发出不同的确认指令；

若预扣款成功（本地事务状态为COMMIT_MESSAGE），则TM向TC发送Global Commit指令
若预扣款失败（本地事务状态为ROLLBACK_MESSAGE），则TM向TC发送Global Rollback指令
若现未知状态（本地事务状态为UNKNOW），则会触发工行系统的本地事务状态回查操作。回
查操作会将回查结果，即COMMIT_MESSAGE或ROLLBACK_MESSAGE Report给TC。TC将结果上报给TM，TM会再向TC发送最终确认指令Global Commit或Global Rollback

1、 TC在接收到指令后会向Broker与工行系统发出确认指令；

TC接收的若是Global Commit指令，则向Broker与工行系统发送Branch Commit指令。此时
Broker中的消息M才可被建行系统看到；此时的工行用户A中的扣款操作才真正被确认
TC接收到的若是Global Rollback指令，则向Broker与工行系统发送Branch Rollback指令。此时
Broker中的消息M将被撤销；工行用户A中的扣款操作将被回滚

以上方案就是为了确保消息投递与扣款操作能够在一个事务中，要成功都成功，有一个失败，
则全部回滚。
以上方案并不是一个典型的XA模式。因为XA模式中的分支事务是异步的，而事务消息方案中的
消息预提交与预扣款操作间是同步的。

3 基础

分布式事务

对于分布式事务，通俗地说就是，一次操作由若干分支操作组成，这些分支操作分属不同应用，分布在
不同服务器上。分布式事务需要保证这些分支操作要么全部成功，要么全部失败。分布式事务与普通事
务一样，就是为了保证操作结果的一致性。

事务消息

RocketMQ提供了类似X/Open XA的分布式事务功能，通过事务消息能达到分布式事务的最终一致。XA
是一种分布式事务解决方案，一种分布式事务处理模式。

半事务消息

暂不能投递的消息，发送方已经成功地将消息发送到了Broker，但是Broker未收到最终确认指令，此时
该消息被标记成“暂不能投递”状态，即不能被消费者看到。处于该种状态下的消息即半事务消息。

本地事务状态

Producer 回调操作执行的结果为本地事务状态，其会发送给TC，而TC会再发送给TM。TM会根据TC发
送来的本地事务状态来决定全局事务确认指令。

// 描述本地事务执行状态
public enum LocalTransactionState {
   
     
COMMIT_MESSAGE, // 本地事务执行成功
ROLLBACK_MESSAGE, // 本地事务执行失败
UNKNOW, // 不确定，表示需要进行回查以确定本地事务的执行结果
}

消息回查

消息回查，即重新查询本地事务的执行状态。本例就是重新到DB中查看预扣款操作是否执行成功。

注意，消息回查不是重新执行回调操作。回调操作是进行预扣款操作，而消息回查则是查看预
扣款操作执行的结果。

引发消息回查的原因最常见的有两个：
1)回调操作返回UNKNWON
2)TC没有接收到TM的最终全局事务确认指令

RocketMQ中的消息回查设置

关于消息回查，有三个常见的属性设置。它们都在broker加载的配置文件中设置，例如：

transactionTimeout=20，指定TM在20秒内应将最终确认状态发送给TC，否则引发消息回查。默
认为60秒
transactionCheckMax=5，指定最多回查5次，超过后将丢弃消息并记录错误日志。默认15次。
transactionCheckInterval=10，指定设置的多次消息回查的时间间隔为10秒。默认为60秒。

4 XA模式三剑客

XA协议

XA（Unix Transaction）是一种分布式事务解决方案，一种分布式事务处理模式，是基于XA协议的。
XA协议由Tuxedo（Transaction for Unix has been Extended for Distributed Operation，分布式操作扩
展之后的Unix事务系统）首先提出的，并交给X/Open组织，作为资源管理器与事务管理器的接口标
准。 XA模式中有三个重要组件：TC、TM、RM。

Transaction Coordinator，事务协调者。维护全局和分支事务的状态，驱动全局事务提交或回滚。

RocketMQ中Broker充当着TC

Transaction Manager，事务管理器。定义全局事务的范围：开始全局事务、提交或回滚全局事务。它
实际是全局事务的发起者

RocketMQ中事务消息的Producer充当着TM。

Resource Manager，资源管理器。管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事
务的状态，并驱动分支事务提交或回滚。

RocketMQ中事务消息的Producer及Broker均是RM。

5 XA模式架构

XA模式是一个典型的2PC，其执行原理如下：

1、 TM向TC发起指令，开启一个全局事务；
2、根据业务要求，各个RM会逐个向TC注册分支事务，然后TC会逐个向RM发出预执行指令；
3、各个RM在接收到指令后会在进行本地事务预执行；
4、 RM将预执行结果Report给TC当然，这个结果可能是成功，也可能是失败；
5、 TC在接收到各个RM的Report后会将汇总结果上报给TM，根据汇总结果TM会向TC发出确认指；
令。

若所有结果都是成功响应，则向TC发送Global Commit指令。
只要有结果是失败响应，则向TC发送Global Rollback指令

1、 TC在接收到指令后再次向RM发送确认指令；

事务消息方案并不是一个典型的XA模式。因为XA模式中的分支事务是异步的，而事务消息方案
中的消息预提交与预扣款操作间是同步的。

6 注意

事务消息不支持延时消息
对于事务消息要做好幂等性检查，因为事务消息可能不止一次被消费（因为存在回滚后再提交的
情况）

7 代码举例

定义工行事务监听器

public class ICBCTransactionListener implements TransactionListener {
   
     
// 回调操作方法
// 消息预提交成功就会触发该方法的执行，用于完成本地事务
@Override
public LocalTransactionState executeLocalTransaction(Message msg,
Object arg) {
   
     
System.out.println("预提交消息成功：" + msg);
// 假设接收到TAGA的消息就表示扣款操作成功，TAGB的消息表示扣款失败，
// TAGC表示扣款结果不清楚，需要执行消息回查
if (StringUtils.equals("TAGA", msg.getTags())) {
   
     
return LocalTransactionState.COMMIT_MESSAGE;
} else if (StringUtils.equals("TAGB", msg.getTags())) {
   
     
return LocalTransactionState.ROLLBACK_MESSAGE;
} else if (StringUtils.equals("TAGC", msg.getTags())) {
   
     
return LocalTransactionState.UNKNOW;
}
return LocalTransactionState.UNKNOW;
}
// 消息回查方法
// 引发消息回查的原因最常见的有两个：
// 1)回调操作返回UNKNWON
// 2)TC没有接收到TM的最终全局事务确认指令
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
   
     
System.out.println("执行消息回查" + msg.getTags());
return LocalTransactionState.COMMIT_MESSAGE;
}
}

定义事物消息生产者

public class TransactionProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
TransactionMQProducer producer = new
TransactionMQProducer("tpg");
producer.setNamesrvAddr("rocketmqOS:9876");
/**
* 定义一个线程池
* @param corePoolSize 线程池中核心线程数量
* @param maximumPoolSize 线程池中最多线程数
* @param keepAliveTime 这是一个时间。当线程池中线程数量大于核心线程数量
是，
* 多余空闲线程的存活时长
* @param unit 时间单位
* @param workQueue 临时存放任务的队列，其参数就是队列的长度
* @param threadFactory 线程工厂
*/
ExecutorService executorService = new ThreadPoolExecutor(2, 5,
100, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2000), new
ThreadFactory() {
   
     
@Override
public Thread newThread(Runnable r) {
   
     
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
// 为生产者指定一个线程池
producer.setExecutorService(executorService);
// 为生产者添加事务监听器
producer.setTransactionListener(new ICBCTransactionListener());
producer.start();
String[] tags = {
   
     "TAGA","TAGB","TAGC"};
for (int i = 0; i < 3; i++) {
   
     
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("TTopic", tags[i], body);
// 发送事务消息
// 第二个参数用于指定在执行本地事务时要使用的业务参数
SendResult sendResult =
producer.sendMessageInTransaction(msg,null);
System.out.println("发送结果为：" +
sendResult.getSendStatus());
}
}
}

定义消费者

直接使用普通消息的SomeConsumer作为消费者即可。

public class SomeConsumer {
   
     
public static void main(String[] args) throws MQClientException {
   
     
// 定义一个pull消费者
// DefaultLitePullConsumer consumer = new
DefaultLitePullConsumer("cg");
// 定义一个push消费者
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("cg");
// 指定nameServer
consumer.setNamesrvAddr("rocketmqOS:9876");
// 指定从第一条消息开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
// 指定消费topic与tag
consumer.subscribe("TTopic", "*");
// 指定采用“广播模式”进行消费，默认为“集群模式”
// consumer.setMessageModel(MessageModel.BROADCASTING);
// 注册消息监听器
consumer.registerMessageListener(new
MessageListenerConcurrently() {
   
     
// 一旦broker中有了其订阅的消息就会触发该方法的执行，
// 其返回值为当前consumer消费的状态
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
   
     
// 逐条消费消息
for (MessageExt msg : msgs) {
   
     
System.out.println(msg);
}
// 返回消费状态：消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
// 开启消费者消费
consumer.start();
System.out.println("Consumer Started");
}
}

批量消息

1 批量发送消息

发送限制

生产者进行消息发送时可以一次发送多条消息，这可以大大提升Producer的发送效率。不过需要注意以
下几点：

批量发送的消息必须具有相同的Topic
批量发送的消息必须具有相同的刷盘策略
批量发送的消息不能是延时消息与事务消息

批量发送大小

默认情况下，一批发送的消息总大小不能超过4MB字节。如果想超出该值，有两种解决方案：

方案一：将批量消息进行拆分，拆分为若干不大于4M的消息集合分多次批量发送
方案二：在Producer端与Broker端修改属性

** Producer端需要在发送之前设置Producer的maxMessageSize属性 **
** Broker端需要修改其加载的配置文件中的maxMessageSize属性 **

生产者发送的消息大小

生产者通过send()方法发送的Message，并不是直接将Message序列化后发送到网络上的，而是通过这个Message生成了一个字符串发送出去的。这个字符串由四部分构成：Topic、消息Body、消息日志（占20字节），及用于描述消息的一堆属性key-value。这些属性中包含例如生产者地址、生产时间、要发送的QueueId等。最终写入到Broker中消息单元中的数据都是来自于这些属性。

2 批量消费消息

修改批量属性

Consumer的MessageListenerConcurrently监听接口的consumeMessage()方法的第一个参数为消息列表，但默认情况下每次只能消费一条消息。若要使其一次可以消费多条消息，则可以通过修改Consumer的consumeMessageBatchMaxSize属性来指定。不过，该值不能超过32。因为默认情况下消费者每次可以拉取的消息最多是32条。若要修改一次拉取的最大值，则可通过修改Consumer的pullBatchSize属性来指定。

存在的问题

Consumer的pullBatchSize属性与consumeMessageBatchMaxSize属性是否设置的越大越好？当然不是。

pullBatchSize值设置的越大，Consumer每拉取一次需要的时间就会越长，且在网络上传输出现问题的可能性就越高。若在拉取过程中若出现了问题，那么本批次所有消息都需要全部重新拉取。
consumeMessageBatchMaxSize值设置的越大，Consumer的消息并发消费能力越低，且这批被消费的消息具有相同的消费结果。因为consumeMessageBatchMaxSize指定的一批消息只会使用一个线程进行处理，且在处理过程中只要有一个消息处理异常，则这批消息需要全部重新再次消费处理。

3 代码举例

该批量发送的需求是，不修改最大发送4M的默认值，但要防止发送的批量消息超出4M的限制。

定义消息列表分割器

// 消息列表分割器：其只会处理每条消息的大小不超4M的情况。
// 若存在某条消息，其本身大小大于4M，这个分割器无法处理，
// 其直接将这条消息构成一个子列表返回。并没有再进行分割
public class MessageListSplitter implements Iterator<List<Message>> {
   
     
// 指定极限值为4M
private final int SIZE_LIMIT = 4 *1024 * 1024;
// 存放所有要发送的消息
private final List<Message> messages;
// 要进行批量发送消息的小集合起始索引
private int currIndex;
public MessageListSplitter(List<Message> messages) {
   
     
this.messages = messages;
}
@Override
public boolean hasNext() {
   
     
// 判断当前开始遍历的消息索引要小于消息总数
return currIndex < messages.size();
}
@Override
public List<Message> next() {
   
     
int nextIndex = currIndex;
// 记录当前要发送的这一小批次消息列表的大小
int totalSize = 0;
for (; nextIndex < messages.size(); nextIndex++) {
   
     
// 获取当前遍历的消息
Message message = messages.get(nextIndex);
// 统计当前遍历的message的大小
int tmpSize = message.getTopic().length() +
message.getBody().length;
Map<String, String> properties = message.getProperties();
for (Map.Entry<String, String> entry :
properties.entrySet()) {
   
     
tmpSize += entry.getKey().length() +
entry.getValue().length();
}
tmpSize = tmpSize + 20;
// 判断当前消息本身是否大于4M
if (tmpSize > SIZE_LIMIT) {
   
     
if (nextIndex - currIndex == 0) {
   
     
nextIndex++;
}
break;
}
if (tmpSize + totalSize > SIZE_LIMIT) {
   
     
break;
} else {
   
     
totalSize += tmpSize;
}
} // end-for
// 获取当前messages列表的子集合[currIndex, nextIndex)
List<Message> subList = messages.subList(currIndex, nextIndex);
// 下次遍历的开始索引
currIndex = nextIndex;
return subList;
}
}

定义批量消息生产者

public class BatchProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
// 指定要发送的消息的最大大小，默认是4M
// 不过，仅修改该属性是不行的，还需要同时修改broker加载的配置文件中的
// maxMessageSize属性
// producer.setMaxMessageSize(8 * 1024 * 1024);
producer.start();
// 定义要发送的消息集合
List<Message> messages = new ArrayList<>();
for (int i = 0; i < 100; i++) {
   
     
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("someTopic", "someTag", body);
messages.add(msg);
}
// 定义消息列表分割器，将消息列表分割为多个不超出4M大小的小列表
MessageListSplitter splitter = new
MessageListSplitter(messages);
while (splitter.hasNext()) {
   
     
try {
   
     
List<Message> listItem = splitter.next();
producer.send(listItem);
} catch (Exception e) {
   
     
e.printStackTrace();
}
}
producer.shutdown();
}
}

定义批量消息消费者

public class BatchConsumer {
   
     
public static void main(String[] args) throws MQClientException {
   
     
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("cg");
consumer.setNamesrvAddr("rocketmqOS:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
consumer.subscribe("someTopicA", "*");
// 指定每次可以消费10条消息，默认为1
consumer.setConsumeMessageBatchMaxSize(10);
// 指定每次可以从Broker拉取40条消息，默认为32
consumer.setPullBatchSize(40);
consumer.registerMessageListener(new
MessageListenerConcurrently() {
   
     
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
   
     
for (MessageExt msg : msgs) {
   
     
System.out.println(msg);
}
// 消费成功的返回结果
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
// 消费异常时的返回结果
// return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
});
consumer.start();
System.out.println("Consumer Started");
}
}

消息过滤

消息者在进行消息订阅时，除了可以指定要订阅消息的Topic外，还可以对指定Topic中的消息根据指定条件进行过滤，即可以订阅比Topic更加细粒度的消息类型。
对于指定Topic消息的过滤有两种过滤方式：Tag过滤与SQL过滤。

1 Tag过滤

通过consumer的subscribe()方法指定要订阅消息的Tag。如果订阅多个Tag的消息，Tag间使用或运算符(双竖线||)连接。

1
2
3

DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");

2 SQL过滤

SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。通过SQL过滤，可以实现对消息的复杂过滤。不过，只有使用PUSH模式的消费者才能使用SQL过滤。

SQL过滤表达式中支持多种常量类型与运算符。

支持的常量类型：

数值：比如：123，3.1415
字符：必须用单引号包裹起来，比如：‘abc’
布尔：TRUE 或 FALSE
NULL：特殊的常量，表示空

支持的运算符有：

数值比较：>，>=，<，<=，BETWEEN，=
字符比较：=，<>，IN
逻辑运算：AND，OR，NOT
NULL判断：IS NULL 或者 IS NOT NULL

默认情况下Broker没有开启消息的SQL过滤功能，需要在Broker加载的配置文件中添加如下属性，以开启该功能：

enablePropertyFilter 1 = true

在启动Broker时需要指定这个修改过的配置文件。例如对于单机Broker的启动，其修改的配置文件是conf/broker.conf，启动时使用如下命令：

sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &

3 代码举例

定义Tag过滤Producer

public class FilterByTagProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
String[] tags = {
   
     "myTagA","myTagB","myTagC"};
for (int i = 0; i < 10; i++) {
   
     
byte[] body = ("Hi," + i).getBytes();
String tag = tags[i%tags.length];
Message msg = new Message("myTopic",tag,body);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
producer.shutdown();
}
}

定义Tag过滤Consumer

public class FilterByTagConsumer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("pg");
consumer.setNamesrvAddr("rocketmqOS:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
consumer.subscribe("myTopic", "myTagA || myTagB");
consumer.registerMessageListener(new
MessageListenerConcurrently() {
   
     
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
   
     
for (MessageExt me:msgs){
   
     
System.out.println(me);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started");
}
}

定义SQL过滤Producer

public class FilterBySQLProducer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
producer.start();
for (int i = 0; i < 10; i++) {
   
     
try {
   
     
byte[] body = ("Hi," + i).getBytes();
Message msg = new Message("myTopic", "myTag", body);
msg.putUserProperty("age", i + "");
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
} catch (Exception e) {
   
     
e.printStackTrace();
}
}
producer.shutdown();
}
}

定义SQL过滤Consumer

public class FilterBySQLConsumer {
   
     
public static void main(String[] args) throws Exception {
   
     
DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("pg");
consumer.setNamesrvAddr("rocketmqOS:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET
);
consumer.subscribe("myTopic", MessageSelector.bySql("age between
0 and 6"));
consumer.registerMessageListener(new
MessageListenerConcurrently() {
   
     
@Override
public ConsumeConcurrentlyStatus
consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext
context) {
   
     
for (MessageExt me:msgs){
   
     
System.out.println(me);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started");
}
}

发送端消息重试

1 说明

Producer对发送失败的消息进行重新发送的机制，称为消息发送重试机制，也称为消息重投机制。
对于消息重投，需要注意以下几点：

生产者在发送消息时，若采用同步或异步发送方式，发送失败会重试，但oneway消息发送方式发送失败是没有重试机制的
只有普通消息具有发送重试机制，顺序消息是没有的
消息重投机制可以保证消息尽可能发送成功、不丢失，但可能会造成消息重复。消息重复在RocketMQ中是无法避免的问题
消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会成为大概率事件
producer主动重发、consumer负载变化（发生Rebalance，不会导致消息重复，但可能出现重复消费）也会导致重复消息
消息重复无法避免，但要避免消息的重复消费。
避免消息重复消费的解决方案是，为消息添加唯一标识（例如消息key），使消费者对消息进行消费判断来避免重复消费
消息发送重试有三种策略可以选择：同步发送失败策略、异步发送失败策略、消息刷盘失败策略

2 同步发送失败策略

对于普通消息，消息发送默认采用round-robin策略来选择所发送到的队列。如果发送失败，默认重试2次。但在重试时是不会选择上次发送失败的Broker，而是选择其它Broker。当然，若只有一个Broker其也只能发送到该Broker，但其会尽量发送到该Broker上的其它Queue。

// 创建一个producer，参数为Producer Group名称
DefaultMQProducer producer = new DefaultMQProducer("pg");
// 指定nameServer地址
producer.setNamesrvAddr("rocketmqOS:9876");
// 设置同步发送失败时重试发送的次数，默认为2次
producer.setRetryTimesWhenSendFailed(3);
// 设置发送超时时限为5s，默认3s
producer.setSendMsgTimeout(5000);

同时，Broker还具有失败隔离功能，使Producer尽量选择未发生过发送失败的Broker作为目标Broker。其可以保证其它消息尽量不发送到问题Broker，为了提升消息发送效率，降低消息发送耗时。

思考：让我们自己实现失败隔离功能，如何来做？

1）方案一：Producer中维护某JUC的Map集合，其key是发生失败的时间戳，value为Broker实
例。Producer中还维护着一个Set集合，其中存放着所有未发生发送异常的Broker实例。选择目标Broker是从该Set集合中选择的。再定义一个定时任务，定期从Map集合中将长期未发生发送异常的Broker清理出去，并添加到Set集合。

2）方案二：为Producer中的Broker实例添加一个标识，例如是一个AtomicBoolean属性。只要该Broker上发生过发送异常，就将其置为true。选择目标Broker就是选择该属性值为false的Broker。再定义一个定时任务，定期将Broker的该属性置为false。

3）方案三：为Producer中的Broker实例添加一个标识，例如是一个AtomicLong属性。只要该Broker上发生过发送异常，就使其值增一。选择目标Broker就是选择该属性值最小的Broker。若该值相同，采用轮询方式选择。

如果超过重试次数，则抛出异常，由Producer去保证消息不丢。当然当生产者出现RemotingException、MQClientException和MQBrokerException时，Producer会自动重投消息

3 异步发送失败策略

异步发送失败重试时，异步重试不会选择其他broker，仅在同一个broker上做重试，所以该策略无法保证消息不丢。

DefaultMQProducer producer = new DefaultMQProducer("pg");
producer.setNamesrvAddr("rocketmqOS:9876");
// 指定异步发送失败后不进行重试发送
producer.setRetryTimesWhenSendAsyncFailed(0);

4 消息刷盘失败策略

消息刷盘超时（Master或Slave）或slave不可用（slave在做数据同步时向master返回状态不是SEND_OK）时，默认是不会将消息尝试发送到其他Broker的。不过，对于重要消息可以通过在Broker的配置文件设置retryAnotherBrokerWhenNotStoreOK属性为true来开启。

接收端消息重试

1 顺序消息的消费重试

对于顺序消息，当Consumer消费消息失败后，为了保证消息的顺序性，其会自动不断地进行消息重试，直到消费成功。消费重试默认间隔时间为1000毫秒。重试期间应用会出现消息消费被阻塞的情况。

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
// 顺序消息消费失败的消费重试时间间隔，单位毫秒，默认为1000，其取值范围为[10,
30000]
consumer.setSuspendCurrentQueueTimeMillis(100);

由于对顺序消息的重试是无休止的，不间断的，直至消费成功，所以，对于顺序消息的消费，务必要保证应用能够及时监控并处理消费失败的情况，避免消费被永久性阻塞。

注意，顺序消息没有发送失败重试机制，但具有消费失败重试机制

2 无序消息的消费重试

对于无序消息（普通消息、延时消息、事务消息），当Consumer消费消息失败时，可以通过设置返回状态达到消息重试的效果。不过需要注意，无序消息的重试只对集群消费方式生效，广播消费方式不提供失败重试特性。即对于广播消费，消费失败后，失败消息不再重试，继续消费后续消息。

3 消费重试次数与间隔

对于无序消息集群消费下的重试消费，每条消息默认最多重试16次，但每次重试的间隔时间是不同的，会逐渐变长。每次重试的间隔时间如下表。

若一条消息在一直消费失败的前提下，将会在正常消费后的第4小时46分后进行第16次重试。若仍然失败，则将消息投递到死信队列

修改消费重试次数

1
2
3

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
// 修改消费重试次数
consumer.setMaxReconsumeTimes(10);

对于修改过的重试次数，将按照以下策略执行：

若修改值小于16，则按照指定间隔进行重试
若修改值大于16，则超过16次的重试时间间隔均为2小时

对于Consumer Group，若仅修改了一个Consumer的消费重试次数，则会应用到该Group中所有其它Consumer实例。若出现多个Consumer均做了修改的情况，则采用覆盖方式生效。即最后被修改的值会覆盖前面设置的值。

4 重试队列

对于需要重试消费的消息，并不是Consumer在等待了指定时长后再次去拉取原来的消息进行消费，而是将这些需要重试消费的消息放入到了一个特殊Topic的队列中，而后进行再次消费的。这个特殊的队列就是重试队列。

当出现需要进行重试消费的消息时，Broker会为每个消费组都设置一个Topic名称为%RETRY%consumerGroup@consumerGroup 的重试队列。

1）这个重试队列是针对消息才组的，而不是针对每个Topic设置的（一个Topic的消息可以让多个消费者组进行消费，所以会为这些消费者组各创建一个重试队列）
2）只有当出现需要进行重试消费的消息时，才会为该消费者组创建重试队列

注意，消费重试的时间间隔与延时消费的延时等级十分相似，除了没有延时等级的前两个时间外，其它的时间都是相同的

Broker对于重试消息的处理是通过延时消息实现的。先将消息保存到SCHEDULE_TOPIC_XXXX延迟队列中，延迟时间到后，会将消息投递到%RETRY%consumerGroup@consumerGroup重试队列中。

5 消费重试配置方式

集群消费方式下，消息消费失败后若希望消费重试，则需要在消息监听器接口的实现中明确进行如下三种方式之一的配置：

方式1：返回ConsumeConcurrentlyStatus.RECONSUME_LATER（推荐）
方式2：返回Null
方式3：抛出异常

6 消费不重试配置方式

集群消费方式下，消息消费失败后若不希望消费重试，则在捕获到异常后同样也返回与消费成功后的相同的结果，即ConsumeConcurrentlyStatus.CONSUME_SUCCESS，则不进行消费重试。

消息幂等性

为了防⽌消息重复消费导致业务处理异常，消息队列 RocketMQ 版的消费者在接收到消息后，有必要根据业务上的唯⼀ Key 对消息做幂等处理。

什么是消息幂等

如果有⼀个操作，多次执⾏与⼀次执⾏所产⽣的影响是相同的，我们就称这个操作是幂等的。
当出现消费者对某条消息重复消费的情况时，重复消费的结果与消费⼀次的结果是相同的，并且多次消费并未对业务系统产⽣任何负⾯影响，那么这整个过程就可实现消息幂等。

适⽤场景

在互联⽹应⽤中，尤其在⽹络不稳定的情况下，消息队列 RocketMQ 版的消息有可能会出现重复。如果消息重复会影响您的业务处理，请对消息做幂等处理。

消息重复的场景如下：

发送时消息重复

当⼀条消息已被成功发送到服务端并完成持久化，此时出现了⽹络闪断或者客户端宕机，导致服务端对客户端应答失败。如果此时⽣产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

投递时消息重复

消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候⽹络闪断。为了保证消息⾄少被消费⼀次，消息队列 RocketMQ 版的服务端将在⽹络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且 Message ID也相同的消息。

负载均衡时消息重复（包括但不限于⽹络抖动、Broker 重启以及消费者应⽤重启）

当消息队列 RocketMQ 版的 Broker 或客户端重启、扩容或缩容时，会触发 Rebalance，此时消费者可能会收到重复消息。

实现消息幂等

定义消息幂等的两要素：

幂等令牌

幂等令牌是⽣产者和消费者两者中的既定协议，在业务中通常是具备唯⼀业务标识的字符串，如：
订单号、流⽔号等。且⼀般由⽣产者端⽣成并传递给消费者端。

处理唯⼀性的确保

缓存唯⼀索引
可以使用Redis缓存

RocketMQ如何处理消息幂等

RocketMQ能够保证消息不丢失但不保证消息不重复。

如果在RocketMQ中实现消息去重实际也是可以的，但是考虑到⾼可⽤以及⾼性能的需求，如果做了服务端的消息去重，RocketMQ就需要对消息做额外的rehash、排序等操作，这会花费较⼤的时间和空间等资源代价，收益并不明显。
RocketMQ考虑到正常情况下出现重复消息的概率其实是很⼩的，因此RocketMQ将消息幂等操作交给了业务⽅处理。

因为Message ID 有可能出现冲突（重复）的情况，因此不建议通过MessageID作为处理依据，⽽最好的⽅式是以业务唯⼀标识作为幂等处理的关键依据如：订单号、流⽔号等作为幂等处理的关键依据。⽽业务的唯⼀标识可以通过消息 Key 设置。

以⽀付场景为例，可以将消息的 Key 设置为订单号，作为幂等处理的依据。具体代码示例如下

1
2
3

Message message = new Message();
message.setKeys("ORDERID_100");
SendResult sendResult = producer.send(message);

消费者收到消息时可以根据消息的 Key，即订单号来实现消息幂等：

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext
context) {
   
     
for(MessageExt msg:msgs){
   
     
String key = msg.getKeys();
// 根据业务唯⼀标识的 Key 做幂等处理
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

消费端常⻅的幂等操作

1、业务操作之前进⾏状态查询；

消费端开始执⾏业务操作时，通过幂等id⾸先进⾏业务状态的查询，如：修改订单状态环节，当订单状态为成功/失败则不需要再进⾏处理。那么我们只需要在消费逻辑执⾏之前通过订单号进⾏订单状态查询，⼀旦获取到确定的订单状态则对消息进⾏提交，通知broker消息状态为：ConsumeConcurrentlyStatus.CONSUME_SUCCESS 。

2、唯⼀性约束保证最后⼀道防线；

上述第⼆点操作并不能保证⼀定不出现重复的数据，如：并发插⼊的场景下，如果没有乐观锁、分布式锁作为保证的前提下，很有可能出现数据的重复插⼊操作，因此我们务必要对幂等id添加唯⼀性索引，这样就能够保证在并发场景下也能保证数据的唯⼀性。

3、引⼊锁机制；

上述的第⼀点中，如果是并发更新的情况，没有使⽤悲观锁、乐观锁、分布式锁等机制的前提下，进⾏更新，很可能会出现多次更新导致状态的不准确。如：对订单状态的更新，业务要求订单只能从初始化->处理中，处理中->成功，处理中->失败，不允许跨状态更新。如果没有锁机制，很可能会将初始化的订单更新为成功，成功订单更新为失败等异常的情况。
⾼并发下，建议通过状态机的⽅式定义好业务状态的变迁，通过乐观锁、分布式锁机制保证多次更新的结果是确定的，悲观锁在并发环境不利于业务吞吐量的提⾼因此不建议使⽤。

消息堆集问题

消息堆积本质

⽣产者的⽣产速度 >> 消费者的处理速度

⽣产者的⽣产速度骤增，⽐如⽣产者的流量突然骤增
消费速度变慢，⽐如消费者实例 IO 阻塞严重或者宕机

如何处理消息堆积

如何处理消息堆积呢？可以从两个当⾯考虑：

如何通过解决系统问题、优化代码来避免消息堆积
消息已经堆积了，线上如何快速处理

发送端性能优化

从消息堆积若⼲原因来看，消息堆积的原因主要在消费端处理上，本身⽣产者端应该遵循的
原则应该是尽可能快的将消息发送到Broker中去，因此发送端除了业务处理时批量发送暂⽆好的
⼿段优化，⽽且并不是所有的业务处理都⽀持批量发送和批量接收处理。

批量发送是发送端预防消息堆积的⽅式之⼀

消费端性能优化

在设计系统的时候，⼀定要保证消费端的消费性能要⾼于⽣产端的发送性能，这样的系统才能健康的
持续运⾏。

⽅式1 增加单个消费者处理能⼒

增加单个消费者的处理能⼒这块没有绝对的办法，只能尽可能的优化消息处理业务逻辑的能
⼒，减少不必要的⾮业务相关处理时间消耗；如果消息处理业务已经优化到⽆法再优化了，那只
能通过⽅式2⽔平扩展消费者个数来优化。

⽅式2 ⽔平扩容消费者个数

消费端的性能优化除了优化消费业务逻辑以外，也可以通过⽔平扩容，增加消费端的并发数
来提升总体的消费性能。

如何快速处理

如果消息已经堆积了，线上如何快速处理。对于系统发⽣消息积压的情况，需要先解决积
压，再分析原因，毕竟保证系统的可⽤性是⾸先要解决的问题。

1、消费端扩容；–通⽤⽅式；
2、服务降级；–快速失败，不⼀定适⽤所有业务场景；
3、跳过⾮重要消息：发⽣消息堆积时，如果消费速度⼀直追不上发送速度，可以选择丢弃不重要的消息；
4、异常监控–属于运维层⾯措施；

消息查询

消息查询介绍

3种消息查询⽅式：

Message Key 查询：消息的key是业务开发在发送消息之前⾃⾏指定的，通常会把具有业务含义，区分度⾼的字段作为消息的key，如⽤户id，订单id等。
Unique Key查询：除了业务开发明确的指定消息中的key，RocketMQ⽣产者客户端在发送发送消息之前，会⾃动⽣成⼀个UNIQ_KEY，设置到消息的属性中，从逻辑上唯⼀代表⼀条消息。
Message Id 查询：Message Id 是消息发送后，在Broker端⽣成的，其包含了Broker的地址，和在CommitLog中的偏移信息，并会将Message Id作为发送结果的⼀部分进⾏返回。Message Id中属于精确匹配，可以唯⼀定位⼀条消息，不需要使⽤哈希索引机制，查询效率更⾼。

RocketMQ有意弱化Unique Key与Message Id的区别，对外都称之为Message Id。在通过RocketMQ的命令⾏⼯具或管理平台进⾏查询时，⼆者可以通⽤。在根据Unique Key进⾏查询时，本身是有可能查询到多条消息的，但是查询⼯具会进⾏过滤，只会返回⼀条消息。
业务开发同学在使⽤RocketMQ时，应该养成良好的习惯，在发送/消费消息时，将这些信息记录下来，通常是记录到⽇志⽂件中，以便在出现问题时进⾏排查。

//1 构建消息对象Message
Message msg = new Message();
msg.setTopic("TopicA");
msg.setKeys("Key1");
msg.setBody("message body".getBytes());
try{
   
     
//2 发送消息
SendResult result = producer.send(msg);
//3 打印发送结果
System.out.println(result);
}catch (Exception e){
   
     
e.printStackTrace();
}

事实上，⽤户主动设置的Key以及客户端⾃动⽣成的Unique Key，最终都会设置到Message对象的
properties属性中，如下图所示

其中： KEYS：表示⽤户通过setKeys⽅法设置的消息key，
UNIQ_KEY：表示客户端⾃动⽣成的Unique Key。

结果中包含Unique Key和Message Id，如下所示：

SendResult [sendStatus=SEND_OK, msgId=0A1427544F4818B4AAC27DD168880000,
offsetMsgId=0A14275400002A9F00000000001F268E, messageQueue=MessageQueue
[topic=TopicTest, brokerName=broker-a, queueId=2], queueOffset=1173]

其中：

sendStatus：表示消息发送结果的状态
msgId：注意这⾥的命名虽然是msgId，但实际上其是Unique Key
offsetMsgId：Broker返回的Message ID 。在后⽂中，未进⾏特殊说明的情况下，Message ID总
是表示offsetMsgId。
messageQueue：消息发送到了哪个的队列。
queueOffset：消息在队列中的偏移量，每次发送到⼀个队列时，offset+1

消息查询⼯具

命令⾏⼯具
管理平台
客户端API

命令⾏⼯具

$ sh bin/mqadmin
The most commonly used mqadmin commands are:
...
queryMsgById 按照Message Id查询消息
queryMsgByKey 按照Key查询消息
queryMsgByUniqueKey 按照UNIQ_KEY查询消息
...

例如，要查询在TopicA中，key为Key-0的消息

这⾥，我们看到输出结果中包含了2条记录。其中：

Message ID列：这⾥这⼀列的名字显示有问题，实际上其代表的是Unique Key
QID列：表示队列的ID，注意在RocketMQ中唯⼀地位⼀个队列需要topic+brokerName+queueId。这⾥只显示了queueId，其实并不能知道在哪个Broker上。
Offset：消息在在队列中的偏移量

在查询到Unique Key之后，我们就可以使⽤另外⼀个命令：queryMsgByUniqueKey，来查询消息的具
体内容。

对于消息体的内容，会存储到Message Body Path字段指定到的路径中。可通过cat命令查看(仅适⽤于
消息体是字符串)：

指定消费者重新消费：
queryMsgByUniqueKey⼦命令还接收另外两个参数：-g参数⽤于指定消费者组名称，-d参数指定消费
者client id。指定了这两个参数之后，消息将由消费者直接消费，⽽不是打印在控制台上。
⾸先，通过consumerStatus命令，查询出消费者组下的client id信息，如：
这⾥显示了消费者组please_rename_unique_group_name下⾯只有⼀个消费者，client id为
10、 20.39.84@20820；
接着我们可以在queryMsgByUniqueKey⼦命令中，添加-g和-d参数，如下所示：

可以看到，这⾥并没有打印出消息内容，取⽽代之的是消息消费的结果。
在内部，主要是分为3个步骤来完成让指定消费者来消费这条消息，如下图所示：

第1步：
命令⾏⼯具给所有Broker发起QUERY_MESSAGE请求查询消息，因为并不知道UNIQ_KEY这条消息在哪个Broker上，且最多只会返回⼀条消息，如果超过1条其他会过滤掉；如果查询不到就直接报错。
第2步：
根据消息中包含了Store Host信息，也就是消息存储在哪个Broker上，接来下命令⾏⼯具会直接给这
个Broker发起CONSUME_MESSAGE_DIRECTLY请求，这个请求会携带msgId，group和client id的信息
第3步：
Broker接收到这个请求，查询出消息内容后，主动给消费者发送CONSUME_MESSAGE_DIRECTLY通知请求，注意虽然与第2步使⽤了同⼀个请求码，但不同的是这个请求中包含了消息体的内容，消费者可直接处理。注意：这⾥并不是将消息重新发送到Topic中，否则订阅这个Topic的所有消费者组，都会重新消费这条消息。

管理平台

根据Topic时间范围查询：

按Topic 查询属于范围查询，不推荐使⽤，因为时间范围内消息很多，不具备区分度。查询时，尽可能设置最为精确的时间区间，以便缩⼩查询范围，提⾼速度。最多返回2000条数据。

根据Message Key查询：

按Message Key 查询属于模糊查询，仅适⽤于没有记录 Message ID 但是设置了具有区分度的Message Key的情况。⽬前，根据Message Key查询，有⼀个很⼤局限性：不能指定时间范围，且最多返回64条数据。如果⽤户指定的key重复率⽐较⾼的话，就有可能搜不到。

根据Message Id查询：

按Message ID 查询属于精确查询，速度快，精确匹配，只会返回⼀条结果，推荐使⽤。在这⾥，传⼊Unique Key，offsetMsgId都可以。

客户端API

除了通过命令⾏⼯具和管理平台，还可以通过客户端API的⽅式来进⾏查询，这其实是最本质的⽅式，命令⾏⼯具和管理平台的查询功能都是基于此实现。

在org.apache.rocketmq.client.MQAdmin接⼝中，定义了以下⼏个⽅法⽤于消息查询：

常⽤的DefaultMQProducer、DefaultMQPushConsumer等，都实现了此接⼝，因此都具备消息查询的能⼒.

public interface MQAdmin {
   
     
...
//msgId参数：仅接收SendResult中的offsetMsgId，返回单条消息
MessageExt viewMessage(final String msgId)
//msgId参数：传⼊SendResult中的offsetMsgId、msgId都可以，返回单条消息
MessageExt viewMessage(String topic,String msgId)
//在指定topic下，根据key进⾏查询，并指定最⼤返回条数，以及开始和结束时间
QueryResult queryMessage(final String topic, final String key,
final int maxNum, final long begin,final long end)
...
}

在内部，实际上都是基于MQAdminImpl这个类来完成的。

viewMessage⽅法：
两种viewMessage⽅法重载形式，都只会返回单条消息。

//初始化Producer
DefaultMQProducer producer = new
DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
//根据UniqueKey查询
String uniqueKey = "0A1427544F4818B4AAC27DD168880000";
MessageExt msg = producer.viewMessage("TopicTest", uniqueKey);
//打印结果：这⾥仅输出Unique Key与offsetMsgId
MessageClientExt msgExt= (MessageClientExt) msg;
System.out.println("Unique Key:"+msgExt.getMsgId()//即UNIQUE_KEY
+"\noffsetMsgId:"+msgExt.getOffsetMsgId());
//Shut down once the producer instance is not longer in use.

输出结果如下：

Unique Key:0A1427544F4818B4AAC27DD168880000
offsetMsgId:0A14275400002A9F00000000001F268E

如果我们把offsetMsgId当做⽅法参数传⼊，也可以查询到相同的结果。这是因为，在⽅法内部实际
上是分两步进⾏查询的：
1、先把参数当做offsetMsgId，即MessageId进⾏查询；
2、如果失败，再尝试当做UniqueKey进⾏查询；
源码如下所示：
DefaultMQProducer#viewMessage(String,String)

public MessageExt viewMessage(String topic,
String msgId) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
   
     
try {
   
     
//1 尝试当做offsetMsgId进⾏查询
MessageId oldMsgId = MessageDecoder.decodeMessageId(msgId);
return this.viewMessage(msgId);
} catch (Exception e) {
   
     
}
//2 尝试当做UNIQ_KEY进⾏查询
return
this.defaultMQProducerImpl.queryMessageByUniqKey(withNamespace(topic), msgId);
}

前⾯提到，Unique Key只是从逻辑上代表⼀条消息，实际上在Broker端可能存储了多条，因此在当做Unique Key进⾏查询时，会进⾏过滤，只取其中⼀条。源码如下所示：
MQAdminImpl#queryMessageByUniqKey

public MessageExt queryMessageByUniqKey(String topic,
String uniqKey) throws InterruptedException, MQClientException {
   
     
//根据uniqKey进⾏查询
QueryResult qr = this.queryMessage(topic, uniqKey, 32,
MessageClientIDSetter.getNearlyTimeFromID(uniqKey).getTime() -
1000, Long.MAX_VALUE, true);
//对查询结果进⾏过滤，最多只取⼀条
if (qr != null && qr.getMessageList() != null &&
qr.getMessageList().size() > 0) {
   
     
return qr.getMessageList().get(0);
} else {
   
     
return null;
}
}

实现原理

Unqiue Key & Message Key都需要利⽤RocketMQ的哈希索引机制来完成消息查询，Message Id是在Broker端⽣成的，其包含了Broker地址和commit Log offset信息，可以精确匹配⼀条消息，查询消息更好。

Unique Key 是⽣产者发送消息之前，由RocketMQ 客户端⾃动⽣成的。
DefaultMQProducerImpl#sendKernelImpl

private SendResult sendKernelImpl(final Message msg,
final MessageQueue mq,
final CommunicationMode communicationMode,
final SendCallback sendCallback,
final TopicPublishInfo topicPublishInfo,
final long timeout) {
   
     //省略异常声明
//...略
try {
   
     
//如果不是批量消息，则⽣成Unique Key
if (!(msg instanceof MessageBatch)) {
   
     
// 设置唯⼀编号
MessageClientIDSetter.setUniqID(msg);
}
//...略

MessageClientIDSetter#setUniqID

public static void setUniqID(final Message msg) {
   
     
// Unique Key为空的情况下，才进⾏设置
if
(msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null)
{
   
     
msg.putProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX,
createUniqID());
}
}

Unique Key作⽤

了解Unique Key的作⽤对于我们理解消息重复的原因有很⼤的帮助。RocketMQ并不保证消息投递过程中的Exactly Once语义，即消息只会被精确消费⼀次，需要消费者⾃⼰做幂等。⽽通常导致消息重复消费的原因，主要包括：

⽣产者发送时消息重复：
消费者Rebalance时消息重复：

导致⽣产者发送重复消息的原因可能是：⼀条消息已被成功发送到服务端并完成持久化，由于⽹络超时此时出现了⽹络闪断或者客户端宕机，导致服务端对客户端应答失败，此时⽣产者将再次尝试发送消息。
在重试发送时，sendKernelImpl会被重复调⽤，意味着setUniqID⽅法会被重复调⽤，不过由于setUniqID⽅法实现中进⾏判空处理，因此重复设置Unique Key。在这种情况下，消费者后续会收到两条内容相同并且 Unique Key 也相同的消息(offsetMsgId不同，因为对Broker来说存储了多次)。

那么消费者如何判断，消费重复是因为重复发送还是Rebalance导致的重复消费呢？

消费者实现MessageListener接⼝监听到的消息类型是MessageExt，可以将其强制转换为
MessageClientExt，之后调⽤getMsgId⽅法获取Unique Key，调⽤getOffsetMsgId获得Message Id。

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
msgs,
ConsumeConcurrentlyContext context) {
   
     
for (MessageExt msg:msgs){
   
     
MessageClientExt mct = (MessageClientExt)msg;
String uniqueKey = mct.getMsgId();
String messageId = mct.getOffsetMsgId();
}
System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

批量发送模式下的Unique Key

DefaultMQProducer提供了批量发送消息的接⼝：

1	public SendResult send(Collection<Message> msgs)

在内部，这批消息⾸先会被构建成⼀个MessageBatch对象。在前⾯sendKernelImpl⽅法中我们也看到了，对于MessageBatch对象，并不会设置Unique Key。

这是因为在将批量消息转换成MessageBatch时，已经设置过了。

⼀个批量消息中每条消息Unique Key是相同的？？？？？？

//Instantiate with a producer group name.
DefaultMQProducer producer = new
DefaultMQProducer("please_rename_unique_group_name");
// Specify name server addresses.
producer.setNamesrvAddr("localhost:9876");
//Launch the instance.
producer.start();
//构建批量消息
ArrayList<Message> msgs = new ArrayList<>();
Message msg1 = new Message("TopicTest",("message3").getBytes());
Message msg2 = new Message("TopicTest",("message4").getBytes());
msgs.add(msg1);
msgs.add(msg2);
//发送
SendResult result = producer.send(msgs);
//打印
System.out.println(result);

输出如下所示：

可以看到，此时输出的msgId(即Unique Key)和offsetMsgId都会包含多个值。客户端给批量消息中每条消息设置不同的Unqiue Key，参考DefaultMQProducer#batch()：

private MessageBatch batch(Collection<Message> msgs) throws MQClientException
{
   
     
MessageBatch msgBatch;
try {
   
     
//1 将消息集合转换为MessageBatch
msgBatch = MessageBatch.generateFromList(msgs);
//2 迭代每个消息，逐⼀设置Unique Key
for (Message message : msgBatch) {
   
     
Validators.checkMessage(message, this);
MessageClientIDSetter.setUniqID(message);
}
//3 设置批量消息的消息体
msgBatch.setBody(msgBatch.encode());
} catch (Exception e) {
   
     
throw new MQClientException("Failed to initiate the MessageBatch", e);
}
return msgBatch;
}

Message Id

Message Id是在Broker端⽣成的，⽤于唯⼀标识⼀条消息，在根据Message Id查询的情况下，最多只能查询到⼀条消息。

package org.apache.rocketmq.common.message;
import java.net.SocketAddress;
public class MessageId {
   
     
private SocketAddress address;
private long offset;

并提供了⼀个MessageDecoder对象来创建或者解码MessageId。

1
2
3

public static String createMessageId(final ByteBuffer input,
final ByteBuffer addr, final long offset)
public static MessageId decodeMessageId(final String msgId)

Broker端在顺序存储消息时，⾸先会通过createMessageId⽅法创建msgId
CommitLog.DefaultAppendMessageCallback#doAppend

..........................
String msgId;
//3 创建msgId
if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
   
     
msgId = MessageDecoder.createMessageId(this.msgIdMemory,
msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
} else {
   
     
msgId = MessageDecoder.createMessageId(this.msgIdV6Memory,
msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
}

⽽客户端在根据msgId向Broker查询消息时，⾸先会将通过MessageDecoder的decodeMessageId⽅法，之后直接向这个broker进⾏查询指定位置的消息。
参⻅：MQAdminImpl#viewMessage

public MessageExt viewMessage(
String msgId) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException {
   
     
//1 根据msgId解码成MessageId对象
MessageId messageId = null;
try {
   
     
messageId = MessageDecoder.decodeMessageId(msgId);
} catch (Exception e) {
   
     
throw new MQClientException(ResponseCode.NO_MESSAGE, "query
message by id finished, but no message.");
}
//2 根据MessageId中的Broker地址和commit log offset信息进⾏查询
return
this.mQClientFactory.getMQClientAPIImpl().viewMessage(RemotingUtil.socketAddre
ss2String(messageId.getAddress()),
messageId.getOffset(), timeoutMillis);
}

由于根据Message Id进⾏查询，实际上是直接从特定Broker的CommitLog中的指定位置进⾏查询的，属于精确匹配，并不像⽤户设置的key，或者Unique Key那么样，需要使⽤到哈希索引机制，因此效率很⾼。

总结

3种消息查询⽅式：Message Key & Unique Key & Message Id
3种消息查询⼯具：命令⾏、管理平台、客户端API，且⽀持将查询到让特定/所有消费者组重新消费
屏蔽Unique Key & Message Id区别，很多地⽅⼆者可以通⽤
Message Key & Unique Key 需要使⽤到哈希索引机制，有额外的索引维护成本
Message Id由Broker和commit log offset组成，属于精确匹配，查询效率更好

高可用性

集群部署

1 数据复制与刷盘策略

复制策略

复制策略是Broker的Master与Slave间的数据同步方式。分为同步复制与异步复制：

同步复制：消息写入master后，master会等待slave同步数据成功后才向producer返回成功ACK
异步复制：消息写入master后，master立即向producer返回成功ACK，无需等待slave同步数据成
功

异步复制策略会降低系统的写入延迟，RT变小，提高了系统的吞吐量

刷盘策略

刷盘策略指的是broker中消息的落盘方式，即消息发送到broker内存后消息持久化到磁盘的方式。分为
同步刷盘与异步刷盘：

同步刷盘：当消息持久化到broker的磁盘后才算是消息写入成功。
异步刷盘：当消息写入到broker的内存后即表示消息写入成功，无需等待消息持久化到磁盘。

1）异步刷盘策略会降低系统的写入延迟，RT变小，提高了系统的吞吐量
2）消息写入到Broker的内存，一般是写入到了PageCache
3）对于异步刷盘策略，消息会写入到PageCache后立即返回成功ACK。但并不会立即做落盘操
作，而是当PageCache到达一定量时会自动进行落盘

2 Broker集群模式

根据Broker集群中各个节点间关系的不同，Broker集群可以分为以下几类：

单Master

只有一个broker（其本质上就不能称为集群）。这种方式也只能是在测试时使用，生产环境下不能使
用，因为存在单点问题。

多Master

broker集群仅由多个master构成，不存在Slave。同一Topic的各个Queue会平均分布在各个master节点
上。

优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器
宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步
刷盘一条不丢），性能最高；
缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅（不可消费），
消息实时性会受到影响。

以上优点的前提是，这些Master都配置了RAID磁盘阵列。如果没有配置，一旦出现某Master宕
机，则会发生大量消息丢失的情况。

多Master多Slave模式-异步复制

broker集群由多个master构成，每个master又配置了多个slave（在配置了RAID磁盘阵列的情况下，一
个master一般配置一个slave即可）。master与slave的关系是主备关系，即master负责处理消息的读写
请求，而slave仅负责消息的备份与master宕机后的角色切换。

异步复制即前面所讲的复制策略中的异步复制策略，即消息写入master成功后，master立即向
producer返回成功ACK，无需等待slave同步数据成功。

该模式的最大特点之一是，当master宕机后slave能够自动切换为master。不过由于slave从master的同
步具有短暂的延迟（毫秒级），所以当master宕机后，这种异步复制方式可能会存在少量消息的丢失问
题。

Slave从Master同步的延迟越短，其可能丢失的消息就越少
对于Master的RAID磁盘阵列，若使用的也是异步复制策略，同样也存在延迟问题，同样也可能
会丢失消息。但RAID阵列的秘诀是微秒级的（因为是由硬盘支持的），所以其丢失的数据量会
更少。

多Master多Slave模式-同步双写

该模式是多Master多Slave模式的同步复制实现。所谓同步双写，指的是消息写入master成功后，
master会等待slave同步数据成功后才向producer返回成功ACK，即master与slave都要写入成功后才会
返回成功ACK，也即双写。

该模式与异步复制模式相比，优点是消息的安全性更高，不存在消息丢失的情况。但单个消息的RT略
高，从而导致性能要略低（大约低10%）。

该模式存在一个大的问题：对于目前的版本，Master宕机后，Slave 不会自动切换到Master。

最佳实践

一般会为Master配置RAID10磁盘阵列，然后再为其配置一个Slave。即利用了RAID10磁盘阵列的高
效、安全性，又解决了可能会影响订阅的问题。

1）RAID磁盘阵列的效率要高于Master-Slave集群。因为RAID是硬件支持的。也正因为如此，
所以RAID阵列的搭建成本较高。

2）多Master+RAID阵列，与多Master多Slave集群的区别是什么？

多Master+RAID阵列，其仅仅可以保证数据不丢失，即不影响消息写入，但其可能会影响到
消息的订阅。但其执行效率要远高于多Master多Slave集群
多Master多Slave集群，其不仅可以保证数据不丢失，也不会影响消息写入。其运行效率要低
于多Master+RAID阵列

集群搭建实例

1 集群架构

这里要搭建一个双主双从异步复制的Broker集群。为了方便，这里使用了两台主机来完成集群的搭建。
这两台主机的功能与broker角色分配如下表。

2 克隆生成rocketmqOS1

克隆rocketmqOS主机，并修改配置。指定主机名为rocketmqOS1。

3 修改rocketmqOS1配置文件

要修改的配置文件在rocketMQ解压目录的conf/2m-2s-async目录中。

修改broker-a.properties

将该配置文件内容修改为如下：

# 指定整个broker集群的名称，或者说是RocketMQ集群的名称
brokerClusterName=DefaultCluster
# 指定master-slave集群的名称。一个RocketMQ集群可以包含多个master-slave集群
brokerName=broker-a
# master的brokerId为0
brokerId=0
# 指定删除消息存储过期文件的时间为凌晨4点
deleteWhen=04
# 指定未发生更新的消息存储文件的保留时长为48小时，48小时后过期，将会被删除
fileReservedTime=48
# 指定当前broker为异步复制master
brokerRole=ASYNC_MASTER
# 指定刷盘策略为异步刷盘
flushDiskType=ASYNC_FLUSH
# 指定Name Server的地址
namesrvAddr=192.168.59.164:9876;192.168.59.165:9876

修改broker-b-s.properties

将该配置文件内容修改为如下：

brokerClusterName=DefaultCluster
# 指定这是另外一个master-slave集群
brokerName=broker-b
# slave的brokerId为非0
brokerId=1
deleteWhen=04
fileReservedTime=48
# 指定当前broker为slave
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
namesrvAddr=192.168.59.164:9876;192.168.59.165:9876
# 指定Broker对外提供服务的端口，即Broker与producer与consumer通信的端口。默认
10911。由于当前主机同时充当着master1与slave2，而前面的master1使用的是默认端口。这
里需要将这两个端口加以区分，以区分出master1与slave2
listenPort=11911
# 指定消息存储相关的路径。默认路径为~/store目录。由于当前主机同时充当着master1与
slave2，master1使用的是默认路径，这里就需要再指定一个不同路径
storePathRootDir=~/store-s
storePathCommitLog=~/store-s/commitlog
storePathConsumeQueue=~/store-s/consumequeue
storePathIndex=~/store-s/index
storeCheckpoint=~/store-s/checkpoint
abortFile=~/store-s/abort

其它配置

除了以上配置外，这些配置文件中还可以设置其它属性。

#指定整个broker集群的名称，或者说是RocketMQ集群的名称
brokerClusterName=rocket-MS
#指定master-slave集群的名称。一个RocketMQ集群可以包含多个master-slave集群
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=nameserver1:9876;nameserver2:9876
#默认为新建Topic所创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议生产环境中关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议生产环境中关闭
autoCreateSubscriptionGroup=true
#Broker对外提供服务的端口，即Broker与producer与consumer通信的端口
listenPort=10911
#HA高可用监听端口，即Master与Slave间通信的端口，默认值为listenPort+1
haListenPort=10912
#指定删除消息存储过期文件的时间为凌晨4点
deleteWhen=04
#指定未发生更新的消息存储文件的保留时长为48小时，48小时后过期，将会被删除
fileReservedTime=48
#指定commitLog目录中每个文件的大小，默认1G
mapedFileSizeCommitLog=1073741824
#指定ConsumeQueue的每个Topic的每个Queue文件中可以存放的消息数量，默认30w条
mapedFileSizeConsumeQueue=300000
#在清除过期文件时，如果该文件被其他线程所占用（引用数大于0，比如读取消息），此时会阻止
此次删除任务，同时在第一次试图删除该文件时记录当前时间戳。该属性则表示从第一次拒绝删除
后开始计时，该文件最多可以保留的时长。在此时间内若引用数仍不为0，则删除仍会被拒绝。不过
时间到后，文件将被强制删除
destroyMapedFileIntervalForcibly=120000
#指定commitlog、consumequeue所在磁盘分区的最大使用率，超过该值，则需立即清除过期文
件
diskMaxUsedSpaceRatio=88
#指定store目录的路径，默认在当前用户主目录中
storePathRootDir=/usr/local/rocketmq-all-4.5.0/store
#commitLog目录路径
storePathCommitLog=/usr/local/rocketmq-all-4.5.0/store/commitlog
#consumeueue目录路径
storePathConsumeQueue=/usr/local/rocketmq-all-4.5.0/store/consumequeue
#index目录路径
storePathIndex=/usr/local/rocketmq-all-4.5.0/store/index
#checkpoint文件路径
storeCheckpoint=/usr/local/rocketmq-all-4.5.0/store/checkpoint
#abort文件路径
abortFile=/usr/local/rocketmq-all-4.5.0/store/abort
#指定消息的最大大小
maxMessageSize=65536
#Broker的角色
# - ASYNC_MASTER 异步复制Master
# - SYNC_MASTER 同步双写Master
# - SLAVE
brokerRole=SYNC_MASTER
#刷盘策略
# - ASYNC_FLUSH 异步刷盘
# - SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH
#发消息线程池数量
sendMessageThreadPoolNums=128
#拉消息线程池数量
pullMessageThreadPoolNums=128
#强制指定本机IP，需要根据每台机器进行修改。官方介绍可为空，系统默认自动识别，但多网卡
时IP地址可能读取错误
brokerIP1=192.168.3.105

4 克隆生成rocketmqOS2

克隆rocketmqOS1主机，并修改配置。指定主机名为rocketmqOS2。

5 修改rocketmqOS2配置文件

对于rocketmqOS2主机，同样需要修改rocketMQ解压目录的conf目录的子目录2m-2s-async中的两个配
置文件。

修改broker-b.properties

将该配置文件内容修改为如下：

brokerClusterName=DefaultCluster
brokerName=broker-b
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
namesrvAddr=192.168.59.164:9876;192.168.59.165:9876

修改broker-a-s.properties

将该配置文件内容修改为如下：

brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
namesrvAddr=192.168.59.164:9876;192.168.59.165:9876
listenPort=11911
storePathRootDir=~/store-s
storePathCommitLog=~/store-s/commitlog
storePathConsumeQueue=~/store-s/consumequeue
storePathIndex=~/store-s/index
storeCheckpoint=~/store-s/checkpoint
abortFile=~/store-s/abort

6 启动服务器

启动NameServer集群

分别启动rocketmqOS1与rocketmqOS2两个主机中的NameServer。启动命令完全相同。

1 2	nohup sh bin/mqnamesrv & tail -f ~/logs/rocketmqlogs/namesrv.log

启动两个Master

分别启动rocketmqOS1与rocketmqOS2两个主机中的broker master。注意，它们指定所要加载的配置
文件是不同的。

nohup sh bin/mqbroker -c conf/2m-2s-async/broker-a.properties &
tail -f ~/logs/rocketmqlogs/broker.log
nohup sh bin/mqbroker -c conf/2m-2s-async/broker-b.properties &
tail -f ~/logs/rocketmqlogs/broker.log

启动两个Slave

分别启动rocketmqOS1与rocketmqOS2两个主机中的broker slave。注意，它们指定所要加载的配置文
件是不同的。

nohup sh bin/mqbroker -c conf/2m-2s-async/broker-b-s.properties &
tail -f ~/logs/rocketmqlogs/broker.log
nohup sh bin/mqbroker -c conf/2m-2s-async/broker-a-s.properties &
tail -f ~/logs/rocketmqlogs/broker.log