RocketMQ基础

搬砖俱乐部

发布于 2019-12-31 13:04:20

1.1K0

发布于 2019-12-31 13:04:20

文章被收录于专栏：BanzClubBanzClub

“ RockeMQ是阿里巴巴自主研发的消息中间件，实现业务削峰，分布式事务等，已捐献给Apache，成为MQ家族的重要成员。”

—

安装RocketMQ

RocketMQ集群的部署可以分为很多种模式：

nameserver可以多个；

broker也可以部署多个：两主两从、两主无从等，本文采用两主无从的模式；

# 下载
wget http://mirrors.tuna.tsinghua.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-bin-release.zip
# 解压
unzip rocketmq-all-4.6.0-bin-release.zip
# 重命名
mv rocketmq-all-4.6.0-bin-release/ rocketmq

# 配置环境变量，三个节点都需要配置！！！
vim .bash_profile 
export ROCKETMQ=/opt/install/rocketmq/
export PATH=$PATH:$ROCKETMQ/bin
source .bash_profile 
# 修改日志文件位置
sed 's#${user.home}#/opt/install/rocketmq#g' *.xml -i
# 分发到其他节点
scp -P 12324 -r rocketmq node02:$PWD
scp -P 12324 -r rocketmq node03:$PWD

在node01上启动nameserver
nohup sh /opt/install/rocketmq/bin/mqnamesrv >/dev/null 2>&1 &

# broker配置
vim runbroker.sh
# 默认 ${JAVA_OPT} -server -Xms 8g -Xmx2 8g -Xmn 4g
# 可能机器内存不足，导致启动不起来，根据机器配置调小点
JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn1g"
# 修改配置文件，这里修改conf/2m-noslave/.properties
# 注：另一个节点是node03
brokerName=node02
# 0 （master） 非0 slave
brokerId=0
brokerRole=ASYNC_MASTER
#所属集群名字
brokerClusterName=rocketmq-cluster
#nameServer地址，分号分割
namesrvAddr=10.255.20.147:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=1073741824
#ConsumeQueue每个文件默认存30W条，根据业务情况调整
mapedFileSizeConsumeQueue=300000
#检测物理文件磁盘空间
diskMaxUsedSpaceRatio=88
#存储路径
storePathRootDir=/opt/install/rocketmq/store
#commitLog 存储路径
storePathCommitLog=/opt/install/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/opt/install/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/opt/install/rocketmq/store/index
#checkpoint 文件存储路径
storeCheckpoint=/opt/install/rocketmq/store/checkpoint
#abort 文件存储路径
abortFile=/opt/install/rocketmq/store/abort
#限制的消息大小
maxMessageSize=65536
#刷盘方式
flushDiskType=ASYNC_FLUSH

# 启动broker node02
nohup sh /opt/install/rocketmq/bin/mqbroker -c /opt/install/rocketmq/conf/2m-noslave/node02.properties >/dev/null 2>&1 &
# 启动broker node03
nohup sh /opt/install/rocketmq/bin/mqbroker -c /opt/install/rocketmq/conf/2m-noslave/node03.properties >/dev/null 2>&1 &

# rocketmq 控制台安装
# 注：需要安装git、maven
git clone https://github.com/apache/rocketmq-externals.git
cd rocketmq-externals-maste/rocketmq-console/src/main/resources/
vim application.properties
# 设置后台访问端口
server.port=8080
# nameserver配置
rocketmq.config.namesrvAddr=10.255.20.147:9876
cd  rocketmq-externals\rocketmq-console
# maven打包
mvn clean package -Dmaven.test.skip=true
前台启动：
java -jar rocketmq-console-ng-1.0.1.jar
后台启动：
nohup java -jar rocketmq-console-ng-1.0.0.jar --server.port=8080 --rocketmq.config.namesrvAddr=10.255.20.147:9876 >/dev/null 2>&1 &

—

RockMQ架构

RocketMQ是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式特点。

由四个组成部分：

