0%

RocketMq学习笔记(一)

RocketMQ 概况

RocketMQ是什么?

特点功能:异步、削峰、解耦

上图是一个典型的消息中间件收发消息的模型,RocketMQ也是这样的设计,简单说来,RocketMQ具有以下特点:

  • 是一个队列模型的消息中间件,具有高性能、高可靠、高实时、分布式特点。
  • Producer、Consumer、队列都可以是分布式。
  • Producer向一些队列轮流发送消息,队列集合称为Topic,Consumer如果做广播消费,则一个Consumer实例消费这些Topic对应的所有队列,如果做集群消费,则多个Consumer实列平均消费这个Topic对应的集合
  • 能够严格保证消息的顺序
  • 提供丰富的消息拉取模式
  • 高效的订阅者水平扩展能力
  • 实时的消息订阅机制
  • 亿级消息堆积能力
  • 较少的依赖

RocketMQ物理部署

如上图所示, RocketMQ的部署结构有以下特点:

  • NameServer是一个几乎无状态节点,可集群部署,节点之间无任何信息同步
  • Broker部署相对复杂,Broker分为Master与Slave,一个Master可以对应多个Slave,但是一个Slave只能对应一个Master,Master与Slave的对应关系通过指定相同的BrokerName,不同的BrokerId来定义,BrokerId为0表示Master,非0表示Slave。Master也可以部署多个。每个Borker与NameServer集群中所有节点都建立长连接,定时注册topic信息到所有NameServer。
  • Producer与NameServer集群中的其中一个节点(随机选择)建立长连接,定期从NameServer取Topic路由信息,并向提供Topic服务的Master建立长连接,且定时向Master发送心跳。Producer完全无状态,可集群部署。
  • Consumer与NameServer集群中的其中一个节点(随机选择)建立长连接,定期从NameServer取Topic路由信息,并向提供Topic服务的Master、Slave建立长连接,且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息,也可以从Slave订阅消息,订阅规则由Broker配置决定。

单机部署步骤

1. 下载RocketMQ,现在版本4.6.1
2. 目录介绍
  • bin – 启动脚本目录
  • conf –配置文件目录
  • lib –依赖目录
  • benchmark – 测试demo目录
3.启动RocketMQ
1
cd /usr/local/rocketmq-all-4.6.1-bin-release/bin
  1. 启动NameServer
1
2
3
4
# 1.启动NameServer
nohup sh mqnamesrv &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log
  1. 启动Broker
1
2
3
4
# 1.启动Broker
nohup sh mqbroker -n localhost:9876 &
# 2.查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log
  • 虚拟机中Broker启动失败,需要将启动配置的内存减少
1
2
3
4
5
6
7
8
#编辑配置文件
vim runserver.sh
vim runbroker.sh
#配置修改为
#runbroker
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"
#runserver
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
  • 查看进程
1
jps
1
2
3
2042 NamesrvStartup
2186 Jps
2125 BrokerStartup

启动成功!

4.测试RocketMQ
  1. 发送消息
1
2
3
4
# 设置环境变量
export NAMESRV_ADDR=localhost:9876
# 使用安装包中Demo发送消息
sh tools.sh org.apache.rocketmq.example.quickstart.Producer
  1. 接受消息
1
2
3
4
# 设置环境变量
export NAMESRV_ADDR=localhost:9876
# 使用安装包中Demo发送消息
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
  1. 关闭RocketMQ
1
2
3
4
# 关闭NameServer
sh mqshutdown namesrv
# 关闭Broker
sh mqshutdown broker

集群部署

集群模式
  1. 单Master模式

这种方式风险大,一旦broker重启或宕机,会导致服务整个不可用。不建议线上使用,可用于本地测试。

  1. 多Master模式

一个Broker集群,全是Master无Slave。

  • 优点:配置简单,单个Master宕机或重启维护对应用无影响,在磁盘配置为RAID10时,即使宕机不可恢复情况下,由于RAID10磁盘非常可靠,消息也不丢失,性能最高;
  • 缺点:单台机器宕机期间,这台机器上未被消费的消息在机器恢复之前不可订阅,消息实时性会受到影响。
  1. 多Master多Slave模式(异步)

每个Master配置一个Slave,有多对Master-Slave,HA采用异步复制方式,主备有短暂消息延迟(毫秒级),这种模式的优缺点如下:

  • 优点:即使磁盘损坏,消息丢失非常少,且消息实时性不会受到影响,同时Master宕机后,消费者仍然可以冲Slave消费,而且此过程应用透明,不需要人工干预,性能同多Master模式几乎一致;
  • 缺点:Master宕机,磁盘损坏的情况下会丢失少量消息。
  1. 多Master多Slave模式(同步)

