官网

为什么选择RocketMQ | RocketMQ

GitHub - apache/rocketmq: Apache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.

RocketMQ安装

下载

可以到RocketMQ官网选择合适的版本下载：https://rocketmq.apache.org/download
也可以到RocketMQ github下载：https://github.com/apache/rocketmq

启动之前先设置好环境变量
ROCKETMQ_HOME：<rocketmq安装目录>

启动

NameServer启动

1	<rocketmq安装目录>\bin\mqnamesrv.cmd

Broker启动

1
2
3

<rocketmq安装目录>\bin\mqbroker.cmd -c <配置文件地址> -n <NameServer地址>
//注：除了使用-n参数外，也可以使用NAMESRV_ADDR环境变量配置NameServer地址
//注：除了使用

Proxy启动

1 2	<rocketmq安装目录>\bin\mqproxy.cmd -pc <配置文件地址> -n <NameServer地址> //注：除了使用-n参数外，也可以使用NAMESRV_ADDR环境变量配置NameServer地址

配置

官方文档中并没有提供全面配置参考，所以只能通过源码查看全面的配置项，源码中每个组件都有一个单独的配置类，变量名就对应配置文件中的配置项

	源码全限定类名	配置文件路径	配置文件名
NameServer	org.apache.rocketmq.common.namesrv.NamesrvConfig	userhome/namesrv/	namesrv.properties
Broker	org.apache.rocketmq.common.BrokerConfig org.apache.rocketmq.store.config.MessageStoreConfig	rabbitmq目录/conf	broker.conf
Proxy	org.apache.rocketmq.proxy.config.ProxyConfig	rabbitmq目录/conf	rmq-proxy.json

可视化

RocketMQ可视化源码下载地址：https://github.com/apache/rocketmq-dashboard
下载后修改application.properties文件rocketmq.config.namesrvAddr配置为上面NameServer的地址，然后打包运行即可

RocketMQ结构

NameServer：负责存储集群元数据，维护了 Broker 地址等信息，通常以集群的方式部署，但是它是无状态的，各实例间相互不进行信息通讯，每个实例维护同样的数据
Broker：主要负责消息的存储、投递和查询以及服务高可用保证。
Proxy：客户端和服务端之间的代理层，提供了额外的安全性和功能增强，Proxy使用GRPC解决了多语言使用RocketMQ的问题，如果在Java环境下使用Remoting协议，该组件不是必须的
生产者：向Broker或者Proxy发送消息
生产者组：多个生产者构成的生产者组，主要作用是当其中一个生产者发送完事务消息后宕机了，Broker可以通过检查同生产者组中的其他生产者以确认事务状态
消费者：从Broker或者Proxy消费消息
消费者组：多个消费者组成的消费者组，是Topic向消费者发送消息的一个逻辑分组，Topic会将消息发送给所有订阅的消费者组，然后消费者组内部根据消息模式，如果是广播模式组内则所有消费者都消费，如果是集群模式则根据负载均衡选择一个消费者消费，负载均衡分为两种
1. 消息粒度的负载均衡，5.0开始支持，简单来说就是单个队列可以将消息按照负载策略投递到消费者组内多个消费者，如果topic有多个队列，每个队列都是这种模式
2. 队列粒度的负载均衡，简单来说就是单个队列只支持将消息发送给固定的消费者，这里的负载均衡指的是，同一个topic下的队列对于一个消费者组都将分配一个消费者，如果消费者少于队列，则多个队列分配同一个消费者，如果消费者多于队列，则每个队列分配一个消费者后，剩余消费者将空闲，如果消费者出现问题，比如下线了，那么将重新分配
Topic：消息主题，消息传输和存储的顶层容器（逻辑容器），用于标识同一类业务逻辑的消息，可以理解为消息分组，所有消息都会属于一个Topic，生产者以Topic为目标投递消息，消费者也以Topic为目标消费消息，5.0开始，特定类型的消息只能投递到特定类型的Topic
Queue：消息队列，消息存储和传输的实际容器，也是消息的最小存储单元，Topic由多个队列组成，消息储存在队列，而队列属于Topic，物理意义上，生产者的消息直接投递到队列，而消费者直接从队列获取消息

消息顺序性

