# RocketMQ Java Client

**Repository Path**: wxhntmy/rocket-mq-java-client

## Basic Information

- **Project Name**: RocketMQ Java Client
- **Description**: RocketMQ Client 5.3.1
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2025-01-03
- **Last Updated**: 2025-08-09

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# RocketMQ 5.3.1
## CentOS 7 安装 RocketMQ
1. 安装 JDK
```shell
yum install java-1.8.0-openjdk* -y
```
2. 下载 Apache RocketMQ
```shell
wget -O rocketmq-all-5.3.1-bin-release.zip https://mirrors.cloud.tencent.com/apache/rocketmq/5.3.1/rocketmq-all-5.3.1-bin-release.zip
```
3. 启动 NameServer
```shell
cd rocketmq-all-5.3.1-bin-release/
nohup bin/mqnamesrv >> mqnamesrv.log 2>&1 &
```
4. 启动 Broker+Proxy
```shell
cd rocketmq-all-5.3.1-bin-release/
nohup bin/mqbroker -n localhost:9876 --enable-proxy >> mqbroker.log 2>&1 &
```
5. 启动 RocketMQ DashBoard
```shell
docker pull apacherocketmq/rocketmq-dashboard:latest
docker run -dt --name rocketmq-dashboard -e "JAVA_OPTS=-Drocketmq.namesrv.addr=localhost:9876 -Dserver.servlet.context-path=/romqdashboard" -p 8080:8080 apacherocketmq/rocketmq-dashboard:latest
```

