RocketMQ安装,双主双从同步集群搭建

为什么要用MQ

  • 应用解耦

  1. 系统的耦合性越高,容错性就越低。以电商应用为例,用户创建订单后,如果耦合调用库存系统、物流系统、支付系统,任何一个子系统出了故障或者因为升级等原因暂时不可用,都会造成下单操作异常,影响用户使用体验。
  2. 使用消息队列解耦合。比如物流系统发生故障,需要几分钟才能来修复,在这段时间内,物流系统要处理的数据被缓存到消息队列中,用户的下单操作正常完成。当物流系统回复后,补充处理存在消息队列中的订单消息即可,终端系统感知不到物流系统发生过几分钟故障。

  • 流量削峰

  1. 应用系统如果遇到系统请求流量的瞬间猛增,有可能会将系统压垮。有了消息队列可以将大量请求缓存起来,分散到很长一段时间处理,这样可以大大提到系统的稳定性和用户体验。

  • 数据分发

  1. 通过消息队列可以让数据在多个系统更加之间进行流通。数据的产生方不需要关心谁来使用数据,只需要将数据发送到消息队列,数据使用方直接在消息队列中直接获取数据即可。

RocketMQ安装

  1. 下载二进制版本 Binary: rocketmq-all-4.7.0-bin-release.zip [PGP] [SHA512]

  2. 解压
  3. 目录介绍

    • bin:启动脚本,包括shell脚本和CMD脚本
    • conf:实例配置文件 ,包括broker配置文件、logback配置文件等
    • lib:依赖jar包,包括Netty、commons-lang、FastJSON等

  4. 启动MQ, 日志文件在 /root/logs/rocketmqlogs

  • 修改JVM配置

# 编辑runbroker.sh和runserver.sh修改默认JVM大小, 都为1GB

vi runbroker.sh

vi runserver.sh

  • 启动NameServer

# 1.启动NameServer

./mqnamesrv &

# 2.查看启动日志

tail -f /root/logs/rocketmqlogs/namesrv.log

  • 启动Broker

# 1.启动Broker

./mqbroker -n localhost:9876 &

# 2.查看启动日志

tail -f /root/logs/rocketmqlogs/broker.log

  • 关闭RocketMQ

# 1.关闭NameServer

sh bin/mqshutdown namesrv

# 2.关闭Broker

sh bin/mqshutdown broker

  • 查看是否启动成功

[root@localhost bin]# jps

2981 Jps

2858 NamesrvStartup

2909 BrokerStartup

RocketMQ 角色介绍

  1. Producer:消息生产者,负责生产消息,一般由业务系统负责生产消息。

  2. Consumer:消息消费者,负责消费消息,一般是后台系统负责异步消费, 以下为RocketMQ不同Consumer简介。

    • Push Consumer:向Consumer对象注册一个监听器,监听推送来的消息,进行消费。

    • Pull Consumer:主动请求Broker拉取消息,然后对消息消费。

  3. Producer Group:生产者集合,一般用于发送同一类消息。

  4. Consumer Group:消费者集合,一般用于接收一类消息进行消费。

  5. Broker:MQ消息服务器,用于消息存储和转发。

  6. NameServer:Broker协调者,类似Zookeeper的功能。

  7. Topic:消息的主题,发送者可以向多个Topic发送消息,消费者也可以向多个Topic订阅消息。

  8. Message Queue:在Topic内部的小类的分区,可以进行发送和接收的处理。

RocketMQ集群模式

  1. 单Master模式

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

  1. 多Master模式

  • 一个集群无Slave,全是Master,例如2个Master或者3个Master,这种模式的优缺点如下:
  • 配置简单,单个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会略高,且目前版本在主节点宕机后,备机不能自动切换为主机。

双主双从集群搭建(同步双写模式)

RocketMQ集群.png

集群工作流程

  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. 注意关闭防火墙

配置HOST信息,方便访问,

  1. vim /etc/hosts

# nameserver

192.168.162.130 rocketmq-nameserver1

192.168.162.131 rocketmq-nameserver2

# broker

192.168.162.130 rocketmq-master1

192.168.162.131 rocketmq-slave1

192.168.162.132 rocketmq-master2

192.168.162.133 rocketmq-slave2

  1. 配置完成后, 重启网卡 systemctl restart network

修改消息存储路径

mkdir /usr/local/rocketmq/store

mkdir /usr/local/rocketmq/store/commitlog

mkdir /usr/local/rocketmq/store/consumequeue

mkdir /usr/local/rocketmq/store/index

集群搭建-Broker配置

master1

  1. vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties
  2. 修改配置如下:

#所属集群名字

brokerClusterName=rocketmq-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