每个Master配置一个Slave,有多对Master-Slave,HA采用同步双写方式,即只有主备都写成功,才向应用返回成功,这种模式的优缺点如下:

  • 优点:数据与服务都无单点故障,Master宕机情况下,消息无延迟,服务可用性与数据可用性都非常高;
  • 缺点:性能比异步复制模式略低(约10%),发送单个消息的RT(Responses Time)会略高,且目前版本在主节点宕机后,备机不能自动切换为主机。
同步双主双从集群搭建(上图的模式)

  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建立链接通道,开始消费消息。

1.配置目录说明

在conf目录中有对应模式的配置文件demo,其中

  • 2m-2s-async 双主双从异步模式

  • 2m-2s-sync 双主双从同步模式

  • 2m-noslave 多master模式

  • broker.conf 单master模式

2.准备工作

准备5台虚拟机,在NameServer虚拟机上启动2个NameServer服务,四台Broker主机分别组成2m-2s模式

序号 Ip 角色 服务数
1 192.168.2.107 NameServer 1
2 192.168.2.105 BrokerServer(Master1)、NameServer 2
3 192.168.2.106 BrokerServer(Slave1) 1
4 192.168.2.103 BrokerServer(Master2) 1
5 192.168.2.110 BrokerServer(Slave2) 1

为了配置方便,将修改服务器host

1
vim /etc/hosts
1
2
3
4
5
6
7
8
#nameserver
192.168.2.107 rocketmq-nameserver1
192.168.2.105 rocketmq-nameserver2
#broker
192.168.2.105 rocketmq-master1
192.168.2.106 rocketmq-slave1
192.168.2.103 rocketmq-master2
192.168.2.110 rocketmq-slave2

配置成功,重启网卡

1
systemctl restart network

关闭防火墙

1
2
3
4
5
6
#关闭防火墙
systemctl stop firewalld.service
#查看防火墙状态
firewall-cmd --state
#禁止开机启动
systemctl disable firewalld.service

或者开放相应端口

1
2
firewall-cmd --remove-port=9876/tcp --permanent
firewall-cmd --reload

配置rocketmq的bin目录为环境变量

1
vim /etc/profile
1
2
3
4
#set rocketmq
ROCKETMQ_HOME=/usr/local/rocketmq-all-4.6.1-bin-release
PATH=$PATH:$ROCKETMQ_HOME/bin
export ROCKETMQ_HOME PATH
1
source /etc/profile

创建消息存储路径

1
2
3
4
mkdir -p /user/local/logs/rocketmq/store
mkdir -p /user/local/logs/rocketmq/store/commitlog
mkdir -p /user/local/logs/rocketmq/store/consumequeue
mkdir -p /user/local/logs/rocketmq/store/index

编辑配置文件

进入conf中的2m-2s-sync目录根据模板编辑。

Broker-Master配置文件参考

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
#所属集群名字
brokerClusterName=rockermq-cluster
#broker名字
brokerName=broker-a
#0表示Master >0表示Slave
brokerId=0
#nameServer地址,分号分隔
namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2: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=/user/local/logs/rocketmq/store
#commitLog存储路径
storePathCommitLog=/user/local/logs/rocketmq/store/commitlog
#消费队列存储路径
storePathConsumeQueue=/user/local/logs/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/user/local/logs/rocketmq/store/index
#storepoint 文件存储路径
storeCheckpoint=/user/local/logs/rocketmq/store/checkpoint
#abort文件存储路径
abortFile=/user/local/logs/rocketmq/store/abort
#限制消息大小
maxMessageSize=65536
#Broker角色
# ASYNC_MASTER 异步复制Master
# SYNC_MASTER 同步双写Master
# SLAVE
brokerRole=SYNC_MASTER
#刷盘方式
# ASYNC_FLUSH 异步刷盘
# SYNC_FLUSH 同步刷盘
flushDiskType=SYNC_FLUSH
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池熟练
#pullMessageThreadPoolNums=128

slave配置只需要修改brokerId为1和brokerRole为SlAVE即可,brokerName要与Master一致,不同的Master只要brokerName不一样即可。

启动服务

NamerServer主机,Master1主机,启动NamerServer服务

1
nohup sh mqnamesrv &

其他Broker主机,指定自己的配置启动

1
2
# 启动broker-master
nohup sh mqbroker -c /usr/local/rocketmq-all-4.6.1-bin-release/conf/2m-2s-sync/broker-a.properties &
1
2
# 启动broker-slave
nohup sh mqbroker -c /usr/local/rocketmq-all-4.6.1-bin-release/conf/2m-2s-sync/broker-a-s.properties &

集群搭建完成!

集群监控平台搭建

1. 下载项目

下载rocket-console,SpringBoot项目。

2. 修改配置文件

在application.properties中配置NameServer服务

1
rocketmq.config.namesrvAddr=192.168.2.107:9876;192.168.2.105:9876
3. 编译打包部署
1
mvn clean package -Dmaven.test.skip=true
1
java -jar rocketmq-console.jar

默认访问地址为 http://IP:8080