zoukankan      html  css  js  c++  java

  • RocketMQ集群搭建

    1. RocketMQ集群搭建

    1.1 各角色介绍

    • Producer：消息的发送者；举例：发信者
    • Consumer：消息接收者；举例：收信者
    • Broker：暂存和传输消息；举例：邮局
    • NameServer：管理Broker；举例：各个邮局的管理机构
    • Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息
    • Message Queue：相当于是Topic的分区；用于并行发送和接收消息

    1.2 集群搭建方式

    1.2.1 集群特点

    • NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

    • Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为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配置决定。

    1.2.2 集群模式

    1）单Master模式

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

    2）多Master模式

    一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：

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

    3）多Master多Slave模式（异步）

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

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

    4）多Master多Slave模式（同步）

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

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

    1.3 双主双从集群搭建

    1.3.1 总体架构

    消息高可用采用2m-2s（同步双写）方式

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

    1.3.3 服务器环境

    序号 IP 角色 架构模式
    1 192.168.1.144 nameserver、brokerserver Master1、Slave2
    2 192.168.1.145 nameserver、brokerserver Master2、Slave1

    1.3.4 Host添加信息

    vim /etc/hosts

    配置如下:

    # nameserver
192.168.1.144 rocketmq-nameserver1
192.168.1.145 rocketmq-nameserver2
# broker
192.168.1.144 rocketmq-master1
192.168.1.144 rocketmq-slave2
192.168.1.145 rocketmq-master2
192.168.1.145 rocketmq-slave1

    配置完成后, 重启网卡

    systemctl restart network

    1.3.5 防火墙配置

    宿主机需要远程访问虚拟机的rocketmq服务和web服务，需要开放相关的端口号，简单粗暴的方式是直接关闭防火墙

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

    或者为了安全，只开放特定的端口号，RocketMQ默认使用3个端口：9876 、10911 、11011 。如果防火墙没有关闭的话，那么防火墙就必须开放这些端口：

    • nameserver 默认使用 9876 端口
    • master 默认使用 10911 端口
    • slave 默认使用11011 端口

    执行以下命令：

    # 开放name server默认端口
firewall-cmd --remove-port=9876/tcp --permanent
# 开放master默认端口
firewall-cmd --remove-port=10911/tcp --permanent
# 开放slave默认端口 (当前集群模式可不开启)
firewall-cmd --remove-port=11011/tcp --permanent 
# 重启防火墙
firewall-cmd --reload

    1.3.6 环境变量配置

    vim /etc/profile

    在profile文件的末尾加入如下命令

    #set rocketmq
ROCKETMQ_HOME=/root/rocketmq-all-4.4.0-bin-release
PATH=$PATH:$ROCKETMQ_HOME/bin
export ROCKETMQ_HOME PATH

    输入:wq! 保存并退出， 并使得配置立刻生效：

    source /etc/profile

    1.3.7 创建消息存储路径

    mkdir -p /usr/local/rocketmq/store
mkdir -p /usr/local/rocketmq/store/commitlog
mkdir -p /usr/local/rocketmq/store/consumequeue
mkdir -p /usr/local/rocketmq/store/index
mkdir -p /usr/local/rocketmq/store1
mkdir -p /usr/local/rocketmq/store1/commitlog
mkdir -p /usr/local/rocketmq/store1/consumequeue
mkdir -p /usr/local/rocketmq/store1/index

    1.3.8 broker配置文件

    1）master1

    服务器：192.168.1.144

    vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties

    修改配置如下：

    #所属集群名字
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

    2）slave2

    服务器：192.168.1.144

    vi /usr/soft/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/store1
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store1/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store1/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store1/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store1/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store1/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

    3）master2

    服务器：192.168.1.145

    vi /usr/soft/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

    4）slave1

    服务器：192.168.1.145

    vi /usr/soft/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/store1
#commitLog 存储路径
storePathCommitLog=/usr/local/rocketmq/store1/commitlog
#消费队列存储路径存储路径
storePathConsumeQueue=/usr/local/rocketmq/store1/consumequeue
#消息索引存储路径
storePathIndex=/usr/local/rocketmq/store1/index
#checkpoint 文件存储路径
storeCheckpoint=/usr/local/rocketmq/store1/checkpoint
#abort 文件存储路径
abortFile=/usr/local/rocketmq/store1/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.3.9 修改启动脚本文件

    1）runbroker.sh

    vi /usr/local/rocketmq/bin/runbroker.sh

    需要根据内存大小进行适当的对JVM参数进行调整：

    #===================================================
