分布式中间件RocketMQ

tag: 上海

tag: 北京

tag: 南京

producer.setDefaultTopicQueueNums(2);

producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode +当前时间 + AutomicInteger自增计数器

brokerIp + 物理分区的offset（Queue中的偏移量）

例如，业务系统产生的日志写入到MQ的过程，就是消息生产的过程

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

例如，QoS系统从MQ中读取日志，并对日志进行解析处理的过程就是消息消费的过程

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

说明：RocketMQ的思想来自于Kafka，而Kafka是依赖了Zookeeper的。所以，在RocketMQ的早期版本，也是依赖于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的信息。

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需要怎么 做？

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

扩展

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

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

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

组成架构图

Remoting Module

Client Manager

Store Service

HA Service

Index Service

为了增强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。

集群模式：该模式下创建的Topic在该集群中，所有Broker中的Queue数量是相同的

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

其这样设计的目的是为了，方便Topic的Queue的缩容

将这一行的初始内存修改为这样

将这一行的初始内存修改为这样

将localhost 替换为公网ip

默认值为listenPort+1

修改为一个不常用的端口比如 8888，默认是8080

例如：rocketmq.config.namesrvAddr=rocketmq.config.namesrvAddr=47.96.111.162:9876

注意：由于我使用的是阿里云的ECS服务器，所以需要去配置一下9876端口，不然无法访问

mvn clean package -Dmaven.test.skip=true

得到一个jar包：rocketmq-console-ng-1.0.0.jar

他会在执行命令的当前目录下创建一个 nohup.out 的输出日志

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

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

同步刷盘：当消息持久化到broker的磁盘后才算是消息写入成功

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

优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高

缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅（不可消费），消息实时性会受到影响

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

多Master多Slave集群(同步双写），其不仅可以保证数据不丢失，也不会影响消息写入。其运行效率要低 于多Master+RAID阵列

默认选择算法。该算法保证了每个Queue中可以均匀的获取到消息

该算法存在一个问题：由于某些原因，在某些Broker上的Queue可能投递延迟较严重。从而导致 Producer的缓存队列中出现较大的消息积压，影响消息的投递性能

该算法会统计每次消息投递的时间延迟，然后根据统计出的结果将消息投递到时间延迟最小的Queue。如果延迟相同，则采用轮询算法投递。该算法可以有效提升消息的投递性能

该算法也存在一个问题：消息在Queue上的分配不均匀。投递延迟小的Queue其可能会存在大量 的消息。而对该Queue的消费者压力会增大，降低消息的消费能力，可能会导致MQ中消息的堆 积

mapedFileSizeCommitLog=1073741824

mappedFile文件是顺序读写的文件，所有其访问效率很高

无论是SSD磁盘还是SATA磁盘，通常情况下，顺序存取效率都会高于随机存取

不难发现，消息单元中包含了Queue相关属性，这里要特别关注一下commitlog与queue间的关系是什么？

CommitLog 采用混合型存储，也就是所有 Topic 都存在一起，顺序追加写入，文件名用起始偏移量命名

消息先写入 CommitLog 再通过后台线程分发到 ConsumerQueue 和 IndexFile 中

消费者先读取 ConsumerQueue 得到真正消息的物理地址，然后访问 CommitLog 得到真正的消息

利用了 mmap 机制减少一次拷贝，利用文件预分配和文件预热提高性能

提供同步和异步刷盘，根据场景选择合适的机制

一个consumequeue文件中所有消息的Topic一定是相同的。但每条消息的Tag可能是不同的

注意看存储路径：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}， 它已经明确了topic和queueId的

consumequue 示意图

commitLog示意图

消费offset即消费进度，consumer对某个Queue的消费offset，即消费到了该Queue的第几 条消息; 消息offset = 消费offset + 1

在$HOME/store/config/consumerOffset.json 文件中记录了每个topic下的queue的消费进度

这个queueOffset 就是consumequeue文件中存储的索引单元,就是20字节的那个内容，找到这个单元后，取前8个字节可以获取消息在 commitlog中实际的偏移量，也就是可以定位消息了

queueOffset = 消息offset * 20字节

PageCache机制，页缓存机制，是OS对文件的缓存机制，用于加速对文件的读写操作。一般来 说，程序对文件进行顺序读写的速度几乎接近于内存读写速度，主要原因是由于OS使用 PageCache机制对读写访问操作进行性能优化，将一部分的内存用作PageCache

写操作：OS会先将数据写入到PageCache中，随后会以异步方式由pdæush（page dirty æush) 内核线程将Cache中的数据刷盘到物理磁盘

读操作：若用户要读取数据，其首先会从PageCache中读取，若没有命中，则OS在从物理磁 盘上加载该数据到PageCache的同时，也会顺序对其相邻数据块中的数据进行预读取

8byte

该indexFile中第一条消息的存储时间

8byte

该indexFile中最后一条消息存储时间

8byte

该indexFile中第一条消息在commitlog中的偏移量commitlog offset

