RocketMQ 消息队列集群部署与运维

一、概述

RocketMQ 是阿里巴巴开源的分布式消息中间件，具有高吞吐、低延迟、高可用的特点，广泛应用于日志收集、削峰填谷、异步解耦等场景。

二、架构组件

组件 说明
NameServer 无状态注册中心，管理 Broker 路由信息
Broker 消息存储与转发核心，支持主从部署
Producer 消息生产者
Consumer 消息消费者，支持集群/广播模式
Topic 消息分类标识
Queue Topic 的物理分区

三、环境准备

3.1 系统要求

  • JDK 1.8+
  • 内存 ≥ 4GB（生产建议 8GB+）
  • 磁盘 ≥ 50GB（根据消息保留策略调整）

3.2 下载安装

1
2
3
4
5
# 下载 RocketMQ 5.x
wget https://archive.apache.org/dist/rocketmq/5.1.1/rocketmq-all-5.1.1-bin-release.zip
unzip rocketmq-all-5.1.1-bin-release.zip
cd rocketmq-5.1.1
export ROCKETMQ_HOME=$(pwd)

四、单机部署（开发环境）

4.1 启动 NameServer

1
2
nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log

4.2 启动 Broker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 修改 broker.conf 配置
cat > conf/broker.conf << EOF
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=ASYNC_MASTER
flushDiskType=ASYNC_FLUSH
storePathRootDir=/data/rocketmq/store
storePathCommitLog=/data/rocketmq/store/commitlog
storePathConsumeQueue=/data/rocketmq/store/consumequeue
storePathIndex=/data/rocketmq/store/index
checkTransactionMessageEnable=false
brokerIP1=192.168.1.100
EOF

# 启动 Broker
nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &
tail -f ~/logs/rocketmqlogs/broker.log

五、集群部署（生产环境）

5.1 多 NameServer 部署

1
2
3
4
# 在三台服务器分别启动 NameServer
# Server1: 192.168.1.101:9876
# Server2: 192.168.1.102:9876
# Server3: 192.168.1.103:9876

5.2 Broker 主从配置

1
2
3
4
5
6
7
8
9
10
11
12
13
# Master 配置 (broker-a-s.conf)
brokerClusterName=ProdCluster
brokerName=broker-a
brokerId=0
brokerRole=ASYNC_MASTER
namesrvAddr=192.168.1.101:9876;192.168.1.102:9876;192.168.1.103:9876

# Slave 配置 (broker-a-s.conf)
brokerClusterName=ProdCluster
brokerName=broker-a
brokerId=1
brokerRole=SLAVE
namesrvAddr=192.168.1.101:9876;192.168.1.102:9876;192.168.1.103:9876

5.3 启动脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
#!/bin/bash
# start-cluster.sh

NAMESERVERS="192.168.1.101:9876;192.168.1.102:9876;192.168.1.103:9876"

# 启动所有 NameServer
for ns in 101 102 103; do
    ssh 192.168.1.$ns "nohup sh /opt/rocketmq/bin/mqnamesrv &"
done

# 启动所有 Broker
for broker in 104 105 106; do
    ssh 192.168.1.$broker "nohup sh /opt/rocketmq/bin/mqbroker -n $NAMESERVERS -c /opt/rocketmq/conf/broker.conf &"
done

六、控制台部署

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 下载 RocketMQ Dashboard
git clone https://github.com/apache/rocketmq-dashboard.git
cd rocketmq-dashboard

# 修改配置
cat > src/main/resources/application.properties << EOF
server.port=8080
rocketmq.namesrv.addr=192.168.1.101:9876;192.168.1.102:9876;192.168.1.103:9876
rocketmq.login.enabled=true
rocketmq.login.username=admin
rocketmq.login.password=admin123
EOF

# 编译运行
mvn clean package -DskipTests
java -jar target/rocketmq-dashboard-*.jar

七、运维管理

7.1 常用命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 查看集群状态
sh bin/mqadmin clusterList -n localhost:9876

# 查看 Topic 列表
sh bin/mqadmin topicList -n localhost:9876

# 查看消费者状态
sh bin/mqadmin consumerProgress -n localhost:9876 -g consumerGroup

# 创建 Topic
sh bin/mqadmin updateTopic -n localhost:9876 -t MyTopic -w 4 -r 4

# 删除 Topic
sh bin/mqadmin deleteTopic -n localhost:9876 -t MyTopic

