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 maven1.1 下载RocketMQ首先从Apache官网下载最新稳定版的RocketMQwget 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.31.2 部署NameServerNameServer是RocketMQ的路由中心负责管理Broker的路由信息。部署NameServer非常简单# 启动NameServer nohup sh bin/mqnamesrv # 查看启动日志确认是否启动成功 tail -f ~/logs/rocketmqlogs/namesrv.log当你看到The Name Server boot success...的日志时说明NameServer已经成功启动。默认情况下NameServer会监听9876端口。1.3 部署BrokerBroker是RocketMQ的核心组件负责消息的存储和转发。在部署Broker前我们需要先修改配置文件# 修改Broker配置文件 vi conf/broker.conf在配置文件中添加以下基本配置brokerClusterName DefaultCluster brokerName broker-a brokerId 0 deleteWhen 04 fileReservedTime 48 brokerRole ASYNC_MASTER flushDiskType ASYNC_FLUSH namesrvAddr127.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 部署DashboardDashboard是RocketMQ的可视化管理控制台我们可以通过Docker快速部署# 拉取最新Dashboard镜像 docker pull apacherocketmq/rocketmq-dashboard:latest # 启动Dashboard容器 docker run -d --name rocketmq-dashboard -e JAVA_OPTS-Drocketmq.config.namesrvAddrs127.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节点信息。如果看不到请检查Dashboard配置的NameServer地址是否正确Broker配置的NameServer地址是否正确防火墙是否放行了相关端口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的消息页面可以查看消息的发送和消费情况。点击消息轨迹可以查看消息的完整生命周期消息发送时间消息存储的Broker消息被哪些消费者消费消费是否成功4.2 创建和管理Topic在主题页面可以创建新的Topic或管理现有Topic点击新增/更新按钮输入Topic名称如TestTopic设置读写权限6表示可读可写设置队列数量默认为8点击提交完成创建4.3 重置消费位点如果消息消费出现问题可以在Dashboard上重置消费位点进入主题页面找到目标Topic点击重置消费位点选择消费者组和重置时间点确认重置注意广播模式不支持重置消费位点且消费者必须在线才能重置。4.4 发送测试消息Dashboard还提供了直接发送测试消息的功能进入主题页面找到目标Topic点击发送消息输入消息内容、Tag和Key点击发送这个功能非常适合在排查问题时快速验证消息通路是否正常。
