首页 > 其他 > 详细

RocketMQ（一）

时间：2020-07-07 22:43:33      阅读：55      评论：0      收藏：0      [点我收藏+]

1 作用、优点

  • 应用解耦

    如果某个子系统故障，不会影响其他子系统的运行。

  • 流量削峰

    如果请求流量猛增，系统负载超过阈值，不会压垮系统。

  • 消息异步

    加快上游系统响应速度；下游系统故障不会影响上游系统的运行；下游系统只需要订阅MQ。

2 缺点

  • 降低系统可用性

    一旦MQ宕机，就会对业务造成影响。

    如何保证MQ的高可用？

  • 提高系统复杂度

    以前系统间是同步调用，现在是通过MQ异步调用。

    如何保证消息没有被重复消费？怎么处理消息丢失情况？怎么保证消息传递的顺序性？

  • 一致性问题

    A系统处理完业务，通过MQ给B、C、D三个系统发消息，如果B、C处理成功，D处理失败。

    如何保证消息处理的一致性？

3 MQ产品比较

特性 ActiveMQ RabbitMQ RocketMQ Kafka
开发语言 java erlang java scala
单机吞吐量 万级 万级 10万级 10万级
时效性 ms级 us级 ms级 ms级以内
可用性 高（主从架构） 高（主从架构） 非常高（分布式架构） 非常高（分布式架构）
功能特性 成熟产品，较多文档，较好支持各种协议 基于erlang开发，所以并发能力强，性能极好，延时低；管理界面丰富 MQ功能比较完备，扩展性佳 只支持主要的MQ功能，比如没有消息查询、消息回溯等功能，在大数据领域应用广

4 下载

下载地址

源码地址

5 启动

5.1 启动NameServer

# 启动NameServer
nohup sh mqnamesrv &
# 查看启动日志
tail -f ~/logs/rocketmqlogs/namesrv.log

技术分享图片

技术分享图片

5.2 启动Broker

# 启动Broker，并连接上NameServer
nohup sh mqbroker -n localhost:9876 &
# 查看启动日志
tail -f ~/logs/rocketmqlogs/broker.log

Broker启动失败原因：RocketMQ默认的虚拟机内存较大，启动Broker可能因为内存不足导致失败。

# 修改默认JVM内存大小
vi runbroker.sh
vi runserver.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

技术分享图片

5.3 关闭RocketMQ

# 关闭nameserver
sh bin/mqshutdown namesrv
# 关闭broker
sh bin/mqshutdown broker

6 角色

  • Producer：消息的发送者
  • Consumer：消息的接收者
  • Broker：暂存和传输消息
  • NameServer：管理Broker
  • Topic：区分消息的种类
  • Message Queue：相当于Topic的分区；用于并行发送和接收消息

技术分享图片

7 集群搭建（介绍）

7.1 集群特点

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

7.2 集群模式

7.2.1 单Master

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

7.2.2 多Master

一个集群无Slave，全是Master。

  • 优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；
  • 缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。

7.2.3 多Master多Slave（异步）[1]

每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级）。

  • 优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎--样；
  • 缺点：Master宕机，磁盘损坏情况下会丢失少量消息。

7.2.4 多Master多Slave（同步）[2]

每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功。

  • 优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。
  • 缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。

8 双主双从集群搭建（实战）

8.1 总体架构

采用同步双写方式

技术分享图片

8.2 工作流程

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

8.3 服务器环境

序号 IP 架构模式
1 192.168.56.130 Master1、Slave2
2 192.168.56.131 Master2、Slave1

注：起4个虚拟机太卡；一组Master和Slave交叉放，防止一台服务器挂了，这组Master和Slave都用不了。

8.4 Host添加信息

vi /etc/hosts

域名和ip地址的映射，方便集群配置文件编写的时候，直接通过域名去访问节点，配置如下：

# nameserver
192.168.56.130 rocketmq-nameserver1
192.168.56.131 rocketmq-nameserver2
# broker
192.168.56.130 rocketmq-master1
192.168.56.130 rocketmq-slave2
192.168.56.131 rocketmq-master2
192.168.56.131 rocketmq-slave1

重启网卡：

systemctl restart network

8.5 防火墙配置

原因：宿主机需要远程访问虚拟机的RocketMQ服务，需要开放相关的端口号。

方法一：简单粗暴直接关闭防火墙

# 关闭防火墙
systemctl stop firewalld.service
# 查看防火墙的状态
firewall-cmd --state
# 禁止防火墙开机启动
systemctl disable firewalld.service

方法二：只开放特定的端口号

# 开放nameserver默认端口9876
firewall-cmd --remove-port=9876/tcp --permanent
# 开放master默认端口10911
firewall-cmd --remove-port=10911/tcp --permanent
# 开放salve默认端口11011（当前集群模式可不开启）
firewall-cmd --remove-port=11011/tcp --permanent
# 重启防火墙
firewall-cmd --reload

8.6 环境变量配置

目的：不用每次进到bin目录下执行命令

vi /etc/profile

在profile文件末尾加入：

# set rocketmq
ROCKETMQ_HOME=/home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release
PATH=$PATH:$ROCKETMQ_HOME/bin
export ROCKETMQ_HOME PATH

# 使配置立刻生效
source /etc/profile

8.7 创建消息存储路径

mkdir /home/lvyou/run/rocketmq/store
mkdir /home/lvyou/run/rocketmq/store/commitlog
mkdir /home/lvyou/run/rocketmq/store/consumequeue
mkdir /home/lvyou/run/rocketmq/store/index

mkdir /home/lvyou/run/rocketmq/store2
mkdir /home/lvyou/run/rocketmq/store2/commitlog
mkdir /home/lvyou/run/rocketmq/store2/consumequeue
mkdir /home/lvyou/run/rocketmq/store2/index