从RocketMQ的结构来看，消息的传递简单来说就是从生产者推送到队列，队列再推送到消费者，只是这两步都受到很多逻辑的控制，而且消息的顺序性又涉及到全局顺序性和局部顺序性，这里默认生产者同步生成和消费者同步消费

全局顺序性：全局顺序性的要求下，那么RocketMQ只能有一个生产者，一个队列，一个消费者
局部顺序性：局部顺序性的要求下，即将按类别保证消息顺序性，那么要保证同一类别消息从同一生产者生产，并且投递到同一队列，再由同一消费者消费
1. 保证同一类别消息由同一生产者生产，这一步很好保证
2. 保证同一类别消息进入同一队列，RocketMQ支持在发送消息时就确定投递到哪个队列，不过并不是直接设置某个消息，而是为生产者设置一个队列选择器，我们自定义实现设置队列的逻辑，5.0支持顺序消息，可以设置消息分组，同一分组的消息，按顺序投递，不过底层就是通过对消息分组的名称进行hash取模以确定队列

public MessageQueueImpl takeMessageQueueByMessageGroup(String messageGroup) {
    long hashCode = Hashing.sipHash24().hashBytes(messageGroup.getBytes(StandardCharsets.UTF_8)).asLong();
    int index = LongMath.mod(hashCode, this.messageQueues.size());
    return (MessageQueueImpl)this.messageQueues.get(index);
}

3. 要保证由同一消费者消费，要考虑消费者组内是否使用到了消息粒度的负载均衡，因为消息粒度的负载均衡下，队列可能随机将消息投递到某一个消费者，队列粒度的负载均衡不存在这个问题，或者直接控制消费者组内部只有一个消费者

消息持久性

RocketMQ具体持久化的路径默认为userhome/store，windows中为UserProfile环境变量，想自定义的话可通过storePathRootDir配置项配置

其中比较重要的三个持久文件是CommitLog，ConsumeQueue，IndexFile

CommitLog：消息的实际存储文件，设计原则是保证高吞吐量，它是一个顺序写磁盘的过程，非常适合于利用现代存储设备的顺序写优势，使用到了内存映射技术（mmap），默认固定大小为1个G，原因是因为内存映射技术对于大小有限制，CommitLog文件是全局的

ConsumeQueue：提供了一种快速访问消息的手段，它记录了消息在CommitLog中的offset和size以及消息tag的hashcode，每条消息的信息占用的大小相同，所以只需要根据消费位点以及信息占用大小，就可以非常快速的定位到消息信息在ConsumeQueue的位置，进而快速从CommitLog中找到消息，它同样是顺序写并且使用到了内存映射技术，每个队列都有一个ConsumeQueue文件

IndexFile：结构有点类似HashMap，使用消息的Key作为索引，提供了一种通过Key快速访问消息的手段，如果消息没有Key，那么不会向IndexFile写入，IndexFile也是全局的，参考文档

除了以上三个，还有一些其他的持久化文件，具体可进入目录中查看，这里额外介绍一下consumerOffset.json文件，默认在config目录下，这个文件存储的就是消费位点，记录了Topic对应每个订阅的消费者组在每个队列的消费位点

消息模式

集群消息：这方方式可以实现类似ActiveMQ负载均衡客户端的功能，同一个ConsumerGroup下的所有Consumer已负载均衡的方式消费消息。比较特殊的是，这种方式可以支持生产端先发送消息到Broker，消费端再订阅主题进行消费，比较灵活。RocketMQ默认为该模式
广播消息：在这种模式下，生产端发送到Topic下的消息，会被订阅了该Topic的所有Consumer消费，即使它们处于同一个ConsumerGroup

消费模式

官方文档：https://rocketmq.apache.org/zh/docs/featureBehavior/06consumertype

Apache RocketMQ 提供了两种消费消息的方式：Pull（拉取）消费和Push（推送）消费。尽管术语上存在区别，但实际上，在RocketMQ中所谓的Push消费也是一种基于Pull机制的实现。

Push（推送）消费

虽然被称为“Push”，但实际上RocketMQ的Push消费模型更像是一种智能化的Pull模型。通过一个长轮询线程，异步拉取消息，再提交给消费线程

特点：

自动拉取：当消息到达时，Broker会通知Consumer，由Consumer主动去拉取。
异步处理：消息到达后，Consumer可以在后台异步处理消息。
简化编程模型：开发者只需要关心消息处理逻辑，不需要频繁地编写拉取消息的代码。

