ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 编程语言> 文章详细

RocketMQ-一文读懂架构、源码、使用与调优

2021-03-28 16:32:45  阅读：164  来源： 互联网

标签：文件 队列 Broker 重试 调优 Master 消息 源码 RocketMQ


Apache Rocketmq是一个分布式消息和流平台，具有低延迟、高性能和可靠性、万亿级容量和灵活的可扩展性，本文将从架构设计、源码解读、使用经验、调优调优，解剖MQ的特征。

1.1、MQ架构(Broker-分片)

MQ分片架构
支持多Master，可实现同步或者异步双写；Broker Master和Broker Slave是通过jdk原生的nio(SocketChannel)通信实现数据同步，Producer与Broker,Consumer与Broker通过Netty进行通信，NameServer与其他通过Netty实现通信；

1.1.1、名称服集群务 NameServer cluster(服务注册中心)

NameServer服务提供了轻量级的服务发现和路由，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。每个NameServer服务记录完整的路由信息，提供一致的读写服务，支持快速存储扩展;
NameServer是一个功能齐全的服务器，主要包括两个功能：

  • broker 管理，nameserver 接受来自broker集群的注册信息并提供心跳来检测他们是否可用。
  • 路由管理 每一个nameserver都持有关于broker集群和队列的全部路由信息，用来向客户端提供查询。
    我们知道 ，rocketMQ客户端（生产者/消费者）会从nameserver查询队列的路由信息，客户端是如何知道nameserver的地址的呢？
    有四种方式能够让客户端湖区到nameserver的地址：
  • 通过程序，像这样producer.setNamesrvAddr(“ip:port”)
  • java 配置项，这么用rocketmq.namesrv.addr
  • 环境变量 NAMESRV_ADDR
  • HTTP 端点
1.1.2、代理服务集群 Broker Cluster

提供消息存储,消息接收，消息查询，消息发送，Broker通过提供轻量级主题和队列机制来处理消息存储，Broker分为Master与Slave，一个Master可以对应多个Slave(当前RocketMQ版本在部署架构上支持一Master多Slave)，每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息(心跳)到所有NameServer。它们支持Push和Pull模型，包含容错机制(2个副本或3个副本)，提供了极强的峰值处理里能力和按照时间顺序存储数以百万记的消息存储能力，此外，代理提供了灾难恢复、丰富的度量统计和警报机制，这些都是在传统的消息传递系统中缺乏的;
Broker Server有一些非常重要的子模块：

  • remoting（远程） 模块，broker的入口，处理从客户端发起的请求;
  • client manager（客户端管理） 管理各个客户端（生产者/消费者）还有维护消费者主题订阅;
  • store（存储服务），提供简单的api来在磁盘保持或者查询消息;
  • HA高可用服务 提供主从broker的数据同步;
  • index(索引服务)为消息建立索引提供消息快速查询。
    索引
