RocketMQ系列：mqadmin命令行使用总览

1.介绍

rocketmq由consumer，producer，broker三大块组成，如何对这三类服务进行运维操作呢？这里介绍rocketmq自带的命令行工具：mqadmin

2.位置

项目：https://github.com/apache/rocketmq.git

${PWD}/rocketmq/distribution/bin/mqadmin

3.脚本

3.1 mqadmin

#这里ROCKETMQ_HOME对应的是rocketmq的主目录
if [ -z "$ROCKETMQ_HOME" ] ; then
## resolve links - $0 may be a link to maven's home
PRG="$0"

# need this for relative symlinks
while [ -h "$PRG" ] ; do
ls=`ls -ld "$PRG"`
link=`expr "$ls" : '.*-> (.*)$'`
if expr "$link" : '/.*' > /dev/null; then
PRG="$link"
else
PRG="`dirname "$PRG"`/$link"
fi
done

saveddir=`pwd`

ROCKETMQ_HOME=`dirname "$PRG"`/..

# make it fully qualified
ROCKETMQ_HOME=`cd "$ROCKETMQ_HOME" && pwd`

cd "$saveddir"
fi

export ROCKETMQ_HOME
#这里实际使用的另外一个脚本tools.sh
sh ${ROCKETMQ_HOME}/bin/tools.sh org.apache.rocketmq.tools.command.MQAdminStartup $@

3.2 tools.sh

主要作用就是设置JVM参数，然后使用java 启动对应classs

#===========================================================================================
# Java Environment Setting
#===========================================================================================
error_exit ()
{
    echo "ERROR: $1 !!"
    exit 1
}

[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME variable in your environment, We need java(x64)!"

export JAVA_HOME
export JAVA="$JAVA_HOME/bin/java"
export BASE_DIR=$(dirname $0)/..
export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}

#===========================================================================================
# JVM Configuration
#===========================================================================================
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m -XX:PermSize=128m -XX:MaxPermSize=128m"
JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${BASE_DIR}/lib:${JAVA_HOME}/jre/lib/ext"
JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"

#这里是实际调用
$JAVA ${JAVA_OPT} $@

3.3 MQAdminStartup.java

MQAdminStartup.java是个入口类，里面注册了如下的各类命令，如需详细了解细节可以直接进入${PWD}/rocketmq/tools/src/main/java/org/apache/rocketmq/tools/command目录下查看：

import org.apache.rocketmq.tools.command.broker.BrokerConsumeStatsSubCommad;
import org.apache.rocketmq.tools.command.broker.BrokerStatusSubCommand;
import org.apache.rocketmq.tools.command.broker.CleanExpiredCQSubCommand;
import org.apache.rocketmq.tools.command.broker.CleanUnusedTopicCommand;
import org.apache.rocketmq.tools.command.broker.GetBrokerConfigCommand;
import org.apache.rocketmq.tools.command.broker.SendMsgStatusCommand;
import org.apache.rocketmq.tools.command.broker.UpdateBrokerConfigSubCommand;
import org.apache.rocketmq.tools.command.cluster.CLusterSendMsgRTCommand;
import org.apache.rocketmq.tools.command.cluster.ClusterListSubCommand;
import org.apache.rocketmq.tools.command.connection.ConsumerConnectionSubCommand;
import org.apache.rocketmq.tools.command.connection.ProducerConnectionSubCommand;
import org.apache.rocketmq.tools.command.consumer.ConsumerProgressSubCommand;
import org.apache.rocketmq.tools.command.consumer.ConsumerStatusSubCommand;
import org.apache.rocketmq.tools.command.consumer.DeleteSubscriptionGroupCommand;
import org.apache.rocketmq.tools.command.consumer.StartMonitoringSubCommand;
import org.apache.rocketmq.tools.command.consumer.UpdateSubGroupSubCommand;
import org.apache.rocketmq.tools.command.message.CheckMsgSendRTCommand;
import org.apache.rocketmq.tools.command.message.ConsumeMessageCommand;
import org.apache.rocketmq.tools.command.message.PrintMessageByQueueCommand;
import org.apache.rocketmq.tools.command.message.PrintMessageSubCommand;
import org.apache.rocketmq.tools.command.message.QueryMsgByIdSubCommand;
import org.apache.rocketmq.tools.command.message.QueryMsgByKeySubCommand;
import org.apache.rocketmq.tools.command.message.QueryMsgByOffsetSubCommand;
import org.apache.rocketmq.tools.command.message.QueryMsgByUniqueKeySubCommand;
import org.apache.rocketmq.tools.command.message.SendMessageCommand;
import org.apache.rocketmq.tools.command.namesrv.DeleteKvConfigCommand;
import org.apache.rocketmq.tools.command.namesrv.GetNamesrvConfigCommand;
import org.apache.rocketmq.tools.command.namesrv.UpdateKvConfigCommand;
import org.apache.rocketmq.tools.command.namesrv.UpdateNamesrvConfigCommand;
import org.apache.rocketmq.tools.command.namesrv.WipeWritePermSubCommand;
import org.apache.rocketmq.tools.command.offset.CloneGroupOffsetCommand;
import org.apache.rocketmq.tools.command.offset.ResetOffsetByTimeCommand;
import org.apache.rocketmq.tools.command.queue.QueryConsumeQueueCommand;
import org.apache.rocketmq.tools.command.stats.StatsAllSubCommand;
import org.apache.rocketmq.tools.command.topic.AllocateMQSubCommand;
import org.apache.rocketmq.tools.command.topic.DeleteTopicSubCommand;
import org.apache.rocketmq.tools.command.topic.TopicClusterSubCommand;
import org.apache.rocketmq.tools.command.topic.TopicListSubCommand;
import org.apache.rocketmq.tools.command.topic.TopicRouteSubCommand;
import org.apache.rocketmq.tools.command.topic.TopicStatusSubCommand;
import org.apache.rocketmq.tools.command.topic.UpdateOrderConfCommand;
import org.apache.rocketmq.tools.command.topic.UpdateTopicPermSubCommand;
import org.apache.rocketmq.tools.command.topic.UpdateTopicSubCommand;