8byte

该indexFile中最后一条消息在commitlog中的偏移量commitlog offset

4byte

已经填充有index的slot数量（并不是每个slot槽下都挂载有index索引单元，这里统计的是所有挂载了index索引单元的slot槽的数量）

4byte

该indexFile中包含的索引单元个数（统计出当前indexFile中所有slot槽下挂载的所有index索引单元的数量之和）

4byte

消息中指定的业务key的hash值

phyOffset

8byte

当前key对应的消息在commitlog中的偏移量commitlog offset

4byte

当前key对应消息的存储时间与当前indexFile创建时间的时间差

4byte

当前slot下当前index索引单元的前一个index索引单元的indexNo

默认就是 ConsumeType.CONSUME_PASSIVELY("PUSH") 

逻辑要用户自己实现

注意：集群消费模式下，不保证每一次失败重投的消息路由到同一台机器上。

消费进度保存在consumer端。因为广播模式下consumer group中每个consumer都会消费所有消息，但它们的消费进度是不同。所以consumer各自保存各自的消费进度

消费进度保存在broker中。consumer group中的所有consumer共同消费同一个Topic中的消息，同一条消息只会被消费一次。消费进度会参与到了消费的负载均衡中，故消费进度是需要共享的

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

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

举个例子: 在只有一个consumer1的情况下，它消费5个queue, 其中假设它从4号queue offset=300处开始消费，在返回给broker ACK之前，consumer2加入了，此时触发rebalance机制，此时假设将4号queue分配给了consumer2,那么由于consumer1没有返回offset给broker, 所以对于consumer2来讲，它也会从consuemr1读取消息的offset处开始消费，这样consuemr1和consumer2就消费了同样的消息，在集群模式下，这就是重复消费

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

Broker扩容或缩容

Broker升级运维

Broker与NameServer间的网络异常

Queue扩容或缩容

Consumer Group扩容或缩容

Consumer升级运维

Consumer与NameServer间网络异常

rocketmq中的rebalance是consumer实例自身完成的

TopicConfigManager

ConsumerManager

ConsumerOffsetManager

该算法不用事先计算每个Consumer需要分配几个Queue，直接一个一个分即可

该算法存在的问题：分配不均

两种平均分配策略的分配效率较高，一致性hash策略的较低。因为一致性hash算法较复杂。另外，一致性hash策略分配的结果也很大可能上存在不平均的情况

其可以有效减少由于消费者组扩容或缩容所带来的大量的Rebalance

Consumer数量变化较频繁的场景

什么是消费进度记录器？

同一个消费者组中的两个Consumer订阅了相同Topic的不同Tag

一个新的订阅组第一次启动从队列的最后位置开始消费, 后续再启动接着上次消费的进度开始消费

一个新的订阅组第一次启动从队列的最前位置开始消费, 后续再启动接着上次消费的进度开始消费

一个新的订阅组第一次启动从指定时间点开始消费，后续再启动接着上次消费的进度开始消费

对于已经存在的消费者组+topic+queue的订阅关系，无论如何都是遵循历史进度进行消费

对于新的消费者组+topic+queue关系，在正常情况下，遵循客户端配置的策略

对于特殊的场景被broker认为queue是新queue的情况下，一律从头开始消费（令可杀错不放过）

消费者在消费完一批消息后会向broker提交这些消息的offset，然后等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量

消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取nextBeginOffset

这里可能会对 Message Id 相同感到疑惑，因为在最前面有说过它的生产规则。注意，它不是消息生产者再次生产相同的消息再次发送，而是将之前发送失败的消息拿过来重新发送，所以和之前的消息是一模一样的

当然也有可能消费者主动重发，这样就可能会存在相同的消息有两个不同msgId的情况

请看前面的rebalance机制

幂等令牌：是生产者和消费者两者中的既定协议，通常指具备唯⼀业务标识的字符串。例如，订单号、流水号。一般由Producer随着消息一同发送来的

唯一性处理：RocketMQ无法避免消息重复（Exactly-Once），所以如果业务对消费重复非常敏感，消费端务必在业务层面进行去重处理。可以借助关系数据库进行去重

1. 首先通过缓存去重。在缓存中如果已经存在了某幂等令牌，则说明本次操作是重复性操作；若缓存没有命中，则进入下一步

2. 在唯一性处理之前，先在数据库中查询幂等令牌作为索引的数据是否存在。若存在，则说明本次操作为重复性操作；若不存在，则进入下一步

3. 在同一事务中完成三项操作：唯一性处理后，将幂等令牌写入到缓存，并将幂等令牌作为唯一索引的数据写入到DB中

1. 当支付请求到达后，首先在Redis缓存中却获取key为支付流水号的缓存value。若value不空，则说明本次支付是重复操作，业务系统直接返回调用侧重复支付标识；若value为空，则进入下一步操作

2. 到DBMS中根据支付流水号查询是否存在相应实例。若存在，则说明本次支付是重复操作，业务系统直接返回调用侧重复支付标识；若不存在，则说明本次操作是首次操作，进入下一步完成唯一性处理