1.1.3、生产者集群 Producer Cluster(用户客户端）

produce支持分布式部署，分布式的produce通过broker集群提供的各种负载均衡策略将消息发送到broker集群中。发送过程支持快速失败是低延迟的。

1.1.4、消费者集群 Consumer Cluster(用户客户端）

消费者也支持在推送和拉取模式下分布式部署，它还支持集群消费和消息广播。提供实时的消息订阅机制，能够满足大多数消费者的需求。RocketMQ的网站为感兴趣的用户提供了一个简单的快速入门指南。

1.1.5、 MQ速度快的原因:

顺序写：消息存储,文件存储结构；
零拷贝：消息发送，减少内存复制(避免理论上2次fd交互,4次内存复制)；消息存储(文件存储机制)

1.2、消息存储架构

MQ消息存储架构
对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，类似数据库Btree数据(CommitLog)与索引(IndexFile)分离；

1.2.1、消息存储结构
  • CommitLog(存储消息元数据)

消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容,消息内容不是定长的，消息包括(起始物理偏移量offset，消息大小size和消息Tag的HashCode值),默认commitlog文件1GB。不同的topic，messageQueue都是写到相同的CommitLog 文件，也就是说CommitLog完全的顺序写。
CommitLog顺序写

  1. ConsumerQueue(通过偏移量查询消息)
    消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的；MessageQueue只是存储CommitLog中对应的位置偏移信息，方便通过MessageQueue找到对应存储在CommitLog的消息。一个Topic里面的ConsumerQueue对应队列数量的文件个数，默认加载到内存；
  2. IndexFile(通过key或时间区间查消息)
    IndexFile（IndexHeader, 槽位Slot和消息索引组成）提供了一种可以通过key或时间区间来查询消息的方法，默认存在500W个槽位，类似HasMap结构(消息的(MessageKey)取hashCode，通过hashCode获取slot位）, 消息索引键与消息偏移量映射关系(CommitLog与IndexFile建立关系)写入到IndexFile。
1.2.2、页缓存与内存映射

页缓存（PageCache)是OS对文件的缓存，用于加速对文件的读写。一般来说，程序对文件进行顺序读写的速度几乎接近于内存的读写速度，主要原因就是由于OS使用PageCache机制对读写访问操作进行了性能优化，将一部分的内存用作PageCache。对于数据的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取；
RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型将磁盘上的物理文件直接映射到用户态的内存，减少拷贝开销。

1.2.3、消息刷盘

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

1.3、消息发送

Linux操作系统分为【用户态】和【内核态】，文件操作，网络操作需要涉及上下文切换，免不了进行数据复制(通过fd描述符)，一台服务器把本机磁盘文件内容发送到客户端，一般分为两个步骤:
3. read:读取本地文件内容；
4. write:读取的内容通过网络发送出去；
这两个看似简单操作，实际进行了4次数据复制(两次fd交换【filefd,socketfd】)，fd交换次数决定内存拷贝次数，分别是：
5. 从磁盘复制数据到内核态内存；
6. 从内核态内存复制到用户态内存；
7. 然后从用户态内存复制到网络驱动的内核态内存；
8. 最后是从网络驱动内核态内存复制到网卡中进行传输；
状态变化
通过使用mmap的方式，可以省去想用户态内存复制，提高速度，这种机制在Java中通过MappedByteBuffer实现的，RocketMq利用上述特性，也就是所谓的“零拷贝”技术，提高消费存盘和网络发送的速度。
注意:采用MapperByteBuffer这种内存映射的几个限制，其中之一是单次只能映射1.5G-2G的文件至虚拟内存，这就是为何Rocketmq默认设置单个CommitLog日志数据为1GB的原因所在。
发送优化(请求堆积):
9. 通过异步发送回调方式；
10. 使用线程池进行消息发送优化；
消息模式:集群消息(支持重试)，广播消息(不支持重试);
消息类型:普通消息(默认使用，需消费端实现幂等)，顺序消息(消息需要在发送到一个Broker才能保证，严重降低效率)，事务消息(2段提交)；

1.3.1、消费模式(Push)
  1. PUSH: rocketMq默认是push模式，消费者订阅主题，然后自动进行集群内消息队列的动态负载，自动拉取消息，准实时，项目中一般采用PUSH模式。
  2. PULL:只是为Push模式服务，消费者无需订阅主题，由业务方（应用程序）直接根据MessageQueue拉取消息。

1.4、高可用性机制

MQ高可用机制
Rocketmq分布式集群通过Master和Slave配合达到高可用的。
Master和Slave的区别: Master支读和写，Slave仅支持读；在Broker配置文件中，参数brokerId的值为0表明这个是Broker的Master，大于0表明这个Broker是Salve，同时brokerRole参数也会说明这个Broker是Master还是Slave;

1.4.1、消息消费高可用

在Consumer的配置文件中，并不需要设置是从Master读还是从Slave读，当Master不可用或者繁忙的时候，Consumer会被自动切换到从Slave读。有了自动切换Consumer的这种机制，当Master角色的机器出现故障后，Consumer仍然可以从Slave读取数据，不影响Consumer程序，这就达到了读消息高可用性；

1.4.2、消息发送高可用

在创建Topic的时候，把Topic的多个MessageQueue创建在多个Broker组上(相同的Broker名称，不同的BrokerId的机器组成一个Broker组，单个Broker组包括Master和对应的Slave)，这样当一个Broker组的Master不可用后，其他组的Master仍然可用，生产者仍然可以发送消息。
RocketMq目前不支持把Slave自动切换到Master，如果机器资源不足，需要把Slave转成Master，则要手动停止Slave角色的Broker，更改配置文件启动Broker。

1.4.3、消息主从复制

如果一个Broker组有Master和Slave，消息需要从Master复制到Slave上，有同步和异步复制方式；

  • 同步复制
    是等Master和Slave均写成功后才反馈给客户端写成功状态。在同步复制下如果Master出现故障，Slave上有备份数据容易恢复，但是同步复制会增大数据写入延迟，降低系统吞吐量。
  • 异步复制
    异步复制方式只要Master写成功，即可反馈客户端写成功标识状态。在异步复制方式下，系统拥有较低的延迟和较高的吞吐量，但如果Master出现故障，有些数据因为没有被写入Slave，有可能存在数据丢失。
  • 配置
    同步复制和异步复制是通过Broker配置文件里面的brokerRole参数进行设置的，这个参数设置成ASYNC_MASTER,SYNC_MASTER,SLAVE三个值中一个；
  • 总结
    刷盘
    在实际应用中结合业务场景，合理设置刷盘方式和主从同步方式，尤其是SYNC_FLUSH方式，由于频繁触发刷盘动作，会明显降低性能。通常情况下，应该把Master和Slave配置成ASYNC_FLUSH的刷盘方式，主从之间配置成SYNC_MASTER的复制方式，这样即使有一台机器出现故障，仍然能保证数据不丢失，是个不错选择。

1.5、负载均衡

1.5.1、Producer负载均衡

Producer端，每个实例在发消息的时候，默认会轮询所有的MessageQueue发送，以达到让消息平均落在不同的queue上。而由于queue可以散落在不同的broker，所以消息就发送到不同的broker下，如下图。
MQ集群负载均衡算法
途中箭头线条上的标号代表顺序，发步方会把一条消息发送到Queue0,然后第二条消息发送到Queue2，以此类推。

1.5.2、Consumer负载均衡

(1) 集群模式
在集群消费模式下,每条消息只需要投递到订阅这个topic的ConsumerGroup下的一个实例即可。RocketMQ采用主动拉取的方式拉取并消费消息，在拉取的时候明确指定拉取那一条messageQueue。
当每个实例的数量有变更，都会触发一次所有实例的复制均衡，这时候会按照queue的数量和实例的数量平均分配queue给每个实例。
默认分配算法是AllocateMessageQueueAveragely,如下图所示:
在这里插入图片描述
还有一种平均算法是AllocateMessageAveragelyByCycle，也是平均分摊每一条queue，只是以环状轮流分queue的形式，如下图:
在这里插入图片描述
一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。某个topic消息量很大，应该给它多配置几个队列，并且尽量多分布在不同broker上，减轻某个broker的压力。
（1）Rebalance(队列分配)：topic下的所有MessageQueue按照一定规则分配给consumerGroup下的consumer，有如下策略：

  • (默认)平均Hash队列算法(Average Hashing queue algorithm)，消费多个topic时，存在排序靠前的consumer可能消费队列较多(机器负载较大);
    想到解决方案:不使用Collections.sort()而使用Collections.shulffe();
  • 平均周期哈希队列算法(Cycle average Hashing queue algorithm);
  • 机房哈希队列算法(Computer room Hashing queue algorithm);
  • 自定义配置算法(QueueByConfig);
    条件:默认20s，consumer出现宕机或者有新的consumer上线；

源码分析: rebalanceByTopic(final String topic){} •
先获取topic下的MessageQueue，一个MessageQueue实际上就是一个partition •
然后获取当前topic和group的client id，即当前group中消费此topic的客户端 •
随后对partition和client id做排序 • 然后调用strategy获取当前客户端需要消费的partition •
最后更新订阅

(2) 广播模式
在实现上，其中一个不同就是在consumer分配queue的时候，会所有consumer都分到所有的queue。
在这里插入图片描述

1.6、消息重试

1.6.1、顺序消息重试

对于顺序消息，当消费者消费消息失败后，消息队列RocketMQ会自动不断进行消息重试(每次间隔时间1秒)，这时应用会出现消息被阻塞的情况。因此，在使用顺序消息时，务必保证应用能够及时监控并处理消息失败的情况，避免阻塞发生。

1.6.2、无序消息

对于无序消息(普通，定时，延时，事务消息)，当消费者消费消息失败时，您可以通过设置返回状态达到消息重试的结果。无序消息的重试只针对集群消息方式生效，广播方式不提供失败重试特性，即消息不再重试。
(1) 重试次数
消息队列RocketMQ默认每条消息最多重试16次，每次重试的间隔时间如下：
第几次重试 与上次重试间隔 第几次重试 与上次重试间隔

如果消息重试16次后仍然失败，消息将不再进行投递。如果严格按照上述时间计算，某一条消息在一直消费失败前提下，将会在接下来的4小时46分钟内进行16次重试，超过此时间范围内将不再重试投递，会放入“死信队列”。
(2) 配置方式
消息失败后，重试配置方式。在集群消费方式下，消息消费失败后期望消息重试，需要在消息监听接口的实现中明确进行配置(三种方式任选一种):

  • 返回Action.ReConsumerLater(推荐)
  • 返回Null;
  • 抛出异常；
    消息消费失败后，不重试配置方式，返回Action.CommitMessage，此后这条消息将不会再重试。
    自定义消息最大重试次数:消息队列RocketMq允许Consumer启动时候设置最大重试次数，重试时间间隔按照如下策略：
  • 最大重试次数小于等于16次，则重试时间间隔同上表所示；
  • 最大重试次数大于16次，超过16次的重试时间间隔均为2h;
Properties properties = new Properties();
properties.put(PropertyKeyConst.MaxReConsumTimes, "20");
Consumer consumer = ONSFactory.createConsumer(properties);

注意：

  • 消息最大重试次数设置对相同Group ID下的所有Consumer实例有效；
  • 如果相同的Group ID下两个Consumer实例中的其中一个设置了MasReconsumeTimes，那么该配置对两个实例生效；
  • 配置方式采用覆盖方式生效，即后启动的Consumer覆盖之前启动的实例配置；
    获取消息重试次数，消息者收到消息后，可按照如下方式获取消息重试次数；
public class MessageListenerImpl implements MessageListener {
    @Override
    public Action consume(Message message, ConsumerContext context){
        System.out.println(message.getReConsumeTimes());
    }
}

1.7、死信队列

当消息重试次数达到最大重试次数后，会被放入死信队列(Dead-Letter Message);

1.7.1、死信消息特征：
  • 不会被消费者正常消费；
  • 有效期与正常消息相同，均为3天，3天后自动删除。因此，请在死信消息产生后的3天内可人工及时处理。
1.7.2、死信队列特征
  • 一个死信队列对应一个Group ID，而不是对应单个的消费者实例；
  • 如果一个Group ID未产生死信消息，消息队列RocketMq不会为其创建相应的死信队列；
  • 一个死信队列包含了对应Group ID产生的所有死信消息，不论消息属于那个Topic，一个死信队列可能包含多个Topic消息;

1.8、消费幂等性

消费者在接收到消息后,有必要根据业务上的唯一key对消息做幂等处理必要性。

1.8.1、消息幂等问题

在互联网应用中，尤其在网络不稳定情况下，消息队列RocketMQ的消息有可能会出现重复，这个重复简单可以概括为一下情况：

  • 发送消息重复(Producer->ConsumeQueue)
    当一条消息已被成功发送到服务端并完成持久化，此时出现“网络闪断或客户端宕机”导致服务端对客户端应答失败。如果此时生产者意识到消息发送失败尝试再次发送消息，消息后续会收到两条内容相同并且Message ID也相同的消息。
  • 投递时消息重复(ConsumeQueue-Consumer)
    消息消费的场景下，消息已投递到消费者并完成业务处理。当客户端给服务端反恐应答的时候网络闪断，为了保证消息至少被消费一次，消息队列RocketMQ的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且Message ID也相同的消息。
  • 负载均衡时消息重复
    包括但不限于网络抖动，Broker重启以及订阅方重启。当消息队列RocketMQ的Broker或客户端重启、扩容或者缩容时，会触发Rebalance，此时消费者可能会收到重复消息。
1.8.2、处理方式

因为Message ID有可能出现重复(重复)的情况，所以真正安全的幂等处理，不建议以Message ID作为处理依据。最好的方式是以业务唯一标识为幂等性关键依据，而业务的唯一标识可以通过key进行设置：

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

订阅方收到消息时可以根据消息的key进行幂等性处理：

consumer.subscribe("one_test", "", new MessageListener(){
    public Action consume(Message message, ConsumeContext context){
        String key = message.getKeys();
    }
})  //此处可以增加key写redis缓存标识是否已经处理过redis-key+数据库唯一索引

1.9、remoting模块

使用Netty的Reactor主从多线程模型，模块使用RemotingCommand(消息头+消息体)进行通信，消息可靠性。
RocketMQ提供同步和异步两种复制方式来增强消息的可靠性与高可用性，同步异步区别：同步复制会等待slave响应提交日志已经被复制，相应的异步复制会在master节点处理成功后快速返回，消息最大允许大小4MB;

  • 消息支持持久化到Commitlog里面，即使宕机后重启，未消费的消息也是可以加载查询出来的;
  • Broker自身支持同步刷盘、异步刷盘的策略，可以保证接收到的消息一定存储在本地的内存中；
  • Broker集群支持 1主N从的策略，支持同步复制和异步复制的方式，同步复制可以保证即使Master 磁盘崩溃，消息仍然不会丢失；
1.9.1、多Master多Slave模式-异步复制

每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式(Master复制完成即返回)，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：

  • 优点：速度快，即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；
  • 缺点：消息存在丢失，Master宕机，磁盘损坏情况下会丢失少量消息。
1.9.2、多Master多Slave模式-同步双写【生产】

每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：

  • 优点：数据安全性高，数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；
  • 缺点：性能较低，性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。
1.9.3、消息传递保证三个级别
  • At most once:消息可能会丢失，但绝对不会重复传递；
  • At least once:消息绝不会丢失，但消息可能会重复传递；
  • Exactly once:每条消息只会被传递一次；

MQ通过两段事务提交，由客户端发送应答，提交完成最近一次offset，从而避免客户端未消费完成发生宕机重启时候不能继续消费。

1.10、延迟消息(kafka无)

延迟消息

  • Producer要将一个延迟消息发送到某个Topic中；
  • Broker判断这是一个延迟消息后，将其通过临时存储进行暂存；
  • Broker内部通过一个延迟服务(delay service)检查消息是否到期，将到期的消息投递到目标Topic中。这个的延迟服务名字为delay service，不同消息中间件的延迟服务模块名称可能不同；
  • 消费者消费目标topic中的延迟投递的消息。

1.11、源码分析

1.11.1、消息存储
  1. 消息存储核心类
    DefaultMessageStore;
    核心方法: putMessage(MessageExtBrokerInner msg);
  2. 消息存储流程
    消息存储
  • 从mapperFileQueue获取MapperFile;
  • MappedFile内存文件写入消息；
  • 刷盘；
  • HA同步；
  1. 存储文件
    在这里插入图片描述
  • commitlog:消息存储目录；
  • config:运行期间一些配置信息；
  • consumequeue:消息消费队列存储目录(提供按照消息offest检索)；
  • index:消息索引文件目录(提供按照key和时间区间检索)；
  • abort:如果存在该文件寿命的Broker非正常关闭(用于ConsumeQueue和IndexFile文件恢复，保持与CommitLog一致);
  • checkpoint:文件检查点，存储CommitLog文件最后一次刷盘时间戳，consumerqueu最后一次刷盘时间，index索引文件最后一次刷盘时间戳(异步刷盘丢失恢复)。
  1. 存储文件内存文件映射
    RocketMQ通过使用内存映射文件提高IO访问性能，无论是CommitLog(1G)、ConsumerQueue还是IndexFile(400MB)，单个文件都被设计为固定长度，如果一个文件写满以后再创建一个新文件，文件名称就为该文件第一条消息对应全局物理偏移量。
    (1) MappedFileQueue
    对应就是物理的CommitLog所有文件；
    按照时间查找偏移量；
    根据消息偏移量查询MappedFile;
    (2) MappedFile(1G-内存映射技术)
    其实对应的就是commitlog里面的每一个文件；

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

  • 开启堆外内存: writebuffer->FileChannel;
  • 没有开启堆外内存：mappedByteBuffer(直接写到内存映射)；
    (3) TransientStorePool
    表示堆外内存，然后通过Commit线程将数据提交到内存映射的Buffer中，在通过Flush线程刷盘；
  1. 实时更新消息消费队列与索引文件
    消息消费队列文件、消息属性索引文件都是基于CommitLog文件构建的，当消息生产者提交的的消息存储在CommitLog文件中，ConsumeQueue、IndexFile需要实时更新，否则消息无法及时被消费，根据消息属性查找消息也会出现较大延迟。RocketMQ通过开启一个线程ReputMessageService来准实时转发CommitLog文件更新时间，相应的任务处理器根据转发消息及时更新CosumeQueue、IndexFile文件。
    在这里插入图片描述在这里插入图片描述
  • 代码:DefaultMessageStore:start
  • 代码DefaultMessageStore:run
    (1) 转发到ConsumeQueue
    在这里插入图片描述
    (2) 转发到IndexFile
    在这里插入图片描述
  1. 消息队列和索引文件恢复
    由于RocketMQ存储首先将消息全量存储在CommitLog文件中，然后异步(单独线程处理)生成转发任务更新ConsumeQueue和Index文件。如果消息成功存储到CommitLog文件中，转发任务未成功执行，此时消息服务器Broker由于某个原因宕机，导致CommitLog、ConsumeQueue、IndexFile文件数据不一致。如果不加以人工修复的话，会有一部分消息即使在CommitLog文件中存在，单由于没有转发到ConsumeQueue，这部分消息将永远无法被消费者消费，通过abort文件获取异常退出点。
    在这里插入图片描述

(1) 存储文件加载
代码DefaultMessageStore#load

(2) 正常恢复
从倒数第3个文件开始进行恢复，如果不足3个文件，则从第一个文件开始恢复
代码CommitLog#recoverNormally

(3) 异常恢复
从最后一个文件往前走，找到第一个消息存储正常文件。
代码CommitLog#recoveryAbnormally

  1. 刷盘机制
    RocketMQ的存储是基于JDK NIO的内存映射机制(MappedByteBuffer)的，消息存储首先将消息追加到内存，在根据配置的刷盘策略在不同时间进行刷写磁盘。
    (1) 同步刷盘
    消息追加到内存后，立即将数据刷写到磁盘文件。
    同步刷盘
    代码CommitLog#handleDiskFlush；
    (2) 异步刷盘
    开启transientStorePoolEnable后异步刷盘步骤：
  • 将消息直接追加到ByteBuffer(堆外内存);
  • CommitRealTimeService线程每隔200ms将ByteBuffer新追加内容提交到MappedByteBuffer中(提交堆外ByteBuffer->内存映射文件MappedByteBuffer）)；
  • MappedByteBuffer在内存中追加提交内容，wrotePosition指针向后移动；
  • Commit操作成功返回，将committedPosition位置恢复；
  • FlushRealTimeService线程默认每500ms将MappedByteBuffer追加的内存刷写到磁盘(提交内存映射文件MappedByteBuffer->磁盘)；
代码CommitLog#CommitRealTimeService#run；
代码CommitLog#FlushRealTimeService#run；
  1. 过期文件删除机制
    由于RocketMQ操作CommitLog、ConsumerQueue文件是基于内存映射机制并在启动时候回加载CommitLog、ConsumerQueue目录下所有文件，为了避免内存与磁盘的浪费，不可能将消息永久存储在服务器上，所以要引入一种机制来删除过期的文件。RocketMQ顺序写CommitLog、ConsumerQueue文件，所有写操作全部落在最后一个CommitLog或者ConsumerQueue文件上，之前的文件在下一个文件创建后将不会再被更新。RocketMQ清除过期文件方法时:如果当前文件在一定时间间隔内没有再次消费，则认定为是过期文件，可以被删除，RocketMQ不会关注这个文件上消息是否全部被消费。默认每个文件过期时间为72h(3天)，通过Broker配置文件中设置fileReservedTime来改变过期时间，单位为小时。
    代码:DefaultMessageStore#addScheduleTask
    代码:DefaultMessageStore#cleanFilesPeriodically;
    代码:DefaultMessageStore#deleteExpiredFiles;
    (1) 删除文件操作的条件
  • 指定删除文件时间点，RocketMQ通过deleteWhen设置一天固定时间执行一次删除文件操作，默认4点；
  • 磁盘空间如果不足，删除过期文件；
  • 预留，手工触发；
    代码:CleanCommitLogService#isSpaceToDelete;
    代码:MappedFileQueue#deleteExpiredFileByTime;
    9. 小结
  • 存储文件：CommitLog(消息文件)、ConsumerQueue(消息消费队列文件)、IndexFile(Hash索引文件)、checkPoint(监测点文件)、abort(关闭异常文件)；
  • 内存映射机制：消息存储文件、消息消费队列文件、Hash索引文件按照固定长度读写操作；
  • 文件命名：组织文件以文件起始偏移量来命名文件，能快速定位真实物理文件；
  • 刷盘机制：同步刷盘和异步刷盘(堆外数据->内存映射文件->刷盘),异步刷盘由Commit线程(每200ms)和Flush线程(每500ms)完成；；
  • 消息存储：为了保证消息存储完全顺序写，单一文件存储所有主题消息；但存在读取不方便，因此引入队列文件和索引文件(可设置消息索引键)检索；
  • 线程转发消息：消息到达CommitLog后，通过ReputMessageService线程近实时将消息转发给消息队列文件(ConsumeQueue)和索引文件(IndexFile)；
  • 各文件一致性保证：引入abort文件，记录Broker停机还是正常关闭还是异常关闭，重启后加载保证CommitLog、ConsumerQueue与Hash索引文件一致性；
  • 消息文件删除：启用过期(72h)文件删除，并在空间不足或凌晨4点删除过期文件，删除时不判断消息文件上消息是否被消费；
    部署推荐：同步复制，异步刷盘；

1.12、消息问题处理

在这里插入图片描述
消息发送broker默认采用(Async_Flush,Async_Master)，Async复制(Master复制)并不能保证消息不丢失，所以SEND_OK并不意味它是可靠的，要确保消息不丢失，需要开启(Sync_Master同步主或Sync_Flush同步写，一般适用于金融领域)，在不影响吞吐量情况下，推荐使用：

messageProducer.send(new Message(), new SendCallback(){}));
1.12.1、消费者消费慢问题
  • 提高消费并行度(增加机器+增加消费线程)
    绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：
    同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数(topic)的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。
    提高单个Consumer 的消费并行线程，通过修改参数consumeThreadMin、consumeThreadMax实现。
  • 批量方式消费
    某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，这样即可大幅度提高消费的吞吐量，通过设置 consumer的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。
  • 跳过非重要消息
    发生消息堆积时，如果消费速度一直追不上发送速度，如果业务对数据要求不高的话，可以选择丢弃不重要的消息。例如，当某个队列的消息数堆积到100000条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度。示例代码如下：
public ConsumeConcurrentlyStatus consumeMessage(
        List<MessageExt> msgs,
        ConsumeConcurrentlyContext context) {
    long offset = msgs.get(0).getQueueOffset();
    String maxOffset =
            msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET);
    long diff = Long.parseLong(maxOffset) - offset;
    if (diff > 100000) {
        // TODO 消息堆积情况的特殊处理
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    // TODO 正常消费过程
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
  • 优化每条消息消费过程(优化业务代码)
    举例如下，某条消息的消费过程如下：
    根据消息从 DB 查询【数据 1】
    根据消息从 DB 查询【数据 2】
    复杂的业务计算
    向 DB 插入【数据 3】
    向 DB 插入【数据 4】
    这条消息的消费过程中有4次与 DB的 交互，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，所以如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，即总体性能提高了 40%。所以应用如果对时延敏感的话，可以把DB部署在SSD硬盘，相比于SCSI磁盘，前者的RT会小很多。
  • 刷盘部署方式
    异步刷盘建议使用自旋锁，同步刷盘建议使用重入锁，调整Broker配置项useReentrantLockWhenPutMessage，默认为false；异步刷盘建议开启TransientStorePoolEnable；建议关闭transferMsgByHeap，提高拉消息效率；同步刷盘建议适当增大sendMessageThreadPoolNums，具体配置需要经过压测。
  • 消息复制方式：
    如果业务系统可以容忍消息丢失，MQ部署可以采用异步复制方式；
1.12.2、消息重投(保证消息消费)

生产者在发送消息时，同步消息失败会重投，异步消息有重试，oneway没有任何保证。消息重投(消息量大/网络抖动)会导致消息重复；
同步重发：默认重投2次，不会选择上次失败的broker，尝试向其他broker发送，最大程度保证消息不丢；
异步重试：异步重试不会选择其他broker，仅在同一个broker上做重试，不保证消息不丢。
1.11.12.3. 死信队列(消息消费失败)
实行队列
当一条消息初次消费失败，消息队列会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

1.12.4、消息积压问题

原因分析：消费者挂掉或生产者大量消息或延时订单消息，因消息消费队列数固定导致MQ消息积压；

  • 前期:做好监控和流量控制；
  • 提高消费者速度：如果比较严重，新建一个临时topic,partition增大倍，queue增大倍数，写中间程序快速转发，并增加Consumer数量进行消费(线程数+多条消费等)；
    消息挤压

标签：文件,队列,Broker,重试,调优,Master,消息,源码,RocketMQ
来源： https://blog.csdn.net/software444/article/details/115278248

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有