RocketMQ的安装和原理

.RocketMQ的安装

一.RocketMQ安装

1.1.下载RocketMQ

下载地址：http://rocketmq.apache.org/release_notes/release-notes-4.2.0/

在这里插入图片描述

下载后解压

Bin : 可执行文件目录
config：配置文件目录
Lib : 依赖库，一堆Jar包

1.2.配置ROCKETMQ_HOME

解压压缩包，配置 ROCKETMQ_HOME
在这里插入图片描述

1.3.启动MQ

启动NameServer

Cmd命令框执行进入至‘MQ文件夹\bin’下，然后执行 start mqnamesrv.cmd，启动NameServer。

成功后会弹出提示框，此框勿关闭。

启动Broker

进入至‘MQ文件夹\bin’下，修改Bean目录下的 runbroker.cmd 中JVM占用内存大小
在这里插入图片描述

CMD执行start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true ，启动Broker。

成功后会弹出提示框，此框勿关闭

1.4.RocketMQ存储结构

RocketMQ安装好之后会在用户目录下产生一个store目录用来存储相关数据：

Commitlog : 消息是存储写在commitlog目录中，一mapperdFile文件顺序存储消息。
Config : 存放运行期间的配置文件
Consumerqueue : 该目录中存放的是队列，consume queue存放着commitlog中的消息的索引位置
Index :存放着消息索引文件 indexFile，用来实现根据key进行消息的快速查询
Abort : 该文件在broker启动后自动创建，正常关闭abort会消失
Checkpoint ：记录 Commitlog ，Consumerqueue 和index 文件的最后刷盘时间戳

[问]RocketMQ数据存储在磁盘会影响性能吗？

不会，RocketMQ的性能在所有的MQ中是比较高的，主要是因为RocketMQ使用了mmap零拷贝技术，consumequeue中的数据是顺序存放的，还引入了PageCache的预读取机制，使得对 consumequeue文件的读取几乎接近于内存读取，即使在有消息堆积情况下也不会影响性能。

2.RocketMQ插件-控制台

为了方便管理，我们需要安装一个可视化插件

2.1.下载插件

RocketMQ可视化管理插件下载地址：https://github.com/apache/rocketmq-externals/releases
在这里插入图片描述

2.2.修改配置

解压后，修改配置：src/main/resource/application.properties ,这里需要指向Name Server 的地址和端口如下：
在这里插入图片描述

2.3.打包插件

回到安装目录,执行: mvn clean package -Dmaven.test.skip=true ，然后会在target目录生成打包后的jar文件
在这里插入图片描述

2.4.启动插件

进入 target 目录，执行 java -jar rocketmq-console-ng-1.0.0.jar ，访问 http://localhost:8080

二.RocketMQ的原理

1.RokcetMQ架构

RocketMQ开发官方文档：

https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md

RocketMQ的集群架构如下
在这里插入图片描述

RocketMQ架构上主要分为四部分，如上图所示

1.1.Producer

消息发布的角色，支持分布式集群方式部署。Producer通过nameserver的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

1.2.Consumer

消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。

1.3.Broker

Broker主要负责消息的存储、投递和查询以及服务高可用保证。

1.4.NameServer

NameServer是一个Broker与Topic路由的注册中心支持Broker的动态注册与发现主要包括两个功能

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

2.RocketMQ入门

官方案例：https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md

2.1.导入依赖

注意和安装的MQ版本一致

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.8.0</version>
</dependency>

2.2.生产者

步骤分析

创建producer组
设置NameServer地址
startr生产者
发送消息获取结果

结束producer

//消息发送者
public class ProducerTest {

 public static void main(String[] args) {
     try {
         // 实例化消息生产者Producer
         DefaultMQProducer producer = new DefaultMQProducer("producergroup");
         // 设置NameServer的地址
         producer.setNamesrvAddr("localhost:9876");
         // 启动Producer实例
         producer.start();
         for (int i = 0; i < 100; i++) {
             //构建消息
             Message message = new Message("topic_log","tags_error",("我是消息"+i).getBytes());
             SendResult sendResult = producer.send(message);
             System.out.printf("%s%n", sendResult);
         }

         // 如果不再发送消息，关闭Producer实例。
         producer.shutdown();

     }catch (Exception e){
         e.printStackTrace();
     }
 }

}

代码解释：

