RocketMQ核心组件实战:从零部署namesrv、broker与dashboard,并完成消息收发(三)

📅 2026/7/14 12:10:29 👁️ 阅读次数 📝 编程学习
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. 点击"发送"

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