3. 在分布式事务中完成三项操作

业务系统上下游能力不匹配造成的持续堆积，且无法自行恢复

业务系统对消息的消费实时性要求较高，即使是短暂的堆积造成的消费延迟也无法接受

SDK获取消息的方式为批量拉取，常见内网环境下都会有很高的吞吐量，例如：1个单线程单分区的低规格机器（4C8GB）可以达到几万TPS，如果是多个分区可以达到几十万TPS。所以这一阶段一般不会成为消息堆积的瓶颈

单节点线程数，即单个Consumer所包含的线程数量

节点数量就是consumer group 中的 consumer 个数

全局顺序消息

分区顺序消息

消息消费逻辑的计算复杂度是否过高，代码是否存在无限循环和递归等缺陷。

消息消费逻辑中的I/O操作（如：外部调用、读写存储等）是否是必须的，能否用本地缓存等方案规避。

消费逻辑中的复杂耗时的操作是否可以做异步化处理，如果可以是否会造成逻辑错乱（消费完成但异步操作未完成）。

说明：rocketmq控制台有统计消息耗时,可以登陆查看

逐步调大线程的单个节点的线程数，并观测节点的系统指标，得到单个节点最优的消费线程数和消息吞吐量

得到单个节点的最优线程数和消息吞吐量后，根据上下游链路的流量峰值计算出需要设置的节点数，节点数=流量峰值/单线程消息吞吐量。

但是实际上是无法连同的，需要做一下修改，请看下面

参考文档：https://blog.csdn.net/qq_21460229/article/details/104351178

将localhost 替换为公网ip

第一次没有就去创建，以后都使用这一个

为了不让消费者去消费，等到期了在消费

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

这个事务管理器TM 就是我们的消息生产者，后面在说

TC 就是Broker

执行结果就是告诉TM本地事务执行结果，是成功了还是失败了，有三种状态

若预扣款成功（本地事务状态为COMMIT_MESSAGE），则TM向TC发送Global Commit指令

若预扣款失败（本地事务状态为ROLLBACK_MESSAGE），则TM向TC发送Global Rollback指令

若现未知状态（本地事务状态为UNKNOW），则会触发工行系统的本地事务状态回查操作

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

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

回调操作返回UNKNWON

TC没有接收到TM的最终全局事务确认指令

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

RocketMQ中Broker充当着TC

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

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

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

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

1. TM向TC发起指令，开启一个全局事务。

2. 根据业务要求，各个RM会逐个向TC注册分支事务，然后TC会逐个向RM发出预执行指令。

3. 各个RM在接收到指令后会在进行本地事务预执行。

4. RM将预执行结果上报给TC。当然，这个结果可能是成功，也可能是失败。

5. TC在接收到各个RM的Report后会将汇总结果上报给TM，根据汇总结果TM会向TC发出确认指令

6. TC在接收到指令后再次向RM发送确认指令。

批量发送的消息必须具有相同的Topic

批量发送的消息必须具有相同的刷盘策略

批量发送的消息不能是延时消息与事务消息

方案一：将批量消息进行拆分，拆分为若干不大于4M的消息集合分多次批量发送

方案二：在Producer端与Broker端同时修改属性

默认值是1，表示从broker拉取消息后，一次性给多少条消息给Consumer去消费

默认值是32，表示从Broker拉取32条消息后给消费者消费

当然不是！

pullBatchSize值设置的越大，Consumer每拉取一次需要的时间就会越长，且在网络上传输出现问题的可能性就越高。若在拉取过程中若出现了问题，那么本批次所有消息都需要全部重新拉取

consumeMessageBatchMaxSize值设置的越大，Consumer的消息并发消费能力越低，且这批被消费的消息具有相同的消费结果。因为consumeMessageBatchMaxSize指定的一批消息只会使用一个线程进行处理，且在处理过程中只要有一个消息处理异常，则这批消息需要全部重新再次消费处理

数值：比如：123，3.1415

字符：必须用单引号包裹起来，比如：'abc'

布尔：TRUE 或 FALSE

NULL：特殊的常量，表示空

数值比较：>，>=，<，<=，BETWEEN，=

字符比较：=，<>，IN

逻辑运算 ：AND，OR，NOT

NULL判断：IS NULL 或者 IS NOT NULL

对于同步发送顺序消息，我觉得也是可以重试的

对于异步发送顺序消息，这个不可以重试，因为可能会破坏顺序关系

同步重试次数 = retryTimesWhenSendFailed + 1

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

失败隔离机制默认是关闭的，如果要使用这个机制，我们需要在发送消息时手动打开

这种情况下默认时不会重试的，如果想要重试，则修改发送消息时作如下修改

Producer#setRetryTimesWhenSendAsyncFailed(3);

默认值是16

若修改值小于16，则按照指定间隔进行重试

若修改值大于16，则超过16次的重试时间间隔均为2小时

官方推荐