# 开发环境配置 JVM Configuration
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

    2）runserver.sh

    vim /usr/local/rocketmq/bin/runserver.sh

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

    1.3.10 服务启动

    1）启动NameServe集群

    分别在192.168.1.144和192.168.1.145启动NameServer

    cd /usr/local/rocketmq/bin
nohup sh mqnamesrv &

    2）启动Broker集群

    • 在192.168.1.144上启动master1和slave2

    master1：

    cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties &

    slave2：

    cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/soft/rocketmq/conf/2m-2s-sync/broker-b-s.properties &
    • 在192.168.1.145上启动master2和slave2

    master2

    cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/soft/rocketmq/conf/2m-2s-sync/broker-b.properties &

    slave1

    cd /usr/local/rocketmq/bin
nohup sh mqbroker -c /usr/soft/rocketmq/conf/2m-2s-sync/broker-a-s.properties &

    1.3.11 查看进程状态

    启动后通过JPS查看启动进程

    image-20201130173135612

    1.3.12 查看日志

    # 查看nameServer日志
tail -500f ~/logs/rocketmqlogs/namesrv.log
# 查看broker日志
tail -500f ~/logs/rocketmqlogs/broker.log

    1.4 mqadmin管理工具

    1.4.1 使用方式

    进入RocketMQ安装位置，在bin目录下执行./mqadmin {command} {args}

    1.4.2 命令介绍

    1）Topic相关

    名称 含义 命令选项 说明
    updateTopic 创建更新Topic配置 -b Broker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port
    -c cluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询）
    -h- 打印帮助
    -n NameServer服务地址，格式 ip:port
    -p 指定新topic的读写权限( W=2|R=4|WR=6 )
    -r 可读队列数（默认为 8）
    -w 可写队列数（默认为 8）
    -t topic 名称（名称只能使用字符 ^[a-zA-Z0-9_-]+$ ）
    deleteTopic 删除Topic -c cluster 名称，表示删除某集群下的某个 topic （集群 可通过 clusterList 查询）
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -t topic 名称（名称只能使用字符 ^[a-zA-Z0-9_-]+$ ）
    topicList 查看 Topic 列表信息 -h 打印帮助
    -c 不配置-c只返回topic列表，增加-c返回clusterName, topic, consumerGroup信息，即topic的所属集群和订阅关系，没有参数
    -n NameServer 服务地址，格式 ip:port
    topicRoute 查看 Topic 路由信息 -t topic 名称
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    topicStatus 查看 Topic 消息队列offset -t topic 名称
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    topicClusterList 查看 Topic 所在集群列表 -t topic 名称
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    updateTopicPerm 更新 Topic 读写权限 -t topic 名称
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -b Broker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port
    -p 指定新 topic 的读写权限( W=2|R=4|WR=6 )
    -c cluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询），-b优先，如果没有-b，则对集群中所有Broker执行命令
    updateOrderConf 从NameServer上创建、删除、获取特定命名空间的kv配置，目前还未启用 -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -t topic，键
    -v orderConf，值
    -m method，可选get、put、delete
    allocateMQ 以平均负载算法计算消费者列表负载消息队列的负载结果 -t topic 名称
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -i ipList，用逗号分隔，计算这些ip去负载Topic的消息队列
    statsAll 打印Topic订阅关系、TPS、积累量、24h读写总量等信息 -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -a 是否只打印活跃topic
    -t 指定topic
    #### 2）集群相关
    名称 含义 命令选项 说明
    clusterList 查看集群信息，集群、BrokerName、BrokerId、TPS等信息 -m 打印更多信息 (增加打印出如下信息 #InTotalYest, #OutTotalYest, #InTotalToday ,#OutTotalToday)
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -i 打印间隔，单位秒
    clusterRT 发送消息检测集群各Broker RT。消息发往${BrokerName} Topic。 -a amount，每次探测的总数，RT = 总时间 / amount
    -s 消息大小，单位B
    -c 探测哪个集群
    -p 是否打印格式化日志，以|分割，默认不打印
    -h 打印帮助
    -m 所属机房，打印使用
    -i 发送间隔，单位秒
    -n NameServer 服务地址，格式 ip:port

    3）Broker相关

    名称 含义 命令选项 说明
    updateBrokerConfig 更新 Broker 配置文件，会修改Broker.conf -b Broker 地址，格式为ip:port
    -c cluster 名称
    -k key 值
    -v value 值
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    brokerStatus 查看 Broker 统计信息、运行状态（你想要的信息几乎都在里面） -b Broker 地址，地址为ip:port
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    brokerConsumeStats Broker中各个消费者的消费情况，按Message Queue维度返回Consume Offset，Broker Offset，Diff，TImestamp等信息 -b Broker 地址，地址为ip:port
    -t 请求超时时间
    -l diff阈值，超过阈值才打印
    -o 是否为顺序topic，一般为false
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    getBrokerConfig 获取Broker配置 -b Broker 地址，地址为ip:port
    -n NameServer 服务地址，格式 ip:port
    wipeWritePerm 从NameServer上清除 Broker写权限 -b Broker 地址，地址为ip:port
    -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    cleanExpiredCQ 清理Broker上过期的Consume Queue，如果手动减少对列数可能产生过期队列 -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    -b Broker 地址，地址为ip:port
    -c 集群名称
    cleanUnusedTopic 清理Broker上不使用的Topic，从内存中释放Topic的Consume Queue，如果手动删除Topic会产生不使用的Topic -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    -b Broker 地址，地址为ip:port
    -c 集群名称
    sendMsgStatus 向Broker发消息，返回发送状态和RT -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    -b BrokerName，注意不同于Broker地址
    -s 消息大小，单位B
    -c 发送次数
    #### 4）消息相关
    名称 含义 命令选项 说明
    queryMsgById 根据offsetMsgId查询msg，如果使用开源控制台，应使用offsetMsgId，此命令还有其他参数，具体作用请阅读QueryMsgByIdSubCommand。 -i msgId
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    queryMsgByKey 根据消息 Key 查询消息 -k msgKey
    -t Topic 名称
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    queryMsgByOffset 根据 Offset 查询消息 -b Broker 名称，（这里需要注意 填写的是 Broker 的名称，不是 Broker 的地址，Broker 名称可以在 clusterList 查到）
    -i query 队列 id
    -o offset 值
    -t topic 名称
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    queryMsgByUniqueKey 根据msgId查询，msgId不同于offsetMsgId，区别详见常见运维问题。-g，-d配合使用，查到消息后尝试让特定的消费者消费消息并返回消费结果 -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -i uniqe msg id
    -g consumerGroup
    -d clientId
    -t topic名称
    checkMsgSendRT 检测向topic发消息的RT，功能类似clusterRT -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -t topic名称
    -a 探测次数
    -s 消息大小
    sendMessage 发送一条消息，可以根据配置发往特定Message Queue，或普通发送。 -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -t topic名称
    -p body，消息体
    -k keys
    -c tags
    -b BrokerName
    -i queueId
    consumeMessage 消费消息。可以根据offset、开始&结束时间戳、消息队列消费消息，配置不同执行不同消费逻辑，详见ConsumeMessageCommand。 -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -t topic名称
    -b BrokerName
    -o 从offset开始消费
    -i queueId
    -g 消费者分组
    -s 开始时间戳，格式详见-h
    -d 结束时间戳
    -c 消费多少条消息
    printMsg 从Broker消费消息并打印，可选时间段 -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -t topic名称
    -c 字符集，例如UTF-8
    -s subExpress，过滤表达式
    -b 开始时间戳，格式参见-h
    -e 结束时间戳
    -d 是否打印消息体
    printMsgByQueue 类似printMsg，但指定Message Queue -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -t topic名称
    -i queueId
    -a BrokerName
    -c 字符集，例如UTF-8
    -s subExpress，过滤表达式
    -b 开始时间戳，格式参见-h
    -e 结束时间戳
    -p 是否打印消息
    -d 是否打印消息体
    -f 是否统计tag数量并打印
    resetOffsetByTime 按时间戳重置offset，Broker和consumer都会重置 -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -g 消费者分组
    -t topic名称
    -s 重置为此时间戳对应的offset
    -f 是否强制重置，如果false，只支持回溯offset，如果true，不管时间戳对应offset与consumeOffset关系
    -c 是否重置c++客户端offset

    5）消费者、消费组相关

    名称 含义 命令选项 说明
    consumerProgress 查看订阅组消费状态，可以查看具体的client IP的消息积累量 -g 消费者所属组名
    -s 是否打印client IP
    -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    consumerStatus 查看消费者状态，包括同一个分组中是否都是相同的订阅，分析Process Queue是否堆积，返回消费者jstack结果，内容较多，使用者参见ConsumerStatusSubCommand -h 打印帮助
    -n NameServer 服务地址，格式 ip:port
    -g consumer group
    -i clientId
    -s 是否执行jstack
    getConsumerStatus 获取 Consumer 消费进度 -g 消费者所属组名
    -t 查询主题
    -i Consumer 客户端 ip
    -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    updateSubGroup 更新或创建订阅关系 -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    -b Broker地址
    -c 集群名称
    -g 消费者分组名称
    -s 分组是否允许消费
    -m 是否从最小offset开始消费
    -d 是否是广播模式
    -q 重试队列数量
    -r 最大重试次数
    -i 当slaveReadEnable开启时有效，且还未达到从slave消费时建议从哪个BrokerId消费，可以配置备机id，主动从备机消费
    -w 如果Broker建议从slave消费，配置决定从哪个slave消费，配置BrokerId，例如1
    -a 当消费者数量变化时是否通知其他消费者负载均衡
    deleteSubGroup 从Broker删除订阅关系 -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    -b Broker地址
    -c 集群名称
    -g 消费者分组名称
    cloneGroupOffset 在目标群组中使用源群组的offset -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    -s 源消费者组
    -d 目标消费者组
    -t topic名称
    -o 暂未使用

    6）连接相关

    名称 含义 命令选项 说明
    consumerConnec tion 查询 Consumer 的网络连接 -g 消费者所属组名
    -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    producerConnec tion 查询 Producer 的网络连接 -g 生产者所属组名
    -t 主题名称
    -n NameServer 服务地址，格式 ip:port
    -h 打印帮助

    7）NameServer相关

    名称 含义 命令选项 说明
    updateKvConfig 更新NameServer的kv配置，目前还未使用 -s 命名空间
    -k key
    -v value
    -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    deleteKvConfig 删除NameServer的kv配置 -s 命名空间
    -k key
    -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    getNamesrvConfig 获取NameServer配置 -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    updateNamesrvConfig 修改NameServer配置 -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    -k key
    -v value

    8）其他

    名称 含义 命令选项 说明
    startMonitoring 开启监控进程，监控消息误删、重试队列消息数等 -n NameServer 服务地址，格式 ip:port
    -h 打印帮助
    ### 1.4.3 注意事项
    • 几乎所有命令都需要配置-n表示NameServer地址，格式为ip:port
    • 几乎所有命令都可以通过-h获取帮助
    • 如果既有Broker地址（-b）配置项又有clusterName（-c）配置项，则优先以Broker地址执行命令；如果不配置Broker地址，则对集群中所有主机执行命令

    1.5 集群监控平台搭建

    1.5.1 概述

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

    image-20201130183452949

    1.5.2 下载并编译打包

    git clone https://github.com/apache/rocketmq-externals
cd rocketmq-console
mvn clean package -Dmaven.test.skip=true

    注意：打包前在rocketmq-console中配置namesrv集群地址：

    rocketmq.config.namesrvAddr=192.168.1.144:9876;192.168.1.145:9876

    启动rocketmq-console：

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

    启动成功后，我们就可以通过浏览器访问http://192.168.1.144:8080进入控制台界面了，如下图：

    image-20201130183119374

    集群状态：

    image-20201130183144378
  • 相关阅读:
    IntelliJ IDEA 添加本地xsd文件
    Dubbox离线约束地址
    IDEA发布运行web项目（曾经遇到的项目启动报404）
    项目导入时报错：The import javax.servlet.http.HttpServletRequest cannot be resolved
    IDEA使用maven中tomcat插件来启动服务器配置
    NetCore实践篇：分布式监控客户端ZipkinTracer从入门到放弃之路
    .Net外包篇:我是怎么看待外包的(二)
    .Net外包篇：我是如何看待外包的
    .Net架构篇:思考如何设计一款实用的分布式监控系统？
    分布式Redis缓存串讲(一)
  • 原文地址：https://www.cnblogs.com/dalianpai/p/14062867.html
Copyright © 2011-2022 走看看