RocketMQ架构、工作流程以及安装

目次
一.MQ概述
1.简介
2.用途
限流削峰
异步解耦 
数据网络 
3.MQ对比
二. RocketMQ概述
1.根本概念
消息（Message）
主题（Topic）
标签（Tag）
队列（Queue）
消息标识（MessageId/Key）
 2.体系架构
Producer
Consumer
NameServer
Broker
工作流程
三.RocketMQ的启动
1.安装JDK
2.设置RocketMQ
①修改NameServer和Broker的内存参数
②设置 NameServer 的情况变量
③修改broker的设置文件
3.启动RocketMQ
①创建一个logs文件夹，用于存放日志（切换到该目次）
②启动NameServer
③启动broker
4.关闭RocketMQ
①关闭broker
②关闭nameserver
5.设置rocketmq-dashboard

---

一.MQ概述

1.简介

   MQ，Message Queue，是一种提供消息队列服务的中央件，也称为消息中央件，是一套提供了消息生产、存储、消耗全过程API的软件体系。消息即数据。一样平常消息的体量不会很大。
  2.用途

限流削峰

   MQ可以将体系的超量哀求暂存此中，以便体系后期可以逐步举行处理惩罚，从而制止了哀求的丢失或体系 被压垮。
  

异步解耦 

   上游体系对鄙俚体系的调用若为同步调用，则会大大低沉体系的吞吐量与并发度，且体系耦合度太高。而异步调用则会办理这些标题。以是两层之间若要实现由同步到异步的转化，一样平常性做法就是，在这两层间添加一个MQ层
  

数据网络 

分布式体系会产生海量级数据流，如：业务日志、监控数据、用户活动等。
针对这些数据流举行实时或批量收罗汇总，然后对这些数据流举行大数据分析，这是当前互联网平台的必备技能。通过MQ完成此类数据网络是最好的选择。（如Kafka）
3.MQ对比

二. RocketMQ概述

1.根本概念

消息（Message）

消息是指，消息体系所传输信息的物理载体，生产和消耗数据的最小单位，每条消息必须属于一个主题。
主题（Topic）

Topic表现一类消息的聚集，每个主题包罗多少条消息，每条消息只能属于一个主题，是RocketMQ举行消息订阅的根本单位。
标签（Tag）

为消息设置的标签，用于同一主题下区分差别范例的消息。来自同一业务单位的消息，可以根据差别业务目的在同一主题下设置差别标签。标签可以或许有用地保持代码的清楚度和连贯性，并优化RocketMQ提供的查询体系。消耗者可以根据Tag实现对差别子主题的差别消耗逻辑，实现更好的扩展性。
队列（Queue）

存储消息的物理实体。一个Topic中可以包罗多个Queue，每个Queue中存放的就是该Topic的消息。一个Topic的Queue也被称为一个Topic中消息的分区（Partition）。
一个 Topic 的 Queue 中的消息只能被一个消耗者组中的一个消耗者消耗。一个Queue中的消息不允许同一个消耗者组中的多个消耗者同时消耗。

消息标识（MessageId/Key）

RocketMQ中每个消息拥有自己的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。MessageId有两个：

• 在生产者send()消息时会自动天生一个MessageId（msgId)
  
• 当消息到达Broker后，Broker也会自动天生一个MessageId(offsetMsgId)。
  

msgId、offsetMsgId与key都称为消息标识。

• msgId：由producer端天生，其天生规则为：producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode +当前时间 + AutomicInteger自增计数器。[重复概率较低]
  
• offsetMsgId：由broker端天生，其天生规则为：brokerIp + 物理分区的offset（Queue中的偏移量） [重复概率较高]
  
  
  
• key：由用户指定的业务相干的唯一标识。[用户可控制不重复]
  

 2.体系架构

RocketMQ架构上重要分为四部分构成：
Producer

   消息生产者，负责生产消息。Producer通过MQ的负载平衡模块选择相应的Broker集群队列(先选择Broker，再选择队列)举行消息投递，投递的过程支持快速失败而且低延伸。
  RocketMQ中的消息生产者都是以生产者组（ProducerGroup）的情势出现的。一个生产者组可以同时发送多个主题的消息。

