RocketMQ核心组件实战：从零部署namesrv、broker与dashboard，并完成消息收发（三）

1. 环境准备与组件部署

在开始部署RocketMQ核心组件之前，我们需要先准备好基础环境。这里以Linux系统为例，假设你已经安装了JDK 1.8+和Maven环境。如果你还没有安装，可以通过以下命令快速安装：

# 安装OpenJDK 8 sudo apt-get update sudo apt-get install openjdk-8-jdk # 验证Java版本 java -version # 安装Maven sudo apt-get install maven

1.1 下载RocketMQ

首先从Apache官网下载最新稳定版的RocketMQ：

wget https://archive.apache.org/dist/rocketmq/5.1.3/rocketmq-all-5.1.3-bin-release.zip unzip rocketmq-all-5.1.3-bin-release.zip cd rocketmq-5.1.3

1.2 部署NameServer

NameServer是RocketMQ的路由中心，负责管理Broker的路由信息。部署NameServer非常简单：

# 启动NameServer nohup sh bin/mqnamesrv & # 查看启动日志，确认是否启动成功 tail -f ~/logs/rocketmqlogs/namesrv.log

当你看到"The Name Server boot success..."的日志时，说明NameServer已经成功启动。默认情况下，NameServer会监听9876端口。

1.3 部署Broker

Broker是RocketMQ的核心组件，负责消息的存储和转发。在部署Broker前，我们需要先修改配置文件：

# 修改Broker配置文件 vi conf/broker.conf

在配置文件中添加以下基本配置：

brokerClusterName = DefaultCluster brokerName = broker-a brokerId = 0 deleteWhen = 04 fileReservedTime = 48 brokerRole = ASYNC_MASTER flushDiskType = ASYNC_FLUSH namesrvAddr=127.0.0.1:9876

保存后启动Broker：

# 启动Broker nohup sh bin/mqbroker -c conf/broker.conf & # 查看启动日志 tail -f ~/logs/rocketmqlogs/broker.log

看到"The broker[broker-a, 192.168.1.1:10911] boot success..."表示Broker启动成功。

1.4 部署Dashboard

Dashboard是RocketMQ的可视化管理控制台，我们可以通过Docker快速部署：

# 拉取最新Dashboard镜像 docker pull apacherocketmq/rocketmq-dashboard:latest # 启动Dashboard容器 docker run -d --name rocketmq-dashboard -e "JAVA_OPTS=-Drocketmq.config.namesrvAddrs=127.0.0.1:9876" -p 8080:8080 -t apacherocketmq/rocketmq-dashboard:latest

启动后，通过浏览器访问http://localhost:8080即可看到Dashboard界面。

2. 组件连通性验证

2.1 验证NameServer与Broker通信

Broker启动后会自动向NameServer注册。我们可以通过以下命令检查Broker是否成功注册：

# 查看Broker注册信息 sh bin/mqadmin clusterList -n 127.0.0.1:9876

如果输出中包含你部署的Broker信息，说明注册成功。

2.2 通过Dashboard验证

在Dashboard的"集群"页面，你应该能看到已注册的Broker节点信息。如果看不到，请检查：

1. Dashboard配置的NameServer地址是否正确
2. Broker配置的NameServer地址是否正确
3. 防火墙是否放行了相关端口（9876, 10911等）

2.3 测试消息收发

我们可以使用RocketMQ自带的tools工具测试消息收发：

# 发送测试消息 sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer # 在另一个终端消费消息 sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

如果能看到消息正常生产和消费，说明整个系统运转正常。

3. 消息发送模式实战

3.1 同步发送模式

同步发送是最常用的模式，生产者会等待Broker返回确认消息后才继续执行。下面是Java代码示例：

public class SyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("sync_producer_group"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start(); for (int i = 0; i < 10; i++) { Message msg = new Message("TestTopic", "TagA", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); // 同步发送，等待Broker返回结果 SendResult sendResult = producer.send(msg); System.out.printf("%s%n", sendResult); } producer.shutdown(); } }

同步发送的优点是可靠性高，缺点是吞吐量相对较低，适合对可靠性要求高的场景。

3.2 异步发送模式

异步发送不需要等待Broker返回结果，通过回调函数处理发送结果：

public class AsyncProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("async_producer_group"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start(); for (int i = 0; i < 10; i++) { Message msg = new Message("TestTopic", "TagA", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); // 异步发送，设置回调函数 producer.send(msg, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.out.printf("发送成功: %s%n", sendResult); } @Override public void onException(Throwable e) { System.out.printf("发送失败: %s%n", e); e.printStackTrace(); } }); } // 异步发送需要等待回调完成 Thread.sleep(5000); producer.shutdown(); } }

异步发送的优点是吞吐量高，缺点是需要在回调函数中处理异常情况。

3.3 单向发送模式

单向发送不关心发送结果，适用于日志收集等场景：

public class OnewayProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("oneway_producer_group"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start(); for (int i = 0; i < 10; i++) { Message msg = new Message("TestTopic", "TagA", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)); // 单向发送，不关心结果 producer.sendOneway(msg); } producer.shutdown(); } }

单向发送的吞吐量最高，但可靠性最低，适合对可靠性要求不高的场景。

4. Dashboard监控与验证

4.1 查看消息轨迹

在Dashboard的"消息"页面，可以查看消息的发送和消费情况。点击"消息轨迹"可以查看消息的完整生命周期：

1. 消息发送时间
2. 消息存储的Broker
3. 消息被哪些消费者消费
4. 消费是否成功

4.2 创建和管理Topic

在"主题"页面，可以创建新的Topic或管理现有Topic：

1. 点击"新增/更新"按钮
2. 输入Topic名称（如"TestTopic"）
3. 设置读写权限（6表示可读可写）
4. 设置队列数量（默认为8）
5. 点击"提交"完成创建

4.3 重置消费位点

如果消息消费出现问题，可以在Dashboard上重置消费位点：

1. 进入"主题"页面
2. 找到目标Topic
3. 点击"重置消费位点"
4. 选择消费者组和重置时间点
5. 确认重置

注意：广播模式不支持重置消费位点，且消费者必须在线才能重置。

4.4 发送测试消息

Dashboard还提供了直接发送测试消息的功能：

1. 进入"主题"页面
2. 找到目标Topic
3. 点击"发送消息"
4. 输入消息内容、Tag和Key
5. 点击"发送"

这个功能非常适合在排查问题时快速验证消息通路是否正常。