8.8 Broker配置文件

8.8.1 master1

服务器：192.1683.56.130

vi /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-a.properties

修改配置如下：

#所属集群名字
brokerClusterName=rocketmq-cluster
#指定broker的IP
brokerIP1=192.168.56.130
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0表示Master，>0表示Slave 
brokerId=0
#nameServer地址，分号分割
#namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
namesrvAddr=192.168.56.130:9876;192.168.56.133: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=/home/lvyou/run/rocketmq/store
#commitLog存储路径
storePathCommitLog=/home/lvyou/run/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/home/lvyou/run/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/home/lvyou/run/rocketmq/store/index
#checkpoint文件存储路径
storeCheckpoint=/home/lvyou/run/rocketmq/store/checkpoint
#abort文件存储路径
abortFile=/home/lvyou/run/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

8.8.2 slave2

服务器：192.1683.56.130

vi /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-b-s.properties

修改配置如下：

#所属集群名字
brokerClusterName=rocketmq-cluster
#指定broker的IP
brokerIP1=192.168.56.130
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0表示Master，>0表示Slave 
brokerId=1
#nameServer地址，分号分割
#namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
namesrvAddr=192.168.56.130:9876;192.168.56.133: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=/home/lvyou/run/rocketmq/store2
#commitLog存储路径
storePathCommitLog=/home/lvyou/run/rocketmq/store2/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/home/lvyou/run/rocketmq/store2/consumequeue
#消息索引存储路径
storePathIndex=/home/lvyou/run/rocketmq/store2/index
#checkpoint文件存储路径
storeCheckpoint=/home/lvyou/run/rocketmq/store2/checkpoint
#abort文件存储路径
abortFile=/home/lvyou/run/rocketmq/store2/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

8.8.3 master2

服务器：192.1683.56.133

vi /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-b.properties

修改配置如下：

#所属集群名字
brokerClusterName=rocketmq-cluster
#指定broker的IP
brokerIP1=192.168.56.133
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0表示Master，>0表示Slave 
brokerId=0
#nameServer地址，分号分割
#namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
namesrvAddr=192.168.56.130:9876;192.168.56.133: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=/home/lvyou/run/rocketmq/store
#commitLog存储路径
storePathCommitLog=/home/lvyou/run/rocketmq/store/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/home/lvyou/run/rocketmq/store/consumequeue
#消息索引存储路径
storePathIndex=/home/lvyou/run/rocketmq/store/index
#checkpoint文件存储路径
storeCheckpoint=/home/lvyou/run/rocketmq/store/checkpoint
#abort文件存储路径
abortFile=/home/lvyou/run/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

8.8.4 slave2

服务器：192.1683.56.133

vi /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-a-s.properties

修改配置如下：

#所属集群名字
brokerClusterName=rocketmq-cluster
#指定broker的IP
brokerIP1=192.168.56.133
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0表示Master，>0表示Slave 
brokerId=1
#nameServer地址，分号分割
#namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
namesrvAddr=192.168.56.130:9876;192.168.56.133: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=/home/lvyou/run/rocketmq/store2
#commitLog存储路径
storePathCommitLog=/home/lvyou/run/rocketmq/store2/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/home/lvyou/run/rocketmq/store2/consumequeue
#消息索引存储路径
storePathIndex=/home/lvyou/run/rocketmq/store2/index
#checkpoint文件存储路径
storeCheckpoint=/home/lvyou/run/rocketmq/store2/checkpoint
#abort文件存储路径
abortFile=/home/lvyou/run/rocketmq/store2/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

8.9 调整启动脚本内存大小

8.9.1 runbroker.sh

vi /bin/runbroker.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

8.9.2 runserver.sh

vi /bin/runserver.sh

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

8.10 服务启动

8.10.1 启动NameServer集群

分别在两台虚拟机上启动NameServer

nohup sh mqnamesrv &

8.10.2 启动Broker集群

  • 在192.168.56.130上启动master1和slave2

    master1：

    # -c是指定配置文件
nohup sh mqbroker -c /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-a.properties &

    slave2：

    nohup sh mqbroker -c /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-b-s.properties &

  • 在192.168.56.130上启动master2和slave1

    master2：

    nohup sh mqbroker -c /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-b.properties &

    slave1：

    nohup sh mqbroker -c /home/lvyou/run/rocketmq/rocketmq-all-4.7.1-bin-release/conf/2m-2s-sync/broker-a-s.properties &

9 mqadmin管理工具

9.1 使用方式

在bin目录下执行./mqadmin {command} {args}

9.2 命令

略

10 集群监控平台搭建

10.1 概述

RocketMQ有一个对其扩展的开源项目incubator-rocketmg-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmg-externals拉到本地，因为我们需要自己对rocketmq-console进行编译打包运行。

10.2 下载并编译打包

# 在本地执行
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true

在打包前在\rocketmq-console\src\main\resources\application.properties中配置namesrv集群地址：

rocketmq.config.namesrvAddr=192.168.56.130:9876;192.168.56.131:9876

将jar包放到任意虚拟机上，启动rocketmq-console：

java -jar rocketmq-console-ng-1.0.1.jar

浏览器访问 http://192.168.56.130:8080/ 进入控制台界面。

  1. producer发送消息到master，落盘后，直接给producer响应，数据异步地复制到slave ??

  2. producer发送消息到master，落盘后，数据同步地复制到slave，再给producer响应 ??

RocketMQ（一）

原文：https://www.cnblogs.com/LVY0U/p/13263520.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
分享档案
更多>
2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)
最新文章
更多>
教程昨日排行
更多>
友情链接
汇智网   PHP教程   插件网  
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣，分享程序人生！