# RocketMQ架构与实战 RocketMQ是阿里巴巴中间件团队自研的一款高性能、高吞吐量、低延迟、高可用、高可靠（具备金融级稳定性）的分布式消息中间件，开源后并于2016年捐赠给Apache社区孵化，目前已经成为了Apache顶级项目。当前在国内被广泛的使用，包括互联网、电商、金融、企业服务等领域，包括：字节跳动、滴滴、微众银行等知名的互联网公司。 ## 1. 介绍 RocketMQ在阿里内部叫做Metaq（最早名为Metamorphosis，中文意思“变形记”，是作家卡夫卡的中篇小说代表作，可见是为了致敬Kafka）RocketMQ是Metaq 3.0之后开源的版本。 Metaq在阿里巴巴集团内部、蚂蚁金服、菜鸟等各业务中被广泛使用，接入了上万个应用系统中。并平稳支撑了历年的双十一大促（万亿级的消息），在性能、稳定性、可靠性等方面表现出色，在整个阿里技术体系和大中台战略中发挥着举足轻重的作用。 Metaq最早源于Kafka，早期借鉴了Kafka很多优秀的设计。但是由于Kafka是Scale语言编写而阿里系主要使用Java，且无法满足阿里的电商、金融业务场景，所以誓嘉（花名）团队用Java重新造轮子，并做了大量的改造和优化。 在此之前，淘宝有一款消息中间件名为Notify，目前已经逐步被Metaq所取代 第一代的Notify主要使用了推模型，解决了事务消息；第二代的MetaQ主要使用了拉模型，解决了顺序消息和海量堆积的问题。相比起Kafka使用的Scale语言编写，RabbitMQ 使用Erlang语言编写，基于Java的RocketMQ开源后更容易被广泛的研究，以及其他大厂定制开发 ## 2. 使用场景 - 应用解耦 - 流量削峰 - 数据分发 通过消息队列可以让数据在多个系统之间进行流通。数据的产生方不需要关心谁来使用数据，只需要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可 ## 3. 部署架构 ### 3.1 角色介绍 - producer: 消息的发送者 与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic 服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署 - consumer:消息的接收者 与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从Master还 是Slave拉取。 - broker: 暂存和传输消者 分为master和slave,一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。 注意：当前RocketMQ版本在部署架构上支持一Master多Slave，但只有BrokerId=1的从服务器才会参与消息的读负载 - NameServer: 箮理broker 是一个无状态节点,可集群部署,节点之间无任何信息同步. - topic:区分消息的各类; 一个发送者可以发送消息给一个或者多个topic;一个消息的接收者可以订阅一个或者多个topic消息 - message queue:相当于topic的分区,用于并行发送和接收消息 ![示意图](https://img.zhaojq.top/20260727091651681.png "示意图") ### 3.2 执行流程 1. 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。 2. Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。 3. 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic 4. Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息 5. Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息 ## 4. 特性 - 订阅与发布 >tips: 消息的订阅是指某个消费者关注了某个topic中带有某些tag的消息 - 消息顺序 消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了三条消息分别是订单创建、订单付款、订单完成。消费时要按照这个顺序消费才能有意义，但是同时订单之间是可以并行消费的。RocketMQ可以严格的保证消息有序。 - 消息过滤 根据tag进行过滤,也支持自定义属性过滤. 在broker端实现,对少了对consumer无用消息的网络传输,但增加了broker的负担 - 消息可靠性 影响可靠性几种情况 ``` 1.broker非正常关闭 2.broker异常crash 3.os crash 4.机器掉电,但是立即恢复供电 5.机器无法开机 6.磁盘设备损坏 ``` 1,2,3,4都属于硬件资源可立即恢复情况,可以保证消息不丢,或者丢失少量数据(依赖刷盘方式是同步还是异步) 5,6属于单点故障,且无法恢复,消息全部丢失,但可以通过异步复制,可保证99%消息不丢 - 至少一次 至少一次(At least Once)指每个消息必须投递一次。Consumer先Pull消息到本地，消费完成后，才向服务器返回ack，如果没有消费一定不会ack消息，所以RocketMQ可以很好的支持此特性。 - 回溯消费 回溯消费是指Consumer已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间维度来回退消费进度。RocketMQ支持按照时间回溯消费，时间维度精确到毫秒 - 事务消息 指应用本地事务和发送消息操作可以被定义到全局事务中,要么同时成功,要么同时失败 RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致性。 - 定时消息 定时消息（延迟队列）是指消息发送到broker后，不会立即被消费，等待特定时间投递给真正的topic。 broker有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”，18个level。 ``` level == 0，消息为非延迟消息 1<=level<=maxLevel，消息延迟特定时间，例如level==1，延迟1s level > maxLevel，则level== maxLevel，例如level==20，延迟2h ``` - 消息重试 Consumer消费消息失败后，要提供一种重试机制，令消息再消费一次。Consumer消费消息失败通常可以认为有以下几种情况： - 由于消息本身的原因，例如反序列化失败，消息数据本身无法处理（例如话费充值，当前消息的手机号被注销，无法充值）等。这种错误通常需要跳过这条消息，再消费其它消息，而这条失败的消息即使立刻重试消费，99%也不成功，所以最好提供一种定时重试机制，即过10秒后再重试。 - 由于依赖的下游应用服务不可用，例如db连接不可用，外系统网络不可达等。遇到这种错误，即使跳过当前失败的消息，消费其他消息同样也会报错。这种情况建议应用sleep 30s，再消费下一条消息，这样可以减轻Broker重试消息的压力。 - 消息重投 生产者在发送消息时 ``` 同步消息失败会重投 异步消息有重试 oneway没有任何保证。 ``` 消息重投保证消息尽可能发送成功、不丢失，但可能会造成消息重复，消息重复在RocketMQ中是无法避免的问题。消息重复在一般情况下不会发生，当出现消息量大、网络抖动，消息重复就会是大概率事件。另外，生产者主动重发、consumer负载变化也会导致重复消息。 如下方法可以设置消息重试策略： - retryTimesWhenSendFailed：同步发送失败重投次数，默认为2，因此生产者会最多尝试发送retryTimesWhenSendFailed + 1次。不会选择上次失败的broker，尝试向其他broker发送，最大程度保证消息不丢失。超过重投次数，抛异常，由客户端保证消息不丢失。当出现RemotingException、MQClientException和部分MQBrokerException时会重投。 - retryTimesWhenSendAsyncFailed：异步发送失败重试次数，异步重试不会选择其他broker，仅在同一个broker上做重试，不保证消息不丢。 - retryAnotherBrokerWhenNotStoreOK：消息刷盘（主或备）超时或slave不可用（返回状态非SEND_OK），是否尝试发送到其他broker，默认false。十分重要消息可以开启。 - 流量控制 生产者流控，因为broker处理能力达到瓶颈；消费者流控，因为消费能力达到瓶颈。 ``` 生产者流控 1. commitLog文件被锁时间超过osPageCacheBusyTimeOutMills时，参数默认为1000ms，发 生流控。 2. 如果开启transientStorePoolEnable = true，且broker为异步刷盘的主机，且 transientStorePool中资源不足，拒绝当前send请求，发生流控。 3. broker每隔10ms检查send请求队列头部请求的等待时间，如果超过 waitTimeMillsInSendQueue，默认200ms，拒绝当前send请求，发生流控。 4.broker通过拒绝send 请求方式实现流量控制。 ``` 注意，**生产者流控，不会尝试消息重投。** ``` 消费者流控 - 消费者本地缓存消息数超过pullThresholdForQueue时，默认1000。 - 消费者本地缓存消息大小超过pullThresholdSizeForQueue时，默认100MB。 - 消费者本地缓存消息跨度超过consumeConcurrentlyMaxSpan时，默认2000。 - 消费者流控的结果是降低拉取频率 ``` - 死信队列 死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败，消息队列会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。 RocketMQ将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。 在RocketMQ中，可以通过使用console控制台对死信队列中的消息进行重发来使得消费者实例再次进行消费。 ## 5. 消费模式 RocketMQ消息订阅有两种模式，一种是Push模式（MQPushConsumer），即MQServer主动向消费端推送；另外一种是Pull模式（MQPullConsumer），即消费端在需要时，主动到MQ Server拉取。但在具体实现时，Push和Pull模式**本质都是采用消费端主动拉取的方式**，即consumer轮询从broker拉取消息 - push 实时性高, 但消费端处理能力有限 - pull 消费端量力而行, 但如何控制拉取频率. 一般用长轮询 - 区别 Push方式里，consumer**把长轮询的动作封装**了，并注册MessageListener监听器，取到消息后，唤醒MessageListener的consumeMessage()来消费，对用户而言，感觉消息是被推送过来的。 Pull方式里，取消息的过程需要用户自己主动调用，首先通过打算消费的Topic拿到MessageQueue的集合，遍历MessageQueue集合，然后针对每个MessageQueue批量取消息，一次取完后，记录该队列下一次要取的开始offset，直到取完了，再换另一个MessageQueue ## 6. 角色及相关术语 - 消息模型 RocketMQ主要由 Producer、Broker、Consumer 三部分组成，其中Producer 负责生产消息，Consumer 负责消费消息，Broker 负责存储消息。 Broker 在实际部署过程中对应一台服务器，每个Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。 MessageQueue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。 - Producer - consumer - pushconsumer - pullconsumer - producerGroupp 同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费 - consumerGroup 同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例**必须订阅完全相同的Topic**。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。 - broker - 广播消息 一条消息被多个 Consumer 消费，即使这些 Consumer 属于同一个 Consumer Group，消息也会被 Consumer Group 中的每个 Consumer 都消费一次，广播消费中的 Consumer Group 概念可以认为在消息划分方面无意义 - 集群消费 一个 Consumer Group 中的 Consumer 实例平均分摊消费消息。例如某个 Topic 有 9 条消息，其中一个 Consumer Group 有 3 个实例(可能是 3 个进程，或者3台机器)，那举每个实例只消费其中的 3条消息 - 顺序消息 消费消息的顺序要同发送消息的顺序一致，**在RocketMQ 中主要指的是局部顺序**，即一类消息为满足顺序性，必须Producer单线程顺序发送，且发送到同一个队列，这样Consumer 就可以按照Producer发送的顺序去消费消息 - 普通顺序消息 顺序消息的一种，正常情况下可以保证完全的顺序消息，但是一旦发生通信异常，Broker 重启，由于队列总数发生发化，哈希取模后定位的队列会发化，产生短暂的消息顺序不一致。 如果业务能容忍在集群异常情况(如某个Broker 宕机或者重启)下，消息短暂的乱序，使用普通顺序方式比较合适。 - 严格顺序消息 顺序消息的一种，无论正常异常情况都能保证顺序，但是牺牲了分布式 Failover特性，即Broker集群中只要有一台机器不可用，则整个集群都不可用，服务可用性大大降低。 如果服务器部署为同步双写模式，此缺陷可通过备机自动切换为主避免，不过仍然会存在几分钟的服务不可用。(依赖同步双写，主备自动切换，自动切换功能目前还未实现) 目前已知的应用只有数据库 binlog 同步强依赖严格顺序消息，其他应用绝大部分都可以容忍短暂乱序，推荐使用普通的顺序消息 - Message Queue 在 RocketMQ 中，所有消息队列都是持久化的，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，访问其中的存储单元使用**Offset**来访问，offset 为 java **long** 类型，64 位，理论上在 100 年内不会溢出，所以认为为是长度无限，另外队列中只保存最近几天的数据，之前的数据会按照过期时间来删除。也可以认为Message Queue是一个长度无限的数组，offset 就是下标。 - 标签Tag 为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。 ## 7. RocketMQ环境搭建 **依赖java环境** - 下载rocketmq ```sh wget https://archive.apache.org/dist/rocketmq/4.5.1/rocketmq-all- 4.5.1-bin-release.zip ``` - 修改脚本 **如果是jdk8,不用作修改**,如果是jdk8以上, 则需要对以下脚本作修改(解压后,安装包bin目录下) ```sh # runserver.sh 删除 UseCMSCompactAtFullCollection UseParNewGC UseConcMarkSweepGC 修改内存： JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m - XX:MetaspaceSize=64mm -XX:MaxMetaspaceSize=160mm" -Xloggc修改为-Xlog:gc ``` ```sh # runbroker.sh 删除： PrintGCDateStamps PrintGCApplicationStoppedTime PrintAdaptiveSizePolicy UseGCLogFileRotation NumberOfGCLogFiles=5 GCLogFileSize=30m ``` ```sh # tools.sh #删除 JAVA_OPT="${JAVA_OPT}-Djava.ext.dirs=${BASE_DIR}/lib:${JAVA_HOME}/jre/lib/ext" ``` - 启动NameServer ```sh mqnamesrv 或 nohup mqnamesrv & ``` - 启动broker ```sh # broker.conf 作如下配置 # 如果想公网访问, 就配置公网ip brokerIP1=公网/内网ip namesrvAddr=内网ip:9876 autoCreateTopicEnable=true autoCreateSubscriptionGroup=true enablePropertyFilter=true nohup sh mqbroker -c [broker.conf文件路径] >/dev/null 2>&1 & ``` 注意: 如果想公网连接rocketmq, 需开放以下端口 ``` 9876 10911 10909 ``` ## 8. 环境测试 - 发送消息 ```sh #1.设置环境变量 export NAMESRV_ADDR=localhost:9876 #2.使用安装包的Demo发送消息 sh bin/tools.shorg.apache.rocketmq.example.quickstart.Producer ``` - 接收消息 ```sh #1.设置环境变量 export NAMESRV_ADDR=localhost:9876 #2.接收消息 sh bin/tools.shorg.apache.rocketmq.example.quickstart.Consumer ``` - 关闭mq ```sh #1.关闭NameServer mqshutdown namesrv #2.关闭Broker mqshutdown broker ``` ## 9. api的使用 示例代码: [demo01](https://gitee.com/ixinglan/rocketmq-demo.git) ## 10. 与srpingboot整合 示例代码: [demo02/demo03](https://gitee.com/ixinglan/rocketmq-demo.git)