#destroyMapedFileIntervalForcibly=120000

#redeleteHangedFileInterval=120000

#检测物理文件磁盘空间

diskMaxUsedSpaceRatio=88

#存储路径

storePathRootDir=/usr/local/rocketmq/store

#commitLog 存储路径

storePathCommitLog=/usr/local/rocketmq/store/commitlog

#消费队列存储路径存储路径

storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue

#消息索引存储路径

storePathIndex=/usr/local/rocketmq/store/index

#checkpoint 文件存储路径

storeCheckpoint=/usr/local/rocketmq/store/checkpoint

#abort 文件存储路径

abortFile=/usr/local/rocketmq/store/abort

#限制的消息大小

maxMessageSize=65536

#flushCommitLogLeastPages=4

#flushConsumeQueueLeastPages=2

#flushCommitLogThoroughInterval=10000

#flushConsumeQueueThoroughInterval=60000

#Broker 的角色

#- ASYNC_MASTER 异步复制Master

#- SYNC_MASTER 同步双写Master

#- SLAVE

brokerRole=SYNC_MASTER

#刷盘方式

#- ASYNC_FLUSH 异步刷盘

#- SYNC_FLUSH 同步刷盘

flushDiskType=SYNC_FLUSH

#checkTransactionMessageEnable=false

#发消息线程池数量

#sendMessageThreadPoolNums=128

#拉消息线程池数量

#pullMessageThreadPoolNums=128

slave1

  1. 修改配置文件 vi /usr/local/rocketmq/conf/2m-2s-sync/broker-a-s.properties

#所属集群名字

brokerClusterName=rocketmq-cluster

#broker名字,注意此处不同的配置文件填写的不一样

brokerName=broker-a

#0 表示 Master,>0 表示 Slave

brokerId=1

#nameServer地址,分号分割

namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876

#在发送消息时,自动创建服务器不存在的topic,默认创建的队列数

defaultTopicQueueNums=4

#是否允许 Broker 自动创建Topic,建议线下开启,线上关闭

autoCreateTopicEnable=true

#是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭

autoCreateSubscriptionGroup=true

#Broker 对外服务的监听端口

listenPort=11011

#删除文件时间点,默认凌晨 4点

deleteWhen=04

#文件保留时间,默认 48 小时

fileReservedTime=120

#commitLog每个文件的大小默认1G

mapedFileSizeCommitLog=1073741824

#ConsumeQueue每个文件默认存30W条,根据业务情况调整

mapedFileSizeConsumeQueue=300000

#destroyMapedFileIntervalForcibly=120000

#redeleteHangedFileInterval=120000

#检测物理文件磁盘空间

diskMaxUsedSpaceRatio=88

#存储路径

storePathRootDir=/usr/local/rocketmq/store

#commitLog 存储路径

storePathCommitLog=/usr/local/rocketmq/store/commitlog

#消费队列存储路径存储路径

storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue

#消息索引存储路径

storePathIndex=/usr/local/rocketmq/store/index

#checkpoint 文件存储路径

storeCheckpoint=/usr/local/rocketmq/store/checkpoint

#abort 文件存储路径

abortFile=/usr/local/rocketmq/store/abort

#限制的消息大小

maxMessageSize=65536

#flushCommitLogLeastPages=4

#flushConsumeQueueLeastPages=2

#flushCommitLogThoroughInterval=10000

#flushConsumeQueueThoroughInterval=60000

#Broker 的角色

#- ASYNC_MASTER 异步复制Master

#- SYNC_MASTER 同步双写Master

#- SLAVE

brokerRole=SLAVE

#刷盘方式

#- ASYNC_FLUSH 异步刷盘

#- SYNC_FLUSH 同步刷盘

flushDiskType=ASYNC_FLUSH

#checkTransactionMessageEnable=false

#发消息线程池数量

#sendMessageThreadPoolNums=128

#拉消息线程池数量

#pullMessageThreadPoolNums=128

master2

  1. 修改配置文件 vi /usr/local/rocketmq/conf/2m-2s-sync/broker-b.properties

#所属集群名字

brokerClusterName=rocketmq-cluster

#broker名字,注意此处不同的配置文件填写的不一样

brokerName=broker-b

#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

#destroyMapedFileIntervalForcibly=120000

#redeleteHangedFileInterval=120000

#检测物理文件磁盘空间

diskMaxUsedSpaceRatio=88

#存储路径

storePathRootDir=/usr/local/rocketmq/store

#commitLog 存储路径

storePathCommitLog=/usr/local/rocketmq/store/commitlog

#消费队列存储路径存储路径

storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue

#消息索引存储路径

storePathIndex=/usr/local/rocketmq/store/index