Consumer

   消息消耗者，负责消耗消息。一个消息消耗者会从Broker服务器中获取到消息，并对消息举行相干业务处理惩罚。
  RocketMQ中的消息消耗者都是以消耗者组（ConsumerGroup）的情势出现的。消耗者组是同一类消耗者的聚集，这类Consumer消耗的是类似Topic范例（可以是一个，可以是多个，但是要类似）、而且是类似的tag（可以是一个，可以是多个，但是要类似）（包管订阅关系的同等性）。消耗者组使得在消息消耗方面，容易实现

• 负载平衡（将一个Topic中的差别的Queue均匀分配给同一个ConsumerGroup的差别的Consumer，注意，并不是将消息负载平衡）
  
• 容错（一个Consmer挂了，该ConsumerGroup中的别的Consumer可以接着消耗原Consumer消耗的Queue）
  

注意：

• 消耗者组中Consumer的数量应该小于即是订阅Topic的Queue数量。假如超出Queue数量，则多出的Consumer将不能消耗消息。
  
• 一个Topic范例的消息可以被多个消耗者组同时消耗
  
• 消耗者组必要包管订阅关系的同等性。这类Consumer消耗的是类似Topic范例（可以是一个，可以是多个，但是要类似）、而且是类似的tag（可以是一个，可以是多个，但是要类似）
  

NameServer

   NameServer是一个Broker与Topic路由的注册中央，支持Broker的动态注册与发现。
  重要包罗两个功能：

• Broker管理：继承Broker集群的注册信息而且生存下来作为路由信息的根本数据；提供心跳检测机制，查抄Broker是否还存活。
  
• 路由信息管理：每个NameServer中都生存着Broker集群的整个路由信息和用于客户端查询的队列信息。Producer和Conumser通过NameServer可以获取整个Broker集群的路由信息，从而举行消息的投递和消耗。
  

①路由注册
   NameServer通常也是以集群的方式摆设，不外，NameServer是无状态的，即NameServer集群中的各个节点间是无差异的，各节点间相互不举行信息通讯。
  
  各节点的数据同步：在Broker节点启动时，轮询NameServer列表，与每个NameServer节点创建长毗连，发起注册哀求。在NameServer内部维护着⼀个Broker列表，用来动态存储Broker的信息。
  注意：NameServer的无状态特性使得它和ZooKeeper、Eureka、Nacos等注册中央差别

• 长处：NameServer集群搭建简朴，扩容简朴。
  
• 缺点：对于Broker，必须明白指出全部NameServer地点。否则未指出的将不会去注册。也正因 为云云，NameServer并不能任意扩容。由于，若Broker不重新设置，新增的NameServer对于 Broker来说是不可见的，其不会向这个NameServer举行注册。
  

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的路由发现接纳的是Pull模子。当Topic路由信息出现变革时，NameServer不会自动推送给客户端，而是客户端定时拉取主题最新的路由。默认客户端每30秒会拉取一次最新的路由。
增补：

• Push模子：推送模子。是一个“发布-订阅”模子（客户端向服务端订阅），必要维护一个长毗连。一旦服务端数据发生变革，立刻推送到客户端，实时性较好。但 长毗连的维护是必要服务端的资源本钱的，该模子得当于实时性要求较高 的场景
  
• Pull模子：拉取模子。存在的标题是，实时性较差。由客户端定时向服务端拉取数据。
  
• Long Polling模子：长轮询模子。客户端发起哀求后，服务端不会立即返回哀求结果，而是将哀求挂起期待一段时间，假如此段时间内服务端数据变动，立即相应客户端哀求，如果不绝无变革则比及指定的超时时间后相应哀求，客户端重新发起长链接。
  
  
  

 实在这是对Push与Pull模子的整合，充实使用了这两种模子的优 势，屏蔽了它们的劣势。
④客户端选择NameServer的计谋
   这里的客户端指的是Producer与Consumer
  客户端在设置时必须要写上NameServer集群的地点