## 快捷部署 RocketMQ Shell
```shell
#!/bin/bash

#安装位置
RocketMQ_DIR=/data/rocketmq
#日志文件
NAMESRV_RUN_FILE=$RocketMQ_DIR/logs/namesrv_run.log
#日志文件
BROKER_RUN_FILE=$RocketMQ_DIR/logs/broker_run.log
#RocketMQ NameServer地址
IP_ADDR=localhost


# RocketMQ的用户名
RocketMQ_USER=rocketmq
# Rocket NameServer端口
RocketMQ_NameServer_PORT=9876

# broker端口：10909,10911,10912
# proxy端口：8080,8081
RocketMQ_Broker_PORT=10911

# RocketMQ DashBoard端口
RocketMQ_DashBoard_PORT=9880

check_err()
{
    if [ $? -ne 0 ]; then
        echo "$1"
        exit 1
	else
	    echo "$2"
    fi
}

start() {
	
	netstat -tln | grep ":$RocketMQ_NameServer_PORT"
	CHECK_RESULT=$?
	if [ $CHECK_RESULT -eq 0 ];then
	  echo "RocketMQ NameServer 已启动，请勿重复启动应用！"
	  exit 1
	fi
	echo "正在启动 RocketMQ NameServer....."
	
	cat /etc/group | grep $RocketMQ_USER
	if [ $? -ne 0 ]; then
		groupadd $RocketMQ_USER
	fi
	cat /etc/passwd | grep $RocketMQ_USER
	if [ $? -ne 0 ]; then
		useradd $RocketMQ_USER -d /home/$RocketMQ_USER -g $RocketMQ_USER
	fi

    mkdir -p $RocketMQ_DIR/logs
    chown -R $RocketMQ_USER:$RocketMQ_USER $RocketMQ_DIR
	
	[ ! -f $NAMESRV_RUN_FILE ] && touch $NAMESRV_RUN_FILE
	chown -R $RocketMQ_USER:$RocketMQ_USER $NAMESRV_RUN_FILE
	[ ! -f $BROKER_RUN_FILE ] && touch $BROKER_RUN_FILE
	chown -R $RocketMQ_USER:$RocketMQ_USER $BROKER_RUN_FILE
	
	chmod +x $RocketMQ_DIR/bin/mqnamesrv
	chmod +x $RocketMQ_DIR/bin/mqshutdown
	chmod +x $RocketMQ_DIR/bin/mqbroker
	
	# 切换用户
su - $RocketMQ_USER << EOF
    # 不加反斜杠，取的值是切换用户前的
    nohup $RocketMQ_DIR/bin/mqnamesrv >> $NAMESRV_RUN_FILE 2>&1 &
    CHECK_RESULT=1
    CHECK_COUNT=60
    while [ \$CHECK_RESULT -ne 0 ];do
        if [ \$CHECK_COUNT -gt 0 ];then
          echo "等待监听端口启动..."
          sleep 1
          netstat -tln | grep ":$RocketMQ_NameServer_PORT"
          CHECK_RESULT=\$?
          CHECK_COUNT=\$((\$CHECK_COUNT-1))
        else
          echo "ERROR: 启动 $RocketMQ_NameServer_PORT 端口超时，请检查!"
          exit 1
        fi
    done
echo "RocketMQ NameServer启动成功!"
EOF

}

start_broker(){
	netstat -tln | grep ":$RocketMQ_NameServer_PORT"
	check_err "RocketMQ NameServer 未启动！" "RocketMQ NameServer 正在运行，开始启动 Broker + Proxy ....."

	
su - $RocketMQ_USER << EOF
    # 不加反斜杠，取的值是切换用户前的
    nohup $RocketMQ_DIR/bin/mqbroker -n $IP_ADDR:$RocketMQ_NameServer_PORT --enable-proxy >> $BROKER_RUN_FILE 2>&1 &
    CHECK_RESULT=1
    CHECK_COUNT=60
    while [ \$CHECK_RESULT -ne 0 ];do
        if [ \$CHECK_COUNT -gt 0 ];then
          echo "等待监听端口启动..."
          sleep 1
          netstat -tln | grep ":$RocketMQ_Broker_PORT"
          CHECK_RESULT=\$?
          CHECK_COUNT=\$((\$CHECK_COUNT-1))
        else
          echo "ERROR: 启动 $RocketMQ_Broker_PORT 端口超时，请检查!"
          exit 1
        fi
    done
echo "RocketMQ Broker + Proxy 启动成功!"
EOF

}

stop(){

	netstat -tln | grep ":$RocketMQ_NameServer_PORT"
	check_err "RocketMQ NameServer 未启动！" "RocketMQ NameServer 正在运行，开始停止应用....."

	su - $RocketMQ_USER -c "$RocketMQ_DIR/bin/mqshutdown namesrv"
	CHECK_RESULT=0
	CHECK_COUNT=30
	while [ $CHECK_RESULT -eq 0 ];do
	  if [ $CHECK_COUNT -gt 0 ];then
		echo "等待监听端口关闭..."
		sleep 1
		netstat -tln | grep ":$RocketMQ_NameServer_PORT"
		CHECK_RESULT=$?
		CHECK_COUNT=$(($CHECK_COUNT-1))
	  else
		echo "ERROR: 关闭 $RocketMQ_NameServer_PORT 端口超时，请检查!"
		exit 1
	  fi
	done
	echo "关闭 RocketMQ NameServer 成功!"
}

stop_broker(){

	netstat -tln | grep ":$RocketMQ_Broker_PORT"
	check_err "RocketMQ Broker + Proxy 未启动！" "RocketMQ Broker + Proxy 正在运行，开始停止应用....."

	su - $RocketMQ_USER -c "$RocketMQ_DIR/bin/mqshutdown broker"
	CHECK_RESULT=0
	CHECK_COUNT=30
	while [ $CHECK_RESULT -eq 0 ];do
	  if [ $CHECK_COUNT -gt 0 ];then
		echo "等待监听端口关闭..."
		sleep 1
		netstat -tln | grep ":$RocketMQ_Broker_PORT"
		CHECK_RESULT=$?
		CHECK_COUNT=$(($CHECK_COUNT-1))
	  else
		echo "ERROR: 关闭 $RocketMQ_Broker_PORT 端口超时，请检查!"
		exit 1
	  fi
	done
	echo "关闭 RocketMQ Broker + Proxy 成功!"
}

restart(){
	stop
	start
}
restart_broker(){
	stop_broker
	echo "停止 Broker + Proxy 后需要等待一段时间才能启动 Broker + Proxy"
	CHECK_COUNT=10
	while [ $CHECK_COUNT -gt 0 ];do
		echo "$CHECK_COUNT"
		sleep 1
		CHECK_COUNT=$(($CHECK_COUNT-1))
	done
	start_broker
}

status(){
	su $RocketMQ_USER -c "ps aux | grep -v \"grep\" | grep \"java\" | grep NamesrvStartup"
}

status_broker(){
	su $RocketMQ_USER -c "ps aux | grep -v \"grep\" | grep \"java\" | grep ProxyStartup"
}

remove(){
	if [ -d $RocketMQ_DIR ]; then
		DATE_TIME=`date "+%Y-%m-%d_%H-%M-%S"`
		mv -f $RocketMQ_DIR $RocketMQ_DIR.$DATE_TIME.bak
	fi
#	if [ -f $NAMESRV_RUN_FILE ]; then
#		DATE_TIME=`date "+%Y-%m-%d_%H-%M-%S"`
#		mv -f $NAMESRV_RUN_FILE $NAMESRV_RUN_FILE.$DATE_TIME.bak
#	fi
#	if [ -f $BROKER_RUN_FILE ]; then
#		DATE_TIME=`date "+%Y-%m-%d_%H-%M-%S"`
#		mv -f $BROKER_RUN_FILE $BROKER_RUN_FILE.$DATE_TIME.bak
#	fi
	
	userdel -r $RocketMQ_USER
}

start_dashboard(){
	netstat -tln | grep ":$RocketMQ_DashBoard_PORT"
	if [ $? -eq 0 ];then
		echo "端口：$RocketMQ_DashBoard_PORT 已被占用，请使用其他端口！"
		exit 1
	fi
	
	OS_NAME=`hostnamectl | grep "Operating System" | awk -F ": " '{print $2}'`
	mkdir -p $RocketMQ_DIR/rocketmq-dashboard
	if [ ! -f $RocketMQ_DIR/rocketmq-dashboard/users.properties ]; then
		touch $RocketMQ_DIR/rocketmq-dashboard/users.properties
cat << EOF > $RocketMQ_DIR/rocketmq-dashboard/users.properties
# 该文件支持热修改，即添加和修改用户时，不需要重新启动console
# 格式， 每行定义一个用户， username=password[,N]  #N是可选项，可以为0 (普通用户)； 1 （管理员）  

#定义管理员 
admin=123456,1

#定义普通用户
wxhntmy=123456,0
EOF
	fi
	if [ ! -f $RocketMQ_DIR/rocketmq-dashboard/role-permission.yml ]; then
		touch $RocketMQ_DIR/rocketmq-dashboard/role-permission.yml
cat << EOF > $RocketMQ_DIR/rocketmq-dashboard/role-permission.yml
# 该文件支持热修改，即添加和修改用户时，不需要重新启动console
# 格式，如果增加和删除接口权限，直接在列表中增加和删除接口地址即可。
# 接口路径配置支持通配符
# * 表示匹配0或多个不是/的字符
# ** 表示匹配0或多个任意字符
# ? 表示匹配1个任意字符

rolePerms:
  # 普通用户
  ordinary:
    - /rocketmq/nsaddr
    - /ops/*
    - /dashboard/**
    - /topic/*.query
    - /topic/sendTopicMessage.do
    - /producer/*.query
    - /message/*
    - /messageTrace/*
    - /monitor/*
EOF
	fi
	
	docker ps -a | grep "rocketmq-dashboard"
	if [ $? -ne 0 ]; then
		type docker &> /dev/null
		if [ $? -ne 0 ]; then
			yum install docker -y
			check_err "\033[31myum安装依赖包 docker 失败，请检查\033[0m" "\033[36myum安装依赖包 docker 成功\033[0m"
		fi
		#docker pull apacherocketmq/rocketmq-dashboard:latest
		docker run -dt --name rocketmq-dashboard -e "JAVA_OPTS=-Drocketmq.config.loginRequired=true -Drocketmq.config.dataPath=/tmp/rocketmq-console/data -Drocketmq.namesrv.addr=$IP_ADDR:$RocketMQ_NameServer_PORT -Dserver.servlet.context-path=/romqdashboard" -p $RocketMQ_DashBoard_PORT:8080 -v $RocketMQ_DIR/rocketmq-dashboard:/tmp/rocketmq-console/data apacherocketmq/rocketmq-dashboard:latest
	else
		docker restart rocketmq-dashboard
	fi
}

stop_dashboard(){
	docker stop rocketmq-dashboard
}
restart_dashboard(){
	docker restart rocketmq-dashboard
}
status_dashboard(){
	docker ps -a | grep "rocketmq-dashboard"
}

usage()
{
    echo " "
    echo "请输入 start 启动 RocketMQ NameServer。"
    echo "请输入 stop 停止 RocketMQ NameServer。"
    echo "请输入 status 获取 RocketMQ NameServer 状态。"
	echo "请输入 restart 重启 RocketMQ NameServer。"
	echo "请输入 start_broker 启动 RocketMQ Broker + Proxy。"
    echo "请输入 stop_broker 停止 RocketMQ Broker + Proxy。"
    echo "请输入 status_broker 获取 RocketMQ Broker + Proxy 状态。"
	echo "请输入 restart_broker 重启 RocketMQ Broker + Proxy。"
	
	echo "请输入 start_dashboard 启动 RocketMQ DashBoard。"
    echo "请输入 stop_dashboard 停止 RocketMQ DashBoard。"
    echo "请输入 status_dashboard 获取 RocketMQ DashBoard 状态。"
	echo "请输入 restart_dashboard 重启 RocketMQ DashBoard。"
	
	echo "请输入 remove 删除 RocketMQ。"
    echo " "
}

INPUT_ACTIVE=$1
ACTIVE=${INPUT_ACTIVE:=usage}
case ${ACTIVE} in
    start)
        start
        ;;
    stop)
        stop
        ;;
	status)
        status
        ;;
	restart)
        restart
        ;;
	start_broker)
        start_broker
        ;;
    stop_broker)
        stop_broker
        ;;
	status_broker)
        status_broker
        ;;
	restart_broker)
        restart_broker
        ;;
	start_dashboard)
        start_dashboard
        ;;
    stop_dashboard)
        stop_dashboard
        ;;
	status_dashboard)
        status_dashboard
        ;;
	restart_dashboard)
        restart_dashboard
        ;;
	remove)
        remove
        ;;
    *)
        usage
        ;;
esac
```