DefaultMQProducer ： MQ生产者，可以指定组名 producerGroupName
producer.setNamesrvAddr ：指定Name Server地址，用作Brocker发现。注意IP和启动name server服务时指定的IP保持一致。
producer.start() ：启动生产者
new Message(“topic_log”,“tags_error”,(“我是消息”+i).getBytes()) ：消息，参数为：topic，tags,内容
producer.send(message) : 发送消息
SendResult ：发送结果，其中包含
- sendStatus=SEND_OK ：发送状态
- msgId ：producer 创建的消息ID
- offsetMsgId ：Brocker创建的消息ID
- messageQueue ：消息存储的队列
- producer.shutdown():关闭生产者

2.3.消费者

创建consumer组
设置Name Server地址
设置消费位置，从最开始销毁
设置消息回调处理监听 -> 处理消息

Start consumer

//消息发送者
public class ConsumerTest {

 public static void main(String[] args) {
     try {
         // 实例化消息生产者Producer
         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer ("consumergroup");

         // 设置NameServer的地址
         consumer.setNamesrvAddr("127.0.0.1:9876");

         //从最开始的位置开始消费
         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

         // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
         //和发送者保持一致才能搜到消息
         consumer.subscribe("topic_log", "tags_error");

         // 注册回调实现类来处理从broker拉取回来的消息
         consumer.registerMessageListener(new MessageListenerConcurrently() {
             @Override
             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                             ConsumeConcurrentlyContext context) {

                 System.out.printf("%s 成功搜到消息: %s %n", Thread.currentThread().getName(), msgs);

                 // 标记该消息已经被成功消费
                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
             }
         });
         // 启动Producer实例
         consumer.start();

     }catch (Exception e){
         e.printStackTrace();
     }
 }

}

DefaultMQPushConsumer ：消费者，可以指定 consumerGroupName
consumer.setNamesrvAddr : 设置name server 地址
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET) ：从什么位置开始消费
consumer.subscribe(“topic_log”, “tags_error”) ：订阅某个topic下的某个tags的消息
consumer.registerMessageListener ：注册消息监听器，拿到消息后，进行消息处理。
ConsumeConcurrentlyStatus ：消费者消费结果状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS代表成功

3.RocketMQ 核心概念

3.1.RocketMQ工作原理

在这里插入图片描述

1.RocketMQ 网络部署特点

为了增强Broker性能与吞吐量，Broker一般都是以集群形式出现的。各集群节点中可能存放着相同Topic的不同Queue。

不过，这里有个问题，如果某Broker节点宕机，如何保证数据不丢失呢？其解决方案是，将每个Broker集群节点进行横向扩展，即将Broker节点再建为一个HA集群，解决单点问题。

Broker节点集群是一个主从集群，即集群中具有Master与Slave两种角色。Master负责处理读写操作请求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slave则会自动切换为Master去工作。所以这个Broker集群是主备集群。Consumer既可以从Master订阅消息，也可以从Slave订阅消息

一个Master可以包含多个Slave，但一个Slave只能隶属于一个Master。 Maste与Slave 的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。BrokerId为0表示Master非0表示Slave。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

2.RocketMQ工作流程

启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。

topicid	broker	队列
topic_log	1	队列1,队列2
topic_log	2	队列3,队列4

Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息

3.2.Producer 生产者

RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。

RocketMQ中的消息生产者都是以生产者组（Producer Group）的形式出现的。生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。

Producer会使用一定的算法(随机轮询+规避故障)选择把消息发送到哪个master的某个queue中。

3.3.Consumer 消费者

Consumer 支持两种消费形式：拉取式消费、推动式消费。（主动，被动），RocketMQ中的消息消费者都是以消费者组（Consumer Group）的形式出现的。消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息，不同的 Consumer Group可以消费同一个Topic。

在这里插入图片描述

一个Consumer Group内的Consumer可以消费多个Topic的消息。
在这里插入图片描述

[注意] 集群模式:一个Queue是不能被同一个ConsumerGroup中的多个Consumer消费的，目的是减少资源竞争提升整体性能。

3.4.Topic 消息主题

Topic表示一类消息的集合，每个topic主题包含若干条message消息，每条message消息只能属于一个topic主题，Topic是RocketMQ进行消息订阅的基本单位。

3.5.Message

消息是指消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题(topic)。

3.6.Tag 标签

为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。Topic是消息的一级分类，Tag是消息的二级分类

3.7.MessageQueue

一个Topic中可以包含多个Queue，一个Topic的Queue也被称为一个Topic中消息的分区（Partition）。

在集群模式下, 在一个Consumer Group内，一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。这样的分配规则，每个Queue只有一个消费者，可以避免消费过程中的多线程处理和资源锁定，有效提高各Consumer消费的并行度和处理效率。

