消息中间件RocketMQ

RocketMQ 是阿里巴巴开源的分布式消息中间件。支持事务消息、顺序消息、批量消息、延时消息、消息回溯等。它里面有几个区别于标准消息中件间的概念，如Group、Topic、Queue等。系统组成则由Producer、Consumer、Broker、NameServer等。

功能优势

削峰填谷：主要解决瞬时写压力大于应用服务能力导致消息丢失、系统奔溃等问题
应用解耦：解决不同重要程度、不同能力级别系统之间依赖导致一死全死
提升性能：当存在一对多调用时，可以发一条消息给消息系统，让消息系统通知相关系统
蓄流压测：线上有些链路不好压测，可以通过堆积一定量消息再放开来压测
异步处理：不需要同步执行的远程调用可以有效提高响应时间

架构设计

部署模型

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HLdplfIJ-1638173732530)(http://wb.paic.com.cn/file/uid/3892c9f51d984504b8fb579f6cded00a)]

rocketmq各种组件

NameServer

一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现
底层由Netty实现，提供了路由管理、服务注册、服务发现的功能，是一个无状态节点
NameServer是服务发现者，集群中各个角色（Producer、Broker、Consumer等）都需要定时向NameServer上报自己的状态，以便互相发现彼此，超时不上报的话，NameServer会把它从列表中剔除
NameServer可以部署多个，当多个NameServer存在的时候，其他角色同时向他们上报信息，以保证高可用，
NameServer集群间互不通信，没有主备的概念
NameServer内存式存储，NameServer中的Broker、Topic等信息默认不会持久化，所以他是无状态节点

NameServer主要包括两个功能：

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

Broker

Broker充当着消息中转角色，负责存储消息、转发消息
Broker在RocketMQ系统中负责接收并存储从生产者发送来的消息，同时为消费者的拉取请求作准备。
Broker会定时向NameSrver提交自己的信息
每个Broker节点在启动时都会遍历NameServer列表，与每个NameServer建立长连接，注册自己的信息，之后定时上报

下图为Broker Server的功能模块示意图。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-klea5IaL-1638173732533)(http://wb.paic.com.cn/file/uid/59c60878446444ddb5930dacdf6b607f)]

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对消息进行快速查询的功能

Producer

消息的生产者
随机选择其中一个NameServer节点建立长连接，获得Topic路由信息（包括Topic下的Queue，这些Queue分布在哪些Broker上等等）
接下来向提供Topic服务的Master建立长连接（因为RocketMQ只有Master才能写消息），且定时向Master发送心跳

Consumer

消息的消费者
通过NameServer集群获得Topic的路由信息，连接到对应的Broker上消费消息
由于Master和Slave都可以读取消息，因此Consumer会与Master和Slave都建立连接进行消费消息

rocketmq工作流程

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

实现原理

RocketMQ由NameServer注册中心集群、Producer生产者集群、Consumer消费者集群和若干Broker（RocketMQ进程）组成，它的架构原理是这样的：

Broker在启动的时候去向所有的NameServer注册，并保持长连接，每30s发送一次心跳
Producer在发送消息的时候从NameServer获取Broker服务器地址，根据负载均衡算法选择一台服务器来发送消息
Conusmer消费消息的时候同样从NameServer获取Broker地址，然后主动拉取消息来消费

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4nUxzTB8-1638173732536)(http://wb.paic.com.cn/file/uid/c88b814f7c0041a9a1f7b6796657321e)]

核心概念

Message(消息)

消息载体。Message发送或者消费的时候必须指定Topic。Message有一个可选的Tag项用于过滤消息，还可以添加额外的键值对。

Topic(主题)

消息的逻辑分类，发消息之前必须要指定一个topic才能发，就是将这条消息发送到这个topic上。消费消息的时候指定这个topic进行消费。就是逻辑分类。

Queue(队列)

1个Topic会被分为N个Queue，数量是可配置的。message本身其实是存储到queue上的，消费者消费的也是queue上的消息。多说一嘴，比如1个topic4个queue，有5个Consumer都在消费这个topic，那么会有一个consumer浪费掉了，因为负载均衡策略，每个consumer消费1个queue，5>4，溢出1个，这个会不工作。

Tag(标签)

Tag 是 Topic 的进一步细分，顾名思义，标签。每个发送的时候消息都能打tag，消费的时候可以根据tag进行过滤，选择性消费。

Producer Group(生产者组)

消息生产者组。标识发送同一类消息的Producer，通常发送逻辑一致。发送普通消息的时候，仅标识使用，并无特别用处。若事务消息，如果某条发送某条消息的producer-A宕机，使得事务消息一直处于PREPARED状态并超时，则broker会回查同一个group的其他producer，确认这条消息应该commit还是rollback。但开源版本并不完全支持事务消息（阉割了事务回查的代码）。

Consumer Group(消费者组)

消息消费者组。标识一类Consumer的集合名称，这类Consumer通常消费一类消息，且消费逻辑一致。同一个Consumer Group下的各个实例将共同消费topic的消息，起到负载均衡的作用。消费进度以Consumer Group为粒度管理，不同Consumer Group之间消费进度彼此不受影响，即消息A被Consumer Group1消费过，也会再给Consumer Group2消费。

注： RocketMQ要求同一个Consumer Group的消费者必须要拥有相同的注册信息，即必须要听一样的topic(并且tag也一样)。

消息标识（MessageId/Key）

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

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

消费进度Offset

消费进度offset是用来记录每个Queue的不同消费组的消费进度的。根据消费进度记录器的不同，可以
分为两种模式：本地模式和远程模式。

1、offset本地管理模式

当消费模式为广播消费时，offset使用本地模式存储。因为每条消息会被所有的消费者消费，每个消费
者管理自己的消费进度，各个消费者之间不存在消费进度的交集。

Consumer在广播消费模式下offset相关数据以json的形式持久化到Consumer本地磁盘文件中，默认文
件路径为当前用户主目录下的.rocketmq_offsets/consumerClientId/groupName/Offsets.json 。
其中consumerClientId为当前消费者实例id，默认为ip@DEFAULT；groupName为消费者组名称。

2、offset远程管理模式

当消费模式为集群消费时，offset使用远程模式管理。因为所有Cosnumer实例对消息采用的是均衡消
费，所有Consumer共享Queue的消费进度。

Consumer在集群消费模式下offset相关数据以json的形式持久化到Broker磁盘文件中，文件路径为当前
用户主目录下的store/config/consumerOffset.json 。

Broker启动时会加载这个文件，并写入到一个双层Map（ConsumerOffsetManager）。外层map的key
为topic@group，value为内层map。内层map的key为queueId，value为offset。当发生Rebalance时，
新的Consumer会从该Map中获取到相应的数据来继续消费。
集群模式下offset采用远程管理模式，主要是为了保证Rebalance机制。

3、offset作用

消费者是如何从最开始持续消费消息的？消费者要消费的第一条消息的起始位置是用户自己通过consumer.setConsumeFromWhere()方法指定的，在Consumer启动后，其要消费的第一条消息的起始位置常用的有三种，这三种位置可以通过枚举类型常量设置。这个枚举类型为ConsumeFromWhere。

当消费完一批消息后，Consumer会提交其消费进度offset给Broker，Broker在收到消费进度后会将其更
新到那个双层Map（ConsumerOffsetManager）及consumerOffset.json文件中，然后向该Consumer进
行ACK，而ACK内容中包含三项数据：当前消费队列的最小offset（minOffset）、最大
offset（maxOffset）、及下次消费的起始offset（nextBeginOffset）

public enum ConsumeFromWhere {/***  从queue的当前最后一条消息开始消费*/CONSUME_FROM_LAST_OFFSET,@DeprecatedCONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST,@DeprecatedCONSUME_FROM_MIN_OFFSET,@DeprecatedCONSUME_FROM_MAX_OFFSET,/***  从queue的第一条消息开始消费*/CONSUME_FROM_FIRST_OFFSET,/***  从指定的具体时间戳位置的消息开始消费。**/CONSUME_FROM_TIMESTAMP,
}

rocketmq核心组件启动流程

nameserv启动流程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DlnF7F5g-1638173732540)(http://wb.paic.com.cn/file/uid/b69d2887e1504216b508186cb2ea5049)]

broker启动流程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3I3NguL4-1638173732543)(http://wb.paic.com.cn/file/uid/2ec4597270d145eda3c32d421eb69458)]

producer启动流程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-hx5XmbGz-1638173732545)(http://wb.paic.com.cn/file/uid/41faefb45cc448a2bc676c4f81d27be1)]

consumer启动流程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-m1I4B7j9-1638173732547)(http://wb.paic.com.cn/file/uid/be7d7b47bea14cd98c17ae478d7da72e)]

消息发送

1、消息类型

rocketmq中有如下几种消息类型

普通消息
顺序消息
延时消息
事务消息

普通消息

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

同步发送消息

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZvnFbB4V-1638173732549)(http://wb.paic.com.cn/file/uid/923bc49337d5440f82b86acaf73498ed)]

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

异步发送消息
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-OHcGeKB4-1638173732551)(http://wb.paic.com.cn/file/uid/53edee2646c74373ab4239668f0641e0)]
异步发送消息是指，Producer发出消息后无需等待MQ返回ACK，直接发送下⼀条消息。该方式的消息可靠性可以得到保障，消息发送效率也可以。

单向发送消息
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xitJODk3-1638173732553)(http://wb.paic.com.cn/file/uid/90d33c53894f4a57a4c8324ab69e15b1)]
单向发送消息是指，Producer仅负责发送消息，不等待、不处理MQ的ACK。该发送方式时MQ也不返回ACK。该方式的消息发送效率最高，但消息可靠性较差。

顺序消息

什么是顺序消息

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

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

顺序消息包含两种类型：

分区顺序：一个queue内所有的消息按照先进先出的顺序进行发布和消费
全局顺序：一个Topic内所有的消息按照先进先出的顺序进行发布和消费

为什么需要顺序消息

现在有TOPIC ORDER_STATUS (订单状态)，其下有4个Queue队列，该Topic中的不同消息用于描述当前订单的不同状态。假设订单有状态：未支付、已支付、发货中、发货成功、发货失败。
根据以上订单状态，生产者从时序上可以生成如下几个消息：

订单T0000001:未支付 --> 订单T0000001:已支付 --> 订单T0000001:发货中 --> 订单T0000001:发货失败

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

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BS0HHl96-1638173732555)(http://wb.paic.com.cn/file/uid/c97d7b33843d4e3c8136423603fd789a)]

这种情况下，我们希望Consumer消费消息的顺序和我们发送是一致的，然而上述MQ的投递和消费方
式，是无法保证顺序是正确的。

基于上述的情况，可以设计如下方案：对于相同订单号的消息，通过一定的策略，将其放置在一个
Queue中，然后消费者采用一定队列选择算法策略（例如，一个线程独立处理一个queue，保证处理消息的顺序
性），能够保证消费的顺序性。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-1yi2Xd8Y-1638173732557)(http://wb.paic.com.cn/file/uid/a520597a756d46d39883248ae76ee755)]

示例代码

        DefaultMQProducer producer = new DefaultMQProducer("producer-group");producer.setNamesrvAddr("127.0.0.1:9876");// 若为全局有序，则需要设置Queue数量为1// producer.setDefaultTopicQueueNums(1);producer.start();for (int i = 0; i < 100; i++) {// 为了演示简单，使用整型数作为orderIdInteger orderNo = i;byte[] body = ("Hello," + i).getBytes();Message msg = new Message("TopicA", "TagA", body);// 将orderId作为消息keymsg.setKeys(orderNo.toString());// send()的第三个参数值会传递给选择器的select()的第三个参数// 该send()为同步发送SendResult sendResult = producer.send(msg, new MessageQueueSelector() {// 具体的选择算法在该方法中定义@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {// 使用消息key作为选择的选择算法String keys = msg.getKeys();Integer orderNo = Integer.valueOf(keys);int index = orderNo % mqs.size();return mqs.get(index);}}, orderNo);System.out.println(sendResult);}producer.shutdown();

事务消息

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ERAYKUJH-1638173732559)(http://wb.paic.com.cn/file/uid/a92bb73210604f4983394f7359f381a8)]

事务消息就是MQ提供的类似XA的分布式事务能力，通过事务消息可以达到分布式事务的最终一致性。半事务消息就是MQ收到了生产者的消息，但是没有收到二次确认，不能投递的消息。实现原理如下：

生产者先发送一条半事务消息到MQ
MQ收到消息后返回ack确认
生产者开始执行本地事务
如果事务执行成功发送commit到MQ，失败发送rollback
如果MQ长时间未收到生产者的二次确认commit或者rollback，MQ对生产者发起消息回查
生产者查询事务执行最终状态
根据查询事务状态再次提交二次确认

如果MQ收到二次确认commit，就可以把消息投递给消费者，反之如果是rollback，消息会保存下来并且在3天后被删除。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Hk2nsWes-1638173732560)(http://wb.paic.com.cn/file/uid/be2bf9a950c5449ebfc2e9509dcd8ce4)]

延时消息

当消息写入到Broker后，在指定的时长后才可被消费处理的消息，称为延时消息。
采用RocketMQ的延时消息可以实现定时任务的功能，而无需使用定时器。典型的应用场景是订单超时未支付自动取消

延时等级

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

    // 延时等级分类private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

若指定的延时等级为3，则表示延迟时长为10s，即延迟等级是从1开始计数的
延时等级可在broker配置文件中自定义

延时消息实现原理
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fJUUi29x-1638173732562)(http://wb.paic.com.cn/file/uid/487c4467a98f4fbeb97b93c525294037)]
Producer将消息发送到Broker后，Broker会首先将消息写入到commitlog文件，然后需要将其分发到相应consumequeue。不过，在分发之前，系统会先判断消息中是否带有延时等级。若没有，则直接正常分发；若有则需要经历一个复杂的过程：

修改消息的Topic为SCHEDULE_TOPIC_XXXX
根据延时等级，在consumequeue目录中SCHEDULE_TOPIC_XXXX主题下创建出相应的queueId目录与consumequeue文件（如果没有这些目录与文件的话）。
修改消息索引单元内容。索引单元中的Message Tag HashCode部分原本存放的是消息的Tag的
Hash值。现修改为消息的投递时间。投递时间是指该消息被重新修改为原Topic后再次被写入到
commitlog中的时间。投递时间 = 消息存储时间 + 延时等级时间。消息存储时间指的是消息
被发送到Broker时的时间戳。
将消息索引写入到SCHEDULE_TOPIC_XXXX主题下相应的consumequeue中
一条消息从生产到被消费，将会经历三个阶段：

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

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

        DefaultMQProducer producer = new DefaultMQProducer("producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();for (int i = 0; i < 100; i++) {byte[] body = ("Hello," + i).getBytes();Message msg = new Message("TopicB", "TagB", body);// 指定消息延迟等级为3级，即延迟10smsg.setDelayTimeLevel(3);SendResult sendResult = producer.send(msg);// 输出消息被发送的时间System.out.print(new SimpleDateFormat("hh:mm:ss").format(new Date())+"，"+sendResult);}producer.shutdown();

2、消息的发送过程

Producer可以将消息写入到某Broker中的某Queue中，其经历了如下过程：

Producer发送消息之前，会先向NameServer发出获取消息Topic的路由信息的请求
NameServer返回该Topic的路由表及Broker列表
Producer根据代码中指定的Queue选择策略，从Queue列表中选出一个队列，用于后续存储消息
Produer对消息做一些特殊处理，例如，消息本身超过4M，则会对其进行压缩
Producer向选择出的Queue所在的Broker发出RPC请求，将消息发送到选择出的Queue

3、消息发送queue选择算法

rocketmq底层默认queue选择算法为轮询方式，并且rocketmq提供一些可选则的队列选择算法

SelectMessageQueueByHash      // 哈希取模算法
SelectMessageQueueByRandom   // 随机数算法
SelectMessageQueueByMachineRoom    // rocketmq暂无实现，需要用户自己去实现具体逻辑
自定义队列选择算法

随机数算法：

public class SelectMessageQueueByHash implements MessageQueueSelector {@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {int value = arg.hashCode() % mqs.size();if (value < 0) {value = Math.abs(value);}return mqs.get(value);}
}

自定义队列选择算法

    // 自定义选择队列算法SendResult sendResult = producer.send(msg, new MessageQueueSelector() {// 具体的选择算法在该方法中定义@Overridepublic MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {// 使用消息key作为选择的选择算法String keys = msg.getKeys();Integer orderNo = Integer.valueOf(keys);int index = orderNo % mqs.size();return mqs.get(index);}}, null);

4、消息发送源码解析

        // 创建一个producer，参数为Producer Group名称DefaultMQProducer producer = new DefaultMQProducer("sync-product-group");// 指定nameServer地址producer.setNamesrvAddr("127.0.0.1:9876");// 设置当发送失败时重试发送的次数，默认为2次producer.setRetryTimesWhenSendFailed(3);// 设置发送超时时限为5s，默认3sproducer.setSendMsgTimeout(5000);// 生产者启动producer.start();// 生产并发送100条消息for (int i = 0; i < 100; i++) {byte[] body = ("hello-" + i).getBytes();Message msg = new Message("testTopic", "testTag", body);// 为消息指定keymsg.setKeys("key-" + i);// 同步发送消息SendResult sendResult = producer.send(msg);System.out.println(sendResult);}// 关闭producerproducer.shutdown();

消息对应的Topic信息以及具体内容被封装在了Message中，并交由DefaultMQProducer，调用send（）进行发送。DefaultMQProducer 只是一个面向调用方的代理，真正的生产者是DefaultMQProducerImpl，而消息发送的具体实现，便在DefaultMQProducerImpl中的这个方法内：

private SendResult sendDefaultImpl(//Message msg,//final CommunicationMode communicationMode,//final SendCallback sendCallback, final long timeout//)

第一步：找到Topic对应的路由信息

  TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());

首先当然是进行本地查表，本地路由信息存放在topicPublishInfoTable中。但是如果本地没有，则会向NameSrv发起请求，获取路由信息，更新本地路由表。接着再次尝试从本地路由表中获取路由信息。

// 本地查表
TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
// 本地缓存中没有，便向NameSrv发起请求，更新本地路由缓存。
if (null == topicPublishInfo || !topicPublishInfo.ok()) {this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);topicPublishInfo = this.topicPublishInfoTable.get(topic);
}

如果比较幸运，从NameSrv上查询到了，此处便会直接返回所找到的路由信息：topicPublishInfo。但是如果Topic事先没有在任何Broker上进行配置，那么Broker在向NameSrv注册路由信息时便不会带上该Topic的路由，所以生产者也就无法从NameSrv中查询到该Topic的路由了。

if (topicPublishInfo.isHaveTopicRouterInfo() || (topicPublishInfo != null && topicPublishInfo.ok())) {return topicPublishInfo;
} else {// 再次查询Topic路由this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);topicPublishInfo = this.topicPublishInfoTable.get(topic);return topicPublishInfo;
}

对于这种没有事先配置Topic的情况，RocketMQ不会直接抛出错误，而是会走到上面的else分支里，再次调用 updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer)，从NameSrv 获取路由信息。

   public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault,DefaultMQProducer defaultMQProducer) {......TopicRouteData topicRouteData;if (isDefault && defaultMQProducer != null) {// 查询默认Topic TBW102的路由信息topicRouteData =this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(), 1000 * 3);}......// 克隆一份，放到路由表中TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();......this.topicRouteTable.put(topic, cloneTopicRouteData);
}

这次调用 updateTopicRouteInfoFromNameServer（）时，传入的参数 isDefault 为true，那么代码自然就进入了上面的 if 分支中。这里依旧是调用getDefaultTopicRouteInfoFromNameServer( defaultMQProducer.getCreateTopicKey(),1000*3) 从NameSrv查询Topic路由，不过此次不是查询消息所属Topic的路由信息，而是查询RocketMQ设置的一个默认Topic的路由，该默认Topic为 TBW102 ，这个Topic就是用来创建其他Topic所用的。

如果某Broker配置了 autoCreateTopicEnable，允许自动创建Topic，那么在该Broker启动后，便会向自己的路由表中插入TBW102这个Topic，并注册到NameSrv，表明处理该Topic类型的消息。

所以当消息所属的Topic，暂且叫Topic X吧，它本身没有在任何Broker上配置的时候，生产者就会查询Topic TBW102的路由信息，暂时作为Topic X的的路由，并插入到本地路由表中。当TopicX利用该路由发送到 Broker后，Broker发现自己并没有该Topic信息后，便会创建好该Topic，并更新到NameSrv中，表明后续接收TopicX的消息。

获取topic路由大致流程如下：

1、先从本地缓存的路由表中查询；

2、没有找到的话，便向NameSrv发起请求，更新本地路由表，再次查询。

3、如果仍然没有查询到，表明Topic没有事先配置，则用Topic TBW102向NameSrv发起查询，返回TBW102的路由信息，暂时作为Topic的路由。

第二步：选择某个Queue用来发送消息

前面提到每个Topic的路由信息中可能包含若干Queue，那么这些Queue是从哪来的呢？不管怎么样，这些Queue的信息肯定是NameSrv返回的。生产者从NameSrv拉取的路由信息为TopicRouteData，我们不妨先来看下它的结构：

public class TopicRouteData extends RemotingSerializable {private List<QueueData> queueDatas;private List<BrokerData> brokerDatas;......
}

queueDatas 中包含了Topic对应的所有Queue信息，其中QueueData结构如下：

public class QueueData implements Comparable<QueueData> {private String brokerName;private int readQueueNums;private int writeQueueNums;
}

对于RokcetMQ来说，Queue是比较抽象的一个概念，并不是说某个具体的队列。Topic、QueueData以及Broker是 1:1:1 的，QueueData本质上是记录某个Topic在某个Broker上的所有路由信息。

brokerName：这个很容易理解，Queue所属的Broker；
readQueueNums：该Broker上，针对该Topic，配置的读队列个数；
writeQueueNums：该Broker上，针对该Topic，配置的写队列个数。

前面的第一步中，当生产者从NameSrv获取到Topic对于的TopicRouteData时，会将其转成TopicPublishInfo,存放在本地路由表中。

在topicRouteData2TopicPublishInfo（topic, topicRouteData）内部转化过程中，便会遍历TopicRouteData中的QueueData，按照配置的读写队列个数，生成MessageQueue，存放在本地queue表中。// Update Pub info
{// 转换成TopicPublishInfoTopicPublishInfo publishInfo =topicRouteData2TopicPublishInfo(topic, topicRouteData);publishInfo.setHaveTopicRouterInfo(true);Iterator<Entry<String, MQProducerInner>> it =this.producerTable.entrySet().iterator();while (it.hasNext()) {Entry<String, MQProducerInner> entry = it.next();MQProducerInner impl = entry.getValue();if (impl != null) {// 更新本地路由表impl.updateTopicPublishInfo(topic, publishInfo);}}
}

在topicRouteData2TopicPublishInfo（topic, topicRouteData）内部转化过程中，便会遍历TopicRouteData中的QueueData，按照配置的读写队列个数，生成MessageQueue，存放在本地queue表中。

List<QueueData> qds = route.getQueueDatas();
Collections.sort(qds);// 开始遍历queueDatas
for (QueueData qd : qds) {if (PermName.isWriteable(qd.getPerm())) {// 查询QueueData对应的BrokerDataBrokerData brokerData = null;for (BrokerData bd : route.getBrokerDatas()) {if (bd.getBrokerName().equals(qd.getBrokerName())) {brokerData = bd;break;}}if (null == brokerData) {continue;}// 只有Master节点的Broker才能接收消息，对于非Master节点的需要过滤掉if (!brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) {continue;}// 按照QueueData配置的写队列个数，生成对应数量的MessageQueue。for (int i = 0; i < qd.getWriteQueueNums(); i++) {MessageQueue mq = new MessageQueue(topic, qd.getBrokerName(), i);info.getMessageQueueList().add(mq);}}
}

回到 sendDefaultImpl（）中，当拿到路由信息后，要开始进行消息发送。下面这部分代码我省略了部分。简单点说，主要逻辑就是在消息发送的基础上加上了超时机制及**重试机制。**当选择某个Queue发送消息失败后，只要还没有超时，且没有超出最大重试次数，就会再次选择某个Queue进行重试。

// 在超时时间及重试次数内进行重试
for (; times < timesTotal && (endTimestamp - beginTimestamp) < maxTimeout; times++) {String lastBrokerName = null == mq ? null : mq.getBrokerName();// 选择某个Queue 用来发送消息MessageQueue tmpmq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);if (tmpmq != null) {mq = tmpmq;brokersSent[times] = mq.getBrokerName();try {// 进行消息发送sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);endTimestamp = System.currentTimeMillis();......}catch (Exception e) {endTimestamp = System.currentTimeMillis();continue;}} else {break;}
} if (sendResult != null) {return sendResult;
}

由上面代码可知，选择Queue的具体逻辑在topicPublishInfo.selectOneMessageQueue（lastBrokerName）中。这里在调用时传入了lastBrokerName

public MessageQueue selectOneMessageQueue(final String lastBrokerName) {if (lastBrokerName != null) {int index = this.sendWhichQueue.getAndIncrement();for (int i = 0; i < this.messageQueueList.size(); i++) {int pos = Math.abs(index++) % this.messageQueueList.size();MessageQueue mq = this.messageQueueList.get(pos);if (!mq.getBrokerName().equals(lastBrokerName)) {return mq;}}return null;}else {int index = this.sendWhichQueue.getAndIncrement();int pos = Math.abs(index) % this.messageQueueList.size();return this.messageQueueList.get(pos);}
}

1、当lastBrokerName不为空时，将计数器进行自增，再遍历TopicPulishInfo中的MessageQueue列表，按照计数器数值对MessageQueue总个数进行取模，再根据取模结果，取出MessageQueue列表中的某个Queue，并判断Queue所属Broker的Name是否和lastBrokerName一致，一致则继续遍历。

2、当lastBrokerName为空时，同样将计数器进行自增，按照计数器数值对MessageQueue总个数进行取模，再根据取模结果，取出MessageQueue列表中的某个Queue，直接返回。

3、从以上过程中可以发现，消费发送选择queue算法为轮询方式

当某条消息第一次发送时，lastBrokerName 为空，此时就是直接取模进行负载均衡操作。但是如果消息发送失败，就会触发重试机制，发送失败有可能是因为Broker出现来某些故障，或者某些网络连通性问题，所以当消息第N次重试时，就要避开第N-1次时消息发往的Broker，也就是lastBrokerName。

第三步：消息发送的核心过程

消息的网络传输在 DefaultMQProducerImpl的sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout)中。

首先，要获取Queue所属Broker的地址：

String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
if (null == brokerAddr) {tryToFindTopicPublishInfo(mq.getTopic());brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
}

拿到Broker地址后，要将消息内容及其他信息封装进请求头：

SendMessageRequestHeader requestHeader = new SendMessageRequestHeader();
requestHeader.setProducerGroup(this.defaultMQProducer.getProducerGroup());
requestHeader.setTopic(msg.getTopic());
requestHeader.setDefaultTopic(this.defaultMQProducer.getCreateTopicKey());
requestHeader.setDefaultTopicQueueNums(this.defaultMQProducer.getDefaultTopicQueueNums());
requestHeader.setQueueId(mq.getQueueId());
......

接着调用MQClientAPIImpl的sendMessage（）方法：

 SendResult sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(//brokerAddr,// 1mq.getBrokerName(),// 2msg,// 3requestHeader,// 4timeout,// 5communicationMode,// 6sendCallback// 7);

sendMessage（）内部便是创建请求，调用封装的Netty进行网络传输了。

首先创建请求：

RemotingCommand request = null;
if (sendSmartMsg) {SendMessageRequestHeaderV2 requestHeaderV2 =SendMessageRequestHeaderV2.createSendMessageRequestHeaderV2(requestHeader);request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE_V2, requestHeaderV2);
}
else {request = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, requestHeader);
}