客户端起首会生产一个随机数，然后再与NameServer节点数量取模，此时得到的就是所要毗连的 节点索引，然后就会举行毗连。假如毗连失败，则会接纳round-robin计谋，逐个实行着去毗连别的节点。 起首接纳的是随机计谋举行的选择，失败后接纳的是轮询计谋。
客户端(Producer与Consumer)起首接纳的是随机计谋举行的选择，失败后接纳的是轮询计谋。
Broker

   Broker充当着消息的中转脚色，负责存储消息、转发消息。Broker在RocketMQ体系中负责吸收并存储从生产者发送来的消息，同时为消耗者的拉取哀求作准备。Broker同时也存储着消息相干的元数据，包罗 消耗者组消耗进度偏移offset、主题、队列等。
  集群摆设(办理单点故障)
可以将每个Broker集群节点举行横向扩展
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。

工作流程

①详细流程

1）启动NameServer，NameServer启动后开始监听端口，期待Broker、Producer、Consumer毗连。
2）启动Broker时，Broker会与全部的NameServer创建并保持长毗连，然后每30秒向NameServer定时发送心跳包。
3）发送消息前，必要先创建Topic，创建Topic时必要指定该Topic要存储在哪些Broker上以及在这些broker上创建几个Queue，在创建Topic时也会将Topic与Broker的关系写入到NameServer中。
增补：Topic的创建模式(手动创建和自动创建)
手动创建Topic时，有两种模式： 

• 集群模式：该模式下创建的Topic在该集群中，全部Broker中的Queue数量是类似的
  
  
  
• Broker模式:该模式下创建的Topic在该集群中，每个Broker中的Queue数量可以差别
  

自动创建Topic时，默认接纳的是Broker模式，会为每个Broker默认创建4个Queue
4）Producer发送消息，启动时与NameServer集群中的此中一台创建长链接，并从NameServer中获取路由信息，即当前发送的Topic消息的Queue与Broker的地点(IP+Port)的映射关系。然后根据算法计谋选择一个Queue，与队列地点的broker创建长链接从而向broker发送消息。在获取到路由信息后，Producer会起首将路由信息缓存到本地，再每隔30s从NameServer更新一个路由信息。
5）Consumer跟Producer类似，跟此中一台NameServer创建长毗连，获取其所订阅Topic的路由信息，然后根据算法计谋从路由信息中获取到其所要消耗的Queue，然后直接跟Broker创建长毗连，开始消耗此中的消息。Consumer在获取到路由信息后，同样也会每30秒从NameServer更新一次路由信息。不外差别于Producer的是，Consumer还会向Broker发送心跳，以确保Broker的存活状态。
②读写队列
从物理上来讲，读/写队列是同一个队列。以是，不存在读/写队列数据同步标题。读/写队列是逻辑上举行区分的概念。一样平常情况下，读/写队列数量是类似的。

• 当写队列多于读队列时：比方，创建Topic时设置的写队列数量为8，读队列数量为4，此时体系会创建8个Queue，分别是0 1 2 3 4 5 6 7。Producer会将消息写入到这8个队列，但Consumer只会消耗0 1 2 3 这4个队列中的消息，4 5 6 7 中的消息是不会被消耗到的。
  
• 当读队列多于写队列时：比方，创建Topic时设置的写队列数量为4，读队列数量为8，此时体系会创建8个Queue，分别是0 1 2 3 4 5 6 7。Producer会将消息写入到0 1 2 3这4个队列，但Consumer只会消耗0 1 2 3 4 5 6 7这8个队列中的消息，但是4 5 6 7中是没有消息的。此时假设ConsumerGroup中包罗两个Consuer，Consumer1消耗0 1 2 3，而Consumer2消耗4 5 6 7。但现实情况是，Consumer2是没有消息可消耗的。
  

也就是说，当读/写队列数量设置差别时，总是有标题的
那为什么还要如许计划呢？
为了方便Topic的Queue的缩容。比方，原来创建的Topic中包罗16个Queue，怎样可以或许使其Queue缩容为8个，还不会丢失消息？
可以先动态修改写队列数量为8，读队列数量稳定。此时新的消息只能写入到前8个队列，而消耗都消耗的却是 16个队列中的数据。当发现后8个Queue中的消息消耗完毕后，就可以再将读队列数量动态设置为8。整个缩容过程，没有丢失任何消息。
在rocketmq-dashboard 的 topic 设置中，perm 的值只能有 3种：