消费者组中Consumer的数量应该小于等于订阅Topic的Queue数量。如果超出Queue数量，则多出的 Consumer将不能消费消息。如果一个Consmer挂了，该Consumer Group中的其它Consumer可以接着消费原Consumer消费的Queue。
在这里插入图片描述

【注意】一个Topic可以对应多个消费者，一个Queue只能对应一个组中的一个消费者。

【注意】为了防止消息紊乱，一个Consumer Group 中的Consumer都是订阅相同Topic下的Queue。

在广播模式下一个队列要可以被多个消费者对应.在广播模式下，同一个 ConsumerGroup 中的每个 Consumer 监听全部的队列。需要注意的是，广播模式下因为每个 Consumer 实例都需要处理全部的消息，因此这种模式仅推荐在**通知推送、配置同步类小流量场景使用。
在这里插入图片描述

读写队列

Queue分为写队列和读队列，默认创建数量是都是4 ，这个读写队列是从逻辑上进行划分在物理上读/写是一个队列，Producer发送的消息进入写队列，Consumer从读队列获取数据，一半情况下读写队列数量是一样的。

可以通过可视化界面修改Topic中的队列数量

perm用于设置对当前创建Topic的操作权限：2表示只写，4表示只读，6表示读写。

3.8.MessageId/Key

RocketMQ中每个消息拥有唯一的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。不过需要注意的是，MessageId有两个：在生产者send()消息时会自动生成一个MessageId（msgId)，

当消息到达Broker后，Broker也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId与key都称为消息标识。

3.9.Rebalance重新负载

当消费者数量或者Queue的数量修改，Rebalance是把⼀个Topic下的多个Queue重新分配给Consumer Group下的Consumer。目的是增加消费能力。

由于一个队列值分配给一个Consumer,那么当Consumer Group中的消费者数量大于队列数量，那么多出来的Consumer分配不到队列。

3.10.消息拉取模式

消息的消费分为：拉取式 pull ,和推送式 push

Pull:拉取式，需要消费者间隔一定时间就去遍历关联的Queue,实时性差但是便于应用控制消息的拉取
Push:推送式，封装了Queue的遍历，实时性强，但是对系统资源占用比较多。

3.11.消息消费模式

广播模式(一对多):同一个Consumer Group 下的所有Consumer都会收到同一个Topic的所有消息。同一个消息可能会被消费多次。
集群模式(一对一):同一个Gonsumer Group 下的Consumer平分同一个Topic下的消息。同一个消息只是被消费一次。

3.12.Queue的分配算法

Queue是如何分配给Consumer的，这对应了四种算法：平均分配策略，环形平均策略，一致性Hash策略，同机房策略。

平均分配【默认】：根据 qeueuCount (4)/ consumerCount (3)作为每个消费者平均分配数量，如果多出来的queue就再依次逐个分配给Consumer。
环形平均策略：根据消费者的顺序，一个一个的分配Queue即可类似于发扑克牌。
一致性Hash策略：该算法将Consumer的Hash值作为节点放到Hash环上，然后将Queue的hash值也放入Hash环上，通过顺时针进行就近分配。
同机房策略：该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。然后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue，则按照平均分配策略或环形平均策略对所有queue进行分配。

平均分配性能比较高，一致性Hash性能不高，但是能减少Rebalance，如果Consumer数量变动频繁可以使用一致性Hash。

3.13.Offset管理

RockertMQ通过Offset来维护Consumer的消费进度，比如：消费者从哪个位置开始持续消费消息的？这里有三个枚举来指定从什么位置消费

CONSUME_FROM_LAST_OFFSET：从queue的最后一条消息开始消费
CONSUME_FROM_FIRST_OFFSET：从queue的第一条消息开始消费
CONSUME_FROM_TIMESTAMP：从某个时间戳位置的消息开始消费。

消费者消费结束之后，会向Consumer会提交其消费进度offset给Broker。Offset信息的存储分为本地 Offset管理和远程Offset管理

远程Offset管理:Brocker通过 store/config/consumerOffset.json 文件以JSON方式来存储offset相关数据以json的形式：适用于集群模式
本地Offset管理：offset相关数据以json的形式持久化到Consumer本地磁盘文件中，路径为当前用户主目录下的.rocketmq_offsets/ ${clientId}/$ {group}/Offsets.json ：适用于广播模式

Offset的同步提交与异步提交: 集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset，其提交方式分为两种：

同步提交：消费者在消费完一批消息后会向broker提交这些消息的offset，等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取 nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量。
异步提交：消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset 后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取 nextBeginOffset。