这里按照是否发送 smartMsg ，创建了不同请求命令号的请求，接下来，按照发送方式（单向、同步、异步），调用不同的发送函数：

switch (communicationMode) {case ONEWAY:this.remotingClient.invokeOneway(addr, request, timeoutMillis);return null;case ASYNC:this.sendMessageAsync(addr, brokerName, msg, timeoutMillis, request, sendCallback);return null;case SYNC:return this.sendMessageSync(addr, brokerName, msg, timeoutMillis, request);default:assert false;break;
}

这里解释下三种发送方式。

单向：只管发送，不管是否发送成功；
同步：阻塞至拿到发送结果；
异步：发送后直接返回，在回调函数中等待发送结果。

到这里，消息的发送就已经结束了，成功的从生产者传输到了Broker

broker消息接收

broker在启动过程中，注册了各种处理类，其中就有用与接收消息处理类SendMessageProcessor

BrokerController#registerProcessor 方法注册了各种处理类

this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);

SendMessageProcessor实现了NettyRequestProcessor接口，并在接口方法 processRequest()中处理接收到的请求,SendMessageProcessor在processRequest()中调用了 asyncSendMessage()方法来进行消息处理

    public CompletableFuture<RemotingCommand> asyncProcessRequest(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {final SendMessageContext mqtraceContext;switch (request.getCode()) {//消费者发送返回消息case RequestCode.CONSUMER_SEND_MSG_BACK:return this.asyncConsumerSendMsgBack(ctx, request);default://解析发送消息请求头SendMessageRequestHeader requestHeader = parseRequestHeader(request);if (requestHeader == null) {return CompletableFuture.completedFuture(null);}//发送消息实体mqtraceContext = buildMsgContext(ctx, requestHeader);this.executeSendMessageHookBefore(ctx, request, mqtraceContext);//是否批量发送if (requestHeader.isBatch()) {return this.asyncSendBatchMessage(ctx, request, mqtraceContext, requestHeader);} else {                    return this.asyncSendMessage(ctx, request, mqtraceContext, requestHeader);}}}

继续看asyncSendMessage方法内部逻辑

    private CompletableFuture<RemotingCommand> asyncSendMessage(ChannelHandlerContext ctx, RemotingCommand request,SendMessageContext mqtraceContext,SendMessageRequestHeader requestHeader) {//创建响应结果final RemotingCommand response = preSend(ctx, request, requestHeader);//发送消息响应头final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader)response.readCustomHeader();if (response.getCode() != -1) {return CompletableFuture.completedFuture(response);}//消息体final byte[] body = request.getBody();//队列idint queueIdInt = requestHeader.getQueueId();//根据topic获取topic配置TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());//队列小于0，随机给一个if (queueIdInt < 0) {queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());}//省略部分代码...CompletableFuture<PutMessageResult> putMessageResult = null;String transFlag = origProps.get(MessageConst.PROPERTY_TRANSACTION_PREPARED);if (transFlag != null && Boolean.parseBoolean(transFlag)) {//如果拒绝事务消息if (this.brokerController.getBrokerConfig().isRejectTransactionMessage()) {response.setCode(ResponseCode.NO_PERMISSION);response.setRemark("the broker[" + this.brokerController.getBrokerConfig().getBrokerIP1()+ "] sending transaction message is forbidden");return CompletableFuture.completedFuture(response);}//处理事务消息putMessageResult = this.brokerController.getTransactionalMessageService().asyncPrepareMessage(msgInner);} else {//处理普通消息putMessageResult = this.brokerController.getMessageStore().asyncPutMessage(msgInner);}return handlePutMessageResultFuture(putMessageResult, response, request, msgInner, responseHeader, mqtraceContext, ctx, queueIdInt);}

最终是通过调用DefaultMessageStore#asyncPutMessage方法来进行消息的存储操作

    @Overridepublic CompletableFuture<PutMessageResult> asyncPutMessage(MessageExtBrokerInner msg) {//检查store状态PutMessageStatus checkStoreStatus = this.checkStoreStatus();if (checkStoreStatus != PutMessageStatus.PUT_OK) {return CompletableFuture.completedFuture(new PutMessageResult(checkStoreStatus, null));}// 消息是否合法PutMessageStatus msgCheckStatus = this.checkMessage(msg);if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {return CompletableFuture.completedFuture(new PutMessageResult(msgCheckStatus, null));}long beginTime = this.getSystemClock().now();//保存消息CompletableFuture<PutMessageResult> putResultFuture = this.commitLog.asyncPutMessage(msg);putResultFuture.thenAccept((result) -> {long elapsedTime = this.getSystemClock().now() - beginTime;if (elapsedTime > 500) {log.warn("putMessage not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);}//统计保存消息花费的时间this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);if (null == result || !result.isOk()) {//统计保存消息失败次数this.storeStatsService.getPutMessageFailedTimes().add(1);}});return putResultFuture;}

commitLog的asyncPutMessage的方法真正把消息存储到commitLog文件中，至此消息被持久化到磁盘中。

消息存储

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MLPo0tv5-1638173732564)(http://wb.paic.com.cn/file/uid/a2af5aabbaed47b2ba0bbc8d01dcb9d6)]

每个Broker都对应有一个MessageStore，专门用来存储发送到它的消息，不过MessageStore本身不是文件，只是存储的一个抽象，MessageStore 中保存着一个 CommitLog，CommitLog 维护了一个 MappedFileQueue，而MappedFileQueue 中又维护了多个 MappedFile，每个MappedFile都会映射到文件系统中一个文件，这些文件才是真正的存储消息的地方，MappedFile的文件名为它记录的第一条消息的全局物理偏移量。

消息存储流程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-VpgnmBfL-1638173732565)(http://wb.paic.com.cn/file/uid/a8551269343e44b1a9c9d4a53f0ac382)]

消息存储目录

RocketMQ中的消息存储在本地文件系统中，这些相关文件默认在当前用户主目录下的store目录中。

abort：该文件在Broker启动后会自动创建，正常关闭Broker，该文件会自动消失。若在没有启动Broker的情况下，发现这个文件是存在的，则说明之前Broker的关闭是非正常关闭。
checkpoint：其中存储着commitlog、consumequeue、index文件的最后刷盘时间戳
commitlog：其中存放着commitlog文件，而消息是写在commitlog文件中的
config：存放着Broker运行期间的一些配置数据
consumequeue：其中存放着consumequeue文件，队列就存放在这个目录中
index：其中存放着消息索引文件indexFile
lock：运行期间使用到的全局资源锁

1、commitlog文件

一个Broker中仅包含一个commitlog目录，所有的mappedFile文件都是存放在该目录中的。即无论当前Broker中存放着多少Topic的消息，这些消息都是被顺序写入到了mappedFile文件中的。也就是说，这些消息在Broker中存放时并没有被按照Topic进行分类存放。

消息单元
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ARKiKdBz-1638173732566)(http://wb.paic.com.cn/file/uid/7074c959b1cd4f82946ae3b748d4e617)]

mappedFile文件内容由一个个的消息单元构成。每个消息单元中包含消息总长度MsgLen、消息的物理位置physicalOffset、消息体内容Body、消息体长度BodyLength、消息主题Topic、Topic长度TopicLength、消息生产者BornHost、消息发送时间戳BornTimestamp、消息所在的队列QueueId、消息在Queue中存储的偏移量QueueOffset等近20余项消息相关属性。

2、consumequeue

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-dNp8hW1N-1638173732567)(http://wb.paic.com.cn/file/uid/599af7a8ea744f97afeb343a7077aef1)]

为了提高效率，会为每个Topic在~/store/consumequeue中创建一个目录，目录名为Topic名称。在该Topic目录下，会再为每个该Topic的Queue建立一个目录，目录名为queueId。每个目录中存放着若干consumequeue文件，consumequeue文件是commitlog的索引文件，可以根据consumequeue定位到具体的消息

consumequeue文件名也由20位数字构成，表示当前文件的第一个索引条目的起始位移偏移量。与mappedFile文件名不同的是，其后续文件名是固定的。因为consumequeue文件大小是固定不变的。

索引条目
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8TMOee2A-1638173732568)(http://wb.paic.com.cn/file/uid/1938160e1f86444a9e4a8bf12f26ad1e)]

每个consumequeue文件可以包含30w个索引条目，每个索引条目包含了三个消息重要属性：消息在
mappedFile文件中的偏移量CommitLog Offset、消息长度、消息Tag的hashcode值。这三个属性占20
个字节，所以每个文件的大小是固定的30w * 20字节。

3、对文件的读写

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PrTekNI0-1638173732569)(http://wb.paic.com.cn/file/uid/75ca9cf9d9844f24a356d6870222aa0e)]

消息写入

一条消息进入到Broker后经历了以下几个过程才最终被持久化。

Broker根据queueId，获取到该消息对应索引条目要在consumequeue目录中的写入偏移量，即QueueOffset
将queueId、queueOffset等数据，与消息一起封装为消息单元
将消息单元写入到commitlog
同时，形成消息索引条目
将消息索引条目分发到相应的consumequeue

消息刷盘
RocketMQ 消息刷盘主要分为同步刷盘和异步刷盘

同步刷盘：只有在消息真正持久化至磁盘后 RocketMQ 的 Broker 端才会真正返回给 Producer 端一个成功的 ACK 响应。同步刷盘对 MQ 消息可靠性来说是一种不错的保障，但是性能上会有较大影响
异步刷盘：能够充分利用 OS 的 Page Cache 的优势，只要消息写入 Page Cache 即可将成功的 ACK 返回给 Producer 端。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了 MQ 的性能和吞吐量。

消息拉取

当Consumer来拉取消息时会经历以下几个步骤：

Consumer获取到其要消费消息所在Queue的消费偏移量offset ，计算出其要消费消息的消息offset

消费offset就是消费进度，consumer对某个Queue的消费offset，即消费到了该Queue的第几条消息
Consumer向Broker发送拉取请求，其中会包含其要拉取消息的Queue、消息offset及消息Tag。
Broker计算在该consumequeue中的queueOffset。

queueOffset = 消息offset * 20字节
从该queueOffset处开始向后查找第一个指定Tag的索引条目。
解析该索引条目的前8个字节，即可定位到该消息在commitlog中的commitlog offset
从对应commitlog offset中读取消息单元，并发送给Consumer

消息清理

Broker中的消息被消费后不会立即删除，每条消息都会持久化到CommitLog中，每个Consumer连接到Broker后会维持消费进度信息，当有消息消费后只是当前Consumer的消费进度（CommitLog的offset）更新了。默认72小时后会删除不再使用的CommitLog文件：

检查这个文件最后访问时间
判断是否大于过期时间
指定时间删除，默认凌晨4点

消息消费

消费者从Broker中获取消息的方式有两种：pull拉取方式和push推动方式。消费者组对于消息消费的模式又分为两种：集群消费Clustering和广播消费Broadcasting。

1、消费类型

拉取式消费
Consumer主动从Broker中拉取消息，主动权由Consumer控制。一旦获取了批量消息，就会启动消费过程。不过，该方式的实时性较弱，即Broker中有了新的消息时消费者并不能及时发现并消费。

推送式消费
broker服务端有数据之后立马推送消息给客户端，需要客户端和服务器建立长连接，该获取方式一般实时性较高。缺点就是服务端不知道客户端处理消息的能力，可能会导致数据积压，同时也增加了服务端的工作量，影响服务端的性能

RocketMQ中push和pull模式的底层实现都是拉取消息，push模式底层实现采用的是长轮询机制，Broker端属性 longPollingEnable 标记是否开启长轮询，默认开启。源码如下：

    /***  是否开启长轮询*/private boolean longPollingEnable = true;

长轮询通过客户端和服务端的配合，达到主动权在客户端，同时也能保证数据的实时性；长轮询本质上也是轮询，只不过对普通的轮询做了优化处理，服务端在没有数据的时候并不是马上返回数据，会 hold 住请求，等待服务端有数据，或者一直没有数据超时处理，然后一直循环下去；

broker端处理拉取请求代码如下：

PullMessageProcessor#processRequest

 switch (response.getCode()) {case ResponseCode.SUCCESS:...省略...break;case ResponseCode.PULL_NOT_FOUND:if (brokerAllowSuspend && hasSuspendFlag) {long pollingTimeMills = suspendTimeoutMillisLong;if (!this.brokerController.getBrokerConfig().isLongPollingEnable()) {pollingTimeMills = this.brokerController.getBrokerConfig().getShortPollingTimeMills();}String topic = requestHeader.getTopic();long offset = requestHeader.getQueueOffset();int queueId = requestHeader.getQueueId();PullRequest pullRequest = new PullRequest(request, channel, pollingTimeMills,this.brokerController.getMessageStore().now(), offset, subscriptionData);// 暂时挂起当前请求一段时间，等有新的消息进来后再返回消费端this.brokerController.getPullRequestHoldService().suspendPullRequest(topic, queueId, pullRequest);response = null;break;}case ResponseCode.PULL_RETRY_IMMEDIATELY:break;case ResponseCode.PULL_OFFSET_MOVED:...省略...break;default:assert false;

push模式核心类
DefaultMQPushConsumer

2、消费模式

消息模型：集群（Clustering）和广播（Broadcasting）

集群模式(Clustering)

集群消费模式下，相同消费者组的每个Consumer实例平均分摊同一个Topic的消息。即每条消
息只会被发送到Consumer Group中的某个Consumer

集群消费

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-M4rCMY8n-1638173732570)(http://wb.paic.com.cn/file/uid/e3a5b025a95d434d90e4b357f453b052)]

每条消息只需要被处理一次，Broker只会把消息发送给消费集群中的一个消费者
在消息重投时，不能保证路由到同一台机器上
消费状态由Broker维护

广播模式(Broadcasting)

广播消费模式下，相同Consumer Group的每个Consumer实例都接收同一个Topic的全量消息。即每条
消息都会被发送到Consumer Group中的每个Consumer。

广播消费

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YjkCv1hz-1638173732571)(http://wb.paic.com.cn/file/uid/6c0c6fe286164de1867396b3dc243c00)]

消费进度由Consumer维护
保证每个消费者都消费一次消息
消费失败的消息不会重投

消息进度保存

广播模式：消费进度保存在consumer端。因为广播模式下consumer group中每个consumer都会消费所有消息，但它们的消费进度是不同。所以consumer各自保存各自的消费进度。
集群模式：消费进度保存在broker中。consumer group中的所有consumer共同消费同一个Topic中的消息，同一条消息只会被消费一次。消费进度会参与到了消费的负载均衡中，故消费进度是需要共享的。下图是broker中存放的各个Topic的各个Queue的消费进度。

3、Rebalance机制

Rebalance即再均衡，指的是，将⼀个Topic下的多个Queue在同⼀个Consumer Group中的多个Consumer间进行重新分配的过程，只有集群模式下才会进行Rebalance操作，广播模式下不存在。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FLzS6L1G-1638173732572)(http://wb.paic.com.cn/file/uid/eacccba8f5cd440993ec40d5ce41c340)]

Rebalance机制的本意是为了提升消息的并行消费能力。例如，⼀个Topic下5个队列，在只有1个消费者的情况下，这个消费者将负责消费这5个队列的消息。如果此时我们增加⼀个消费者，那么就可以给其中⼀个消费者分配2个队列，给另⼀个分配3个队列，从而提升消息的并行消费能力。

Rebalance限制
由于⼀个队列最多分配给⼀个消费者，因此当某个消费者组下的消费者实例数量大于队列的数量时，
多余的消费者实例将分配不到任何队列。

Rebalance产生的问题

消费暂停：在只有一个Consumer时，其负责消费所有队列；在新增了一个Consumer后会触发Rebalance的发生。此时原Consumer就需要暂停部分队列的消费，等到这些队列分配给新的Consumer后，这些暂停消费的队列才能继续被消费。

消费重复：Consumer 在消费新分配给自己的队列时，必须接着之前Consumer 提交的消费进度的offset继续消费。然而默认情况下，offset是异步提交的，这个异步性导致提交到Broker的offset与Consumer实际消费的消息并不一致。这个不一致的差值就是可能会重复消费的消息。

消费突刺：由于Rebalance可能导致重复消费，如果需要重复消费的消息过多，或者因为Rebalance暂停时间过长从而导致积压了部分消息。那么有可能会导致在Rebalance结束之后瞬间需要消费很多消息。

Rebalance产生的原因
导致Rebalance产生的原因有两个：消费者所订阅Topic的Queue数量发生变化，或消费者组中消费者的数量发生变化。

Rebalance过程
在Broker中维护着多个Map集合，这些集合中动态存放着当前Topic中Queue的信息、Consumer Group中Consumer实例的信息。一旦发现消费者所订阅的Queue数量发生变化，或消费者组中消费者的数量发生变化，立即向Consumer Group中的每个实例发出Rebalance通知。

4、Queue分配算法

一个Topic中的Queue只能由Consumer Group中的一个Consumer进行消费，而一个Consumer可以同时消费多个Queue中的消息。那么Queue与Consumer间的配对关系是如何确定的，即Queue要分配给哪个Consumer进行消费，也是有算法策略的。常见的有四种策略。这些策略是通过在创建Consumer时的构造器传进去的。

AllocateMessageQueueAveragely   //平均分配策略(默认分配算法)
AllocateMessageQueueAveragelyByCircle   //环形平均策略
AllocateMessageQueueConsistentHash     //一致性hash策略
AllocateMessageQueueByConfig     //手动配置分配策略
AllocateMessageQueueByMachineRoom    //机房分配策略
AllocateMachineRoomNearby             //靠近机房策略

平均分配策略(默认分配算法)

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8p41lMnh-1638173732574)(http://wb.paic.com.cn/file/uid/c3ab21b0ff574e688ee72c5be2f9c32d)]

该算法是要根据avg = QueueCount / ConsumerCount 的计算结果进行分配的。如果能够整除，则按顺序将avg个Queue逐个分配Consumer；如果不能整除，则将多余出的Queue按照Consumer顺序逐个分配。

平均分配算法源码：

    @Overridepublic List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,List<String> cidAll) {// 此处省略参数校验等逻辑...// 计算当前消费者在消费者集合cidALL中下标的位置int index = cidAll.indexOf(currentCID);// 计算当前消息队列 Queue中的消息是否能被消费者集合cidAll平均分配完int mod = mqAll.size() % cidAll.size();// 计算当前消费者消费的平均数量int averageSize = mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size()  + 1 : mqAll.size() / cidAll.size());//计算当前消费者开始消费消息的下标int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;int range = Math.min(averageSize, mqAll.size() - startIndex);// 从 startIndex 开始的下标位置，加载数量为 range 的消息到 result 集合中，最后返回这个 resultfor (int i = 0; i < range; i++) {result.add(mqAll.get((startIndex + i) % mqAll.size()));}return result;}

环形平均策略
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Je7MMTo8-1638173732575)(http://wb.paic.com.cn/file/uid/1ea9adb3c7b64677aee6ba9b23fcb42c)]
环形平均算法是指，根据消费者的顺序，依次在由queue队列组成的环形图中逐个分配。

一致性hash策略
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mt6iG1Z6-1638173732576)(http://wb.paic.com.cn/file/uid/de4db1f308f74113973beb4127b20249)]
该算法会将consumer的hash值作为Node节点存放到hash环上，然后将queue的hash值也放到hash环上，通过顺时针方向，距离queue最近的那个consumer就是该queue要分配的consumer。

消息拉取

消息者客户端在启动时，就同步启动了消息拉取服务，具体是在MQClientInstance#start方法中启动拉取消息服务。

  // 启动拉取消息服务this.pullMessageService.start();

消息拉取的入口为PullMessageService#run()方法，while循环的获取拉取任务，再根据拉取任务拉取消息。

    public void run() {log.info(this.getServiceName() + " service started");while (!this.isStopped()) {try {//从pullRequestQueue中获取拉取任务，如果获取不到，线程阻塞，直到有任务可被拉取PullRequest pullRequest = this.pullRequestQueue.take();//根据拉取任务拉取消息this.pullMessage(pullRequest);} catch (InterruptedException ignored) {} catch (Exception e) {log.error("Pull Message Service Run Method exception", e);}}log.info(this.getServiceName() + " service end");}

消息拉取主体流程如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-P9kosaf7-1638173732577)(http://wb.paic.com.cn/file/uid/0c598c05c49d45a9a6bb097baa58b415)]

其拉取流程主要分如下3步：

1、拉取请求参数的封装；
2、消息服务器查找并返回消息；
3、客户端处理返回消息。

批量消息

1、批量发送消息

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

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

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

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

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

2、批量消费消息

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

批量生产消息示例代码

        DefaultMQProducer producer = new DefaultMQProducer("batch-producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();// 定义要发送的消息集合List<Message> messages = new ArrayList<>();for (int i = 0; i < 10000; i++) {byte[] body = ("Hello," + i).getBytes();Message msg = new Message("TopicD", "TagD", 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();

批量消费消息

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("batch-consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.subscribe("TopicD", "*");// 指定每次可以消费10条消息，默认为1consumer.setConsumeMessageBatchMaxSize(10);// 指定每次可以从Broker拉取40条消息，默认为32consumer.setPullBatchSize(40);consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic 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();

消息过滤

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

1、Tag过滤

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

        // 仅订阅Tag为myTagA与myTagB的消息，不包含myTagCconsumer.subscribe("TopicC", "myTagA || myTagB");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic 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");

2、SQL过滤

SQL过滤是一种通过特定表达式对事先埋入到消息中的用户属性进行筛选过滤的方式。通过SQL过滤，可以实现对消息的复杂过滤。不过，只有使用PUSH模式的消费者才能使用SQL过滤。
默认情况下Broker没有开启消息的SQL过滤功能，需要在Broker加载的配置文件中添加如下属性，以开启该功能：

enablePropertyFilter = true

定义SQL过滤Producer

        DefaultMQProducer producer = new DefaultMQProducer("producer-group");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();for (int i = 0; i < 10; i++) {try {byte[] body = ("Hello," + i).getBytes();Message msg = new Message("TopicE", "myTag", body);// 事先埋入用户属性agemsg.putUserProperty("age", i + "");SendResult sendResult = producer.send(msg);System.out.println(sendResult);} catch (Exception e) {e.printStackTrace();}}producer.shutdown();

定义SQL过滤Consumer

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);// 要从TopicE的消息中过滤出age在[0, 6]间的消息consumer.subscribe("TopicE", MessageSelector.bySql("age between 0 and 6"));consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic 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");

消息发送重试

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

生产者在发送消息时，若采用同步或异步发送方式，发送失败会重试，但oneway消息发送方式发送失败是没有重试机制的
只有普通消息具有发送重试机制，顺序消息没有重试机制
消息重投机制可以保证消息尽可能发送成功、不丢失，但可能会造成消息重复。消息重复在RocketMQ中是无法避免的问题
消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会成为大概率事件
producer主动重发、consumer负载变化（发生Rebalance，不会导致消息重复，但可能出现重复消费）也会导致重复消息

发送消息如果失败或者超时了，则会自动重试。默认是重试3次，可以根据api进行更改，比如改为10次：

producer.setRetryTimesWhenSendFailed(10);

底层源码逻辑如下：

/*** {@link org.apache.rocketmq.client.producer.DefaultMQProducer#sendDefaultImpl(Message, CommunicationMode, SendCallback, long)}*/// 自动重试次数，this.defaultMQProducer.getRetryTimesWhenSendFailed()默认为2，如果是同步发送，默认重试3次，否则重试1次
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
for (; times < timesTotal; times++) {// 选择发送的消息queueMessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);if (mqSelected != null) {try {// 真正的发送逻辑，sendKernelImpl。sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);switch (communicationMode) {case ASYNC:return null;case ONEWAY:return null;case SYNC:// 如果发送失败了，则continue，意味着还会再次进入for，继续重试发送if (sendResult.getSendStatus() != SendStatus.SEND_OK) {if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {continue;}}// 发送成功的话，将发送结果返回给调用者return sendResult;default:break;}} catch (RemotingException e) {continue;} catch (...) {continue;}}
}

消息消费重试

当消费模式为 MessageModel.CLUSTERING(集群模式) 时，Broker 会自动进行重试，对于广播消息是不会重试的。对于一直无法消费成功的消息，RocketMQ 会在达到最大重试次数之后，将该消息投递至死信队列。然后我们需要关注死信队列，并对该死信消息业务做人工的补偿操作。

顺序消息的消费重试

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

无序消息的消费重试

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

消费重试次数与间隔

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

重试次数	与上次重试的间隔时间
1	10秒
2	30秒
3	1分钟
4	2分钟
5	3分钟
6	4分钟
7	5分钟
8	6分钟
9	7分钟
10	8分钟
11	9分钟
12	10分钟
13	20分钟
14	30分钟
15	1小时
16	2小时

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

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

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

手动ACK确认

消费者会先把消息拉取到本地，然后进行业务逻辑，业务逻辑完成后手动进行ack确认，这时候才会真正的代表消费完成。而不是说pull到本地后消息就算消费完了。举个例子

 consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {try{for (MessageExt msg : msgs) {String str = new String(msg.getBody());System.out.println(str);}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;} catch(Throwable t){log.error("消费异常:{}", msgs, t);return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}});

消费异常自动重试

业务消费方返回 ConsumeConcurrentlyStatus.RECONSUME_LATER
业务消费方返回 null
业务消费方主动/被动抛出异常

针对以上3种情况下，Broker一般会进行重试（默认最大重试16次），RocketMQ 采用了“时间衰减策略”进行消息的重复投递，即重试次数越多，消息消费成功的可能性越小。

消费者客户端，首先判断消费端有没有显式设置最大重试次数 MaxReconsumeTimes，如果没有，则设置默认重试次数为 16，否则以设置的最大重试次数为准。

private int getMaxReconsumeTimes() {// default reconsume times: 16if (this.defaultMQPushConsumer.getMaxReconsumeTimes() == -1) {return 16;} else {return this.defaultMQPushConsumer.getMaxReconsumeTimes();}
}

重试队列

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

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

%RETRY%consumerGroup@consumerGroup

这个重试队列是针对消息组的，而不是针对每个Topic设置的（一个Topic的消息可以让多个消费者组进行消费，所以会为这些消费者组各创建一个重试队列）
只有当出现需要进行重试消费的消息时，才会为该消费者组创建重试队列
Broker对于重试消息的处理是通过延时消息实现的。先将消息保存到SCHEDULE_TOPIC_XXXX延迟队列中，延迟时间到后，会将消息投递到%RETRY%consumerGroup@consumerGroup重试队列中。

死信队列

死信的处理逻辑：

首先判断消息当前重试次数是否大于等于 16，或者消息延迟级别是否小于 0
只要满足上述的任意一个条件，设置新的 topic（死信 topic）为：%DLQ%+consumerGroup
进行前置属性的添加
将死信消息投递到上述步骤 2 建立的死信 topic 对应的死信队列中并落盘，使消息持久化

最后单独启动一个死信队列的消费者进行消费，然后进行人工干预处理失败的消息。

思考问题

1、为什么顺序消息发送失败不能重试
2、顺序消息消费失败指定重试次数为什么不会生效
3、DefaultLitePullConsumer的拉取消息流程是在哪里发起