# 重置消费位点
sh bin/mqadmin resetOffsetByTime -n localhost:9876 -g consumerGroup -t MyTopic -s 1678320000000

7.2 监控指标

指标 说明 告警阈值
消息堆积量 Consumer 未消费消息数 > 10000
Broker TPS 每秒消息吞吐量 持续 < 100
消息延迟 消息生产到消费的时间 > 5s
NameServer 存活 注册中心健康状态 节点数 < 2
磁盘使用率 消息存储磁盘占用 > 85%

7.3 日志管理

1
2
3
4
5
6
7
8
9
10
11
# 日志路径
~/logs/rocketmqlogs/namesrv.log    # NameServer 日志
~/logs/rocketmqlogs/broker.log     # Broker 日志
~/logs/rocketmqlogs/consumer.log   # 消费者日志
~/logs/rocketmqlogs/producer.log   # 生产者日志

# 日志轮转配置 (logback.xml)
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
    <fileNamePattern>~/logs/rocketmqlogs/broker.%d{yyyy-MM-dd}.log</fileNamePattern>
    <maxHistory>30</maxHistory>
</rollingPolicy>

八、故障排查

8.1 Broker 无法启动

1
2
3
4
5
6
7
8
9
# 检查端口占用
netstat -tlnp | grep 9876
netstat -tlnp | grep 10911

# 检查磁盘空间
df -h /data/rocketmq

# 检查 JVM 内存
jps -v | grep BrokerStartup

8.2 消息堆积处理

1
2
3
4
5
6
7
# 1. 查看堆积情况
sh bin/mqadmin consumerProgress -n localhost:9876 -g orderConsumer

# 2. 增加消费者实例
# 3. 检查消费者消费逻辑是否阻塞
# 4. 必要时重置消费位点
sh bin/mqadmin resetOffsetByTime -n localhost:9876 -g orderConsumer -t OrderTopic -s $(date -d "1 hour ago" +%s)000

8.3 消息重复消费

  • 确保消费者幂等性设计
  • 检查消费 ACK 机制
  • 排查网络抖动导致的重平衡

九、性能优化

9.1 Broker 参数调优

1
2
3
4
5
6
7
8
9
10
11
# 异步刷盘，提高吞吐
flushDiskType=ASYNC_FLUSH

# 增加队列数
defaultTopicQueueNums=8

# 消息保留时间
fileReservedTime=72

# 预分配文件大小
mappedFileSizeCommitLog=1073741824

9.2 JVM 调优

1
2
3
4
5
# bin/runbroker.sh
JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g"
JAVA_OPT="${JAVA_OPT} -XX:+UseG1GC"
JAVA_OPT="${JAVA_OPT} -XX:MaxGCPauseMillis=200"
JAVA_OPT="${JAVA_OPT} -XX:InitiatingHeapOccupancyPercent=45"

十、安全加固

10.1 ACL 配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# conf/plain_acl.yml
globalWhiteRemoteAddresses:
- 192.168.1.*
accounts:
- accessKey: rocketmq_admin
  secretKey: admin_secret
  whiteRemoteAddress: 192.168.1.*
  admin: true
- accessKey: rocketmq_app
  secretKey: app_secret
  whiteRemoteAddress: 192.168.1.*
  admin: false
  defaultTopicPerm: PUB|SUB
  defaultGroupPerm: SUB

10.2 网络隔离

  • NameServer/Broker 端口不对外暴露
  • 使用内网通信
  • 配置防火墙规则

十一、备份与恢复

11.1 消息备份

1
2
3
4
5
# 导出 Topic 配置
sh bin/mqadmin updateTopic -n localhost:9876 -t MyTopic > topic_backup.txt

# 备份存储目录
tar -czf rocketmq_store_$(date +%Y%m%d).tar.gz /data/rocketmq/store

11.2 灾难恢复

  1. 恢复存储目录
  2. 重启 NameServer
  3. 重启 Broker
  4. 验证 Topic 和消息完整性

十二、总结

RocketMQ 作为成熟的消息中间件，在生产环境中需要重点关注：

  1. 高可用: NameServer 多节点 + Broker 主从
  2. 监控告警: 消息堆积、延迟、磁盘使用率
  3. 性能调优: 刷盘策略、JVM 参数、队列数
  4. 安全加固: ACL 权限控制、网络隔离

合理规划和运维是保障消息队列稳定运行的关键。