Pull（拉取）消费

Pull消费模式下，消费者主动向Broker请求消息，决定何时拉取以及拉取多少消息。这种方式给予了消费者更多的控制权。

特点：

主动权在消费者：消费者可以根据自己的处理能力主动向Broker拉取消息。
灵活控制：消费者可以根据实际负载情况灵活地控制拉取消息的速度和数量。
流量控制：能够更好地控制消息的处理速率，避免因为瞬时大量消息涌入而导致系统过载。

实现机制

尽管Push模式表面上看起来是由Broker主动推送消息给Consumer，但实际上，RocketMQ依然依赖于Pull机制来实现消息消费。

消息过滤

官方文档：https://rocketmq.apache.org/zh/docs/featureBehavior/07messagefilter

基于标签（Tag）的过滤：
- 在 RocketMQ 中，生产者可以为每条消息设置一个或多个标签（Tag），而消费者则可以通过指定特定的标签来过滤消息。这是最基本的过滤方式，通常用于根据消息的主题或类别来筛选消息。
  由于ConsumerQueue中就包含了Tag的hashcode，所以通过Tag过滤，在ConsumerQueue可以实现第一次过滤，但是由于hashcode可能会发生hash碰撞，所以拿到消息后，还需要再过滤一次
SQL 表达式过滤：
- SQL92 模式的过滤允许消费者使用类似于 SQL 的语法来定义更为复杂的过滤条件。这种方式提供了比基于标签过滤更强的表达能力，允许在生产者自定义属性，在消费者端编写过滤条件。

消息丢失

生产者可通过设置同步发送保证消息成功到达Broker

同步发送：使用同步发送模式，即<font style="color:rgb(44, 44, 54);">sendSync</font>方法。这种模式下，发送消息后，会阻塞等待broker的回复，异步发送可以异步接收broker的回复，但是同步发送安全性更高，单向发送是最不安全的，无法确定消息是否成功推送到

Broker可通过同步刷盘以及搭建集群保证

同步刷盘：即消息到达broker后，等待刷盘完成后才会生产者，如果在没有刷盘之前broker就下线了，那么对生产者而言，这条消息是没有推送成功的，那么在broker上线之后，它会尝试继续推送，如果刷盘之后broker宕机，但是消息已经持久化了，重新上线后，仍然可以被消费
搭建集群：如果觉得单机节点还不够安全，还可以采取多副本的模式，rocketmq支持一主多从，多主多从，在消息复制方面，rocketmq支持同步复制（也叫同步双写）以及异步复制，同步复制会保证消息复制给所有从节点后才返回给生产者

消费者可通过手动应答应答保证消息被消费者消费

手动应答：消费者消费完消息后，手动向Broker确认消息已被成功消费。如果消费过程中出现问题，消费者不应确认消息消费，以便消息可以被重新消费。这种模式下只要broker没有收到消费者的应答，就会继续尝试推送，当然，超过一定次数，就会被认为是死信消息

集群架构

RocketMQ集群

消息积压

处理 RocketMQ 中的消息积压可以从很多方面考虑解决问题，比如在代码层面优化消费者代码逻辑，提升消费速度，或者直接提升硬件资源，提升消费速度，从RocketMQ来考虑，可以考虑以下几点

增加消费者数量，这里需要注意RocketMQ消费者组负载均衡的方式：
1. 如果是消息粒度的负载均衡，基于一个队列可以负载多个消费者的特性，可以通过增加更多的消费者来提高消息的处理速度，从而减少队列中的积压。确保你的消费者程序能够并行处理消息。
2. 如果是队列粒度的负载均衡，就得判断消费者的数量了，如果消费者的数量已经等于甚至大于队列的数量，多余的消费者就无法发挥作用了，这时候最好的方法就是通过消费者将消息转移到新的Topic中，尽可能的多一点队列，这样就可以分配更多的消费者，提升消费速度
消息批处理：
- 允许消费者一次获取多条消息，而不是一条一条地处理。

处理消息积压时，重要的是要根据具体的应用场景和资源限制选择最合适的方法。同时，持续监控系统性能和消息队列状态对于预防未来的问题也是至关重要的。

消息事务

Spring Boot整合RocketMQ

延时队列

RocketMQ延时队列

附件