3.4 命令详解

整体来说，命令分为如下几大类：

1. 集群相关：

• clusterList：查看集群列表 
• clusterRT：测试集群的响应耗时

2. broker相关

• updateBrokerConfig ：更新broker的配置
• brokerStatus ：获取broker的运行时状态数据
• wipeWritePerm：设置某broker为只读
• getBrokerConfig：获取broker的配置信息

3.topic相关：

• updateTopic：创建或更新topic
• deleteTopic ：从nameserver和broker中删除topic信息
• updateTopic：更新topic的perm信息
• topicRoute：获取topic的路由信息
• topicStatus： 获取topic的当前状态信息
• topicClusterList：获取topic对应的集群信息
• topicList：查看集群中的topic列表
• statsAll: 查看所有topic已经对应的consumer的消费进度
• cleanUnusedTopic：清理集群中无用的topic

4.message相关：

• queryMsgById: 通过message的id查询消息
• queryMsgByKey：通过message的key查询消息
• queryMsgByUniqueKey：通过message的UniqueKey查询消息
• queryMsgByOffset：通过偏移查询message 、
• printMsg：打印某条message的详情
• printMsgByQueue：打印某个queue（队列）里的消息的详情
• sendMsgStatus：向broker发送消息
• sendMessage：发送一条消息
• producerConnection：查询producer的信息(socket连接，客户端版本） 

5. 消费相关:

• consumerConnection ：查询consumer的信息（socker连接，客户端版本，消费组）
• consumerProgress ：查询consumer的消费进度，tps
• consumerStatus ：查询consumer的的内部数
• brokerConsumeStats：查看所有topic对应的消费数据（broker的offset，consumer的offset，是否有diff）  
• cleanExpiredCQ：清理集群中无用的消费组cloneGroupOffset clone offset from other group.
• cloneGroupOffset ：克隆指定topic下某个消费组的消费进度到指定消费组
• resetOffsetByTime：设置consumer的offset到某个时间点
• checkMsgSendRT：测试消费发生的响应耗时
• consumeMessage：消费消息
• queryCq：查询consumer指定队列和索引位置的消费信息
• updateSubGroup：创建或者更新消费组（订阅组）
• deleteSubGroup：从broker中删除消费组（订阅组）

6. nameserver相关:

• updateKvConfig：更新nameserver的kv配置
• deleteKvConfig：删除nameserver的kv配置      
• getNamesrvConfig：获取nameserver的配置
• updateNamesrvConfig：更新nameserver的配置

7. acl相关：

• updateAclConfig：更新acl配置   
• deleteAccessConfig：删除acl配置的账户  
• clusterAclConfigVersion：查看集群的acl配置版本
• updateGlobalWhiteAddr：更新acl里面的白名单 
• getAccessConfigSubCommand：查看acl的配置信息

8. 其他：

• updateOrderConf Create or update or delete order conf
• startMonitoring Start Monitoring
• allocateMQ Allocate MQ

博主：测试生财（一个不为996而996的测开码农）

座右铭：专注测试开发与自动化运维，努力读书思考写作，为内卷的人生奠定财务自由。

内容范畴：技术提升，职场杂谈，事业发展，阅读写作，投资理财，健康人生。

csdn：https://blog.csdn.net/ccgshigao

博客园：https://www.cnblogs.com/qa-freeroad/

51cto：https://blog.51cto.com/14900374

微信公众号：测试生财（定期分享独家内容和资源）