## Broker 区分内、外网IP
Broker 默认只监听内网 IP，如果需要监听外网 IP，需要修改 Broker 的配置文件  
修改 conf/broker.conf 文件，添加 brokerIP1=外网IP  

## 开启 ACL 验证功能
参考文档：https://rocketmq.apache.org/zh/docs/bestPractice/03access  
### 1、开启功能
Broker 默认不开启 ACL 验证功能，如果需要开启，需要修改 Broker 的配置文件  
修改 conf/broker.conf 文件，添加 aclEnable = true  
修改完成后需要重启，否则 Broker 不会生效
### 2、功能配置
配置文件：conf/plain_acl.yml  
用户可以动态修改权限控制定义的属性，而不需重新启动Broker服务节点。
```yaml
globalWhiteRemoteAddresses:
  - 10.0.20.14
  - 127.0.0.1
    
accounts:
  - accessKey: RocketMQ
    secretKey: 12345678
    whiteRemoteAddress:
    admin: false
    defaultTopicPerm: DENY
    defaultGroupPerm: SUB
    topicPerms:
      - topicA=DENY
      - topicB=PUB|SUB
      - topicC=SUB
    groupPerms:
      - groupA=DENY
      - groupB=PUB|SUB
      - groupC=SUB

  - accessKey: admin678
    secretKey: 123456789
    whiteRemoteAddress: 
    admin: true
```