#checkpoint 文件存储路径

storeCheckpoint=/usr/local/rocketmq/store/checkpoint

#abort 文件存储路径

abortFile=/usr/local/rocketmq/store/abort

#限制的消息大小

maxMessageSize=65536

#flushCommitLogLeastPages=4

#flushConsumeQueueLeastPages=2

#flushCommitLogThoroughInterval=10000

#flushConsumeQueueThoroughInterval=60000

#Broker 的角色

#- ASYNC_MASTER 异步复制Master

#- SYNC_MASTER 同步双写Master

#- SLAVE

brokerRole=SYNC_MASTER

#刷盘方式

#- ASYNC_FLUSH 异步刷盘

#- SYNC_FLUSH 同步刷盘

flushDiskType=SYNC_FLUSH

#checkTransactionMessageEnable=false

#发消息线程池数量

#sendMessageThreadPoolNums=128

#拉消息线程池数量

#pullMessageThreadPoolNums=128

slave2

  1. 修改配置文件 vi /usr/local/rocketmq/conf/2m-2s-sync/broker-b-s.properties

#所属集群名字

brokerClusterName=rocketmq-cluster

#broker名字,注意此处不同的配置文件填写的不一样

brokerName=broker-b

#0 表示 Master,>0 表示 Slave

brokerId=1

#nameServer地址,分号分割

namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876

#在发送消息时,自动创建服务器不存在的topic,默认创建的队列数

defaultTopicQueueNums=4

#是否允许 Broker 自动创建Topic,建议线下开启,线上关闭

autoCreateTopicEnable=true

#是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭

autoCreateSubscriptionGroup=true

#Broker 对外服务的监听端口

listenPort=11011

#删除文件时间点,默认凌晨 4点

deleteWhen=04

#文件保留时间,默认 48 小时

fileReservedTime=120

#commitLog每个文件的大小默认1G

mapedFileSizeCommitLog=1073741824

#ConsumeQueue每个文件默认存30W条,根据业务情况调整

mapedFileSizeConsumeQueue=300000

#destroyMapedFileIntervalForcibly=120000

#redeleteHangedFileInterval=120000

#检测物理文件磁盘空间

diskMaxUsedSpaceRatio=88

#存储路径

storePathRootDir=/usr/local/rocketmq/store

#commitLog 存储路径

storePathCommitLog=/usr/local/rocketmq/store/commitlog

#消费队列存储路径存储路径

storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue

#消息索引存储路径

storePathIndex=/usr/local/rocketmq/store/index

#checkpoint 文件存储路径

storeCheckpoint=/usr/local/rocketmq/store/checkpoint

#abort 文件存储路径

abortFile=/usr/local/rocketmq/store/abort

#限制的消息大小

maxMessageSize=65536

#flushCommitLogLeastPages=4

#flushConsumeQueueLeastPages=2

#flushCommitLogThoroughInterval=10000

#flushConsumeQueueThoroughInterval=60000

#Broker 的角色

#- ASYNC_MASTER 异步复制Master

#- SYNC_MASTER 同步双写Master

#- SLAVE

brokerRole=SLAVE

#刷盘方式

#- ASYNC_FLUSH 异步刷盘

#- SYNC_FLUSH 同步刷盘

flushDiskType=ASYNC_FLUSH

#checkTransactionMessageEnable=false

#发消息线程池数量

#sendMessageThreadPoolNums=128

#拉消息线程池数量

#pullMessageThreadPoolNums=128

启动集群

1)启动NameServe集群

cd /usr/local/rocketmq/bin

./mqnamesrv &

2)启动Broker集群

  1. master1:

cd /usr/local/rocketmq/bin

./mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a.properties &

  1. slave1

cd /usr/local/rocketmq/bin

./mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a-s.properties &

  1. master2

cd /usr/local/rocketmq/bin

./mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b.properties &

  1. slave2

cd /usr/local/rocketmq/bin

./mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b-s.properties &

集群监控平台搭建

  1. 概述: RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals,这个项目中有一个子模块叫rocketmq-console,这个便是管理控制台项目了,先将incubator-rocketmq-externals拉到本地,因为我们需要自己对rocketmq-console进行编译打包运行。
  2. 打包前在rocketmq-console中配置namesrv集群地址:

rocketmq.config.namesrvAddr=192.168.25.135:9876;192.168.25.138:9876

  1. 编译并打包

mvn clean package -Dmaven.test.skip=true

  1. 启动rocketmq-console:

java -jar rocketmq-console-ng-1.0.0.jar

  1. 访问http://localhost:8080

以上是 RocketMQ安装,双主双从同步集群搭建 的全部内容, 来源链接: utcz.com/a/23118.html

回到顶部