• 2：该topic 只读不写
  
• 4：该topic 只写不读
  
• 6：该topic 读写都可
  

三.RocketMQ的启动

1.安装JDK

上传到服务器后解压（记得切换目次）

1. tar -zxvf jdk-8u341-linux-x64.tar.gz

复制代码

设置情况

1. vim /etc/profile

复制代码

添加以下设置

1. # java env
2. # jdk安装路径，根据自己安装路径更改
3. export JAVA_HOME=/usr/local/java/jdk1.8.0_171
4. export CLASSPATH=.:${JAVA_HOME}/jre/lib/rt.jar:${JAVA_HOME}/lib/dt.jar:${JAVA_HOME}/lib/tools.jar
5. export PATH=$PATH:${JAVA_HOME}/bin

复制代码

重新编译文件收效

1. source /etc/profile

复制代码

2.设置RocketMQ

①修改NameServer和Broker的内存参数

修改runserver.sh
使用vim下令打开bin/runserver.sh文件

修改runbroker.sh 
使用vim下令打开bin/runbroker.sh文件

②设置 NameServer 的情况变量

 设置情况

1. vim /etc/profile

复制代码

添加以下设置（若为假造机可以写localhost）

1. export NAMESRV_ADDR=阿里云公网IP:9876

复制代码

重新编译文件收效

1. source /etc/profile

复制代码

③修改broker的设置文件

进入conf目次下，修改broker.conf文件，添加以下设置

1. namesrvAddr=阿里云公网IP:9876
2. autoCreateTopicEnable=true
3. brokerIP1=阿里云公网IP

复制代码

• namesrvAddr：nameSrv地点
  
• autoCreateTopicEnable：自动创建主题，否则必要手动创建出来
  
• brokerIP1：broker也必要一个公网ip
  

假如不指定，那么是阿里云的内网地点，我们在本地无法毗连使用
而且云服务器的安全组必要打开9876 (nameserver)，10100~11000之间的端口 (broker)，假如开了安全组本地照旧连不上，可以查察防火墙开放的端口

1. firewall-cmd --list-all

复制代码

打开防火墙端口 

1. firewall-cmd --zone=public --add-port=端口号/tcp --permanent

复制代码

重启一下（重启收效）：

1. firewall-cmd --reload

复制代码

3.启动RocketMQ

①创建一个logs文件夹，用于存放日志（切换到该目次）

②启动NameServer

1. nohup sh bin/mqnamesrv > ./logs/namesrv.log &

复制代码

③启动broker

这里的 -c 是指定使用的设置文件

1. nohup sh bin/mqbroker -c conf/broker.conf > ./logs/broker.log &

复制代码

4.关闭RocketMQ

切换到bin目次下
①关闭broker

1. ./mqshutdown broker

复制代码

②关闭nameserver

1. ./mqshutdown namesrv

复制代码

5.设置rocketmq-dashboard

下载地点：
https://github.com/apache/rocketmq-dashboard/archive/refs/tags/rocketmq-dashboard-1.0.0.zip
假如是JDK17，请先在pom.xml文件升级netty依赖，否则访问面板时会报错

1. <dependency>
2. <groupId>io.netty</groupId>
3. <artifactId>netty-all</artifactId>
4. <version>4.1.82.Final</version>
5. </dependency>

复制代码

下载后解压出来，在跟目次下实行

1. mvn clean package -Dmaven.test.skip=true

复制代码

打成jar包

将jar包上传到服务器上去 

指定NameServer的地点和启动端口(8000)以及输出日志

1. nohup java -jar rocketmq-dashboard-1.0.0.jar --server.port=8000 --rocketmq.config.namesrvAddr=127.0.0.1:9876 > ./rocketmq-all-4.9.0-bin-release/logs/dashboard.log &

复制代码

直接访问8000端口即可，假如是云服务器记得设置安全组

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金