NameServer：提供轻量级的服务发现和路由。每个 NameServer 记录完整的路由信息，提供相应的读写服务，支持快速的存储扩展；

Broker 管理：NameServer 接受 Broker 群集的注册，并提供心跳机制以检查 Broker 是否还存活
路由管理：每个 NameServer 保存 Broker 群集的完整路由信息以及客户端（producer和consumer）查询的 Queue 信息

Broker：提供轻量级的 TOPIC 和 QUEUE 机制来照顾消息存储。它们支持“推和拉”模型，包含容错机制（2个或3个副本），并提供强大的峰值填充功能和按其原始时间顺序累积数千亿条消息的能力。此外，代理提供故障恢复，丰富的指标统计信息和警报机制；

负责消息的存储和传递，消息查询，HA保证
- 远程处理模块：处理客户端对broker的请求；
- 客户端管理：管理客户端（producer和consumer），维护消费者的 topic 订阅；
- 存储服务：在物理磁盘上存储和查询消息；
- 高可用服务：在Master和Slave之间建立消息同步机制；
- 索引服务：通过指定key为消息建立索引，提供快速查询消息；

每个 Broker 与 Name Server 集群中的所有节点建立长连接，定时注册 Topic 信息到所有Name Server；

Broker 分为 Master 与 Slave；

Master 与 Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId 为0 表示Master，非0 表示Slave；

Producer：支持分布式部署。分布式生产者通过多种负载均衡模式将消息发送到Broker集群。发送过程支持快速失败（如何避免消息丢失）并且延迟低。

Producer 与 Name Server 集群中的其中一个节点（随机选择）建立长连接，拿Topic信息；
Producer 向提供Topic 服务的Master 建立长连接，且定时向Master 发送心跳；

Consumer：消费者也支持Push和Pull模型中的分布式部署。它还支持集群使用和消息广播。它提供了实时消息订阅机制，可以满足大多数消费者的需求。

Consumer 与Name Server 集群中的其中一个节点（随机选择）建立长连接，拿Topic信息；
Consumer 向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳；
Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定；

Topic是生产者在发送消息和消费者在拉取消息的类别。Topic与生产者和消费者之间的关系非常松散。具体来说，一个Topic可能有0个，一个或多个生产者向它发送消息；相反，一个生产者可以发送不同类型Topic的消息。类似的，消费者组可以订阅一个或多个主题，只要该组的实例保持其订阅一致即可。

Tag，换句话的意思就是子主题，为用户提供了额外的灵活性。有了Tag，来自同一业务模块的具有不同目的的消息可以具有相同的主题和不同的标记。标签有助于保持代码的清晰和连贯，同时标签也方便RocketMQ提供的查询功能。

GroupName，代表具有相同角色的生产者组合或消费者组合，称为生产者组或消费者组。

作用是在集群HA的情况下，一个生产者down之后，本地事务回滚后，可以继续联系该组下的另外一个生产者实例，不至于导致业务走不下去。在消费者组中，可以实现消息消费的负载均衡和消息容错目标。
另外，有了GroupName，在集群下，动态扩展容量很方便。只需要在新加的机器中，配置相同的GroupName。启动后，就立即能加入到所在的群组中，参与消息生产或消费。

Queue，是数据分片的产物，首先，一个Topic分布在一个Broker上的子集定义为一个Topic分片，将每个分片在划分成多个Queue，每个Topic分片等分的Queue的数量可以不同，由用户在创建Topic时指定。

RocketMQ的负载均衡策略规定：Consumer数量应该小于等于Queue数量，如果Consumer超过Queue数量，那么多余的Consumer 将不能消费消息。

在一个Consumer Group内，Queue和Consumer之间的对应关系是一对多的关系：一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。这样的分配规则，每个Queue只有一个消费者，可以避免消费过程中的多线程处理和资源锁定，有效提高各Consumer消费的并行度和处理效率。

Queue是Topic在一个Broker上的分片等分为指定份数后的其中一份，是负载均衡过程中资源分配的基本单元。