当前位置: 开发笔记 > 编程语言 > 正文

开发笔记:原创《从0开始学RocketMQ》—集群搭建

作者：yoyokk99的秋天 | 来源：互联网 | 2023-09-24 21:08

本文由编程笔记#小编为大家整理，主要介绍了原创《从0开始学RocketMQ》—集群搭建相关的知识，希望对你有一定的参考价值。用两台服务器，搭建出一个双master双slave、

本文由编程笔记#小编为大家整理，主要介绍了原创《从0开始学RocketMQ》—集群搭建相关的知识，希望对你有一定的参考价值。

用两台服务器，搭建出一个双master双slave、无单点故障的高可用 RocketMQ 集群。此处假设两台服务器的物理 IP 分别为：192.168.50.1、192.168.50.2。

内容目录

1. 启动 NameServer 集群

2. 启动 Broker 集群

3. RocketMQ 可视化管理控制台：rocketmq-console

4. 集群测试

1. 启动 NameServer 集群

在两台服务器上分别启动 NameServer，可以得到一个无单点故障的 NameServer 服务，服务地址分别为：192.168.50.1:9876、192.168.50.2:9876。

2. 启动 Broker 集群

修改 Broker 配置文件，以使每台服务器上都可以启动一个 Master 角色的 Broker 和一个Slave 角色的 Broker。
首先找到 Broker 配置文件，此处我们搭建一个同步双写模式的集群，所以需要修改 2m-2s-sync 目录下的 broker 配置文件：

[[email&＃160;protected]157-89 ~]# cd /usr/local/rocketmq-all-4.3.2-bin-release/conf/
[[email&＃160;protected]157-89 conf]# ls
2m-2s-async 2m-2s-sync 2m-noslave broker.conf logback_broker.xml logback_namesrv.xml logback_tools.xml
[[email&＃160;protected]157-89 conf]# cd 2m-2s-sync/
[[email&＃160;protected]157-89 2m-2s-sync]# ls
broker-a.properties broker-a-s.properties broker-b.properties broker-b-s.properties

1）修改 192.168.50.1 服务器上的 broker-a.properties 为 Master 角色的 Broker：

namesrvAddr=192.168.50.1:9876;192.168.50.2:9876
brokerClusterName=rocketMqCluster
brokerIP1=192.168.50.1
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=120
mapedFileSizeConsumeQueue=500000
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10911
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
storePathRootDir=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a
storePathCommitLog=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a/commitlog
storePathConsumeQueue=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a/consumequeue
storePathIndex=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a/index
storeCheckpoint=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a/checkpoint
abortFile=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a/abort

2) 修改 192.168.50.2 服务器上的 broker-b.properties 为 Master 角色的 Broker：

namesrvAddr=192.168.50.1:9876;192.168.50.2:9876
brokerClusterName=rocketMqCluster
brokerIP1=192.168.50.2
brokerName=broker-b
brokerId=0
deleteWhen=04
fileReservedTime=120
mapedFileSizeConsumeQueue=500000
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10911
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
storePathRootDir=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b
storePathCommitLog=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b/commitlog
storePathConsumeQueue=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b/consumequeue
storePathIndex=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b/index
storeCheckpoint=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b/checkpoint
abortFile=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b/abort

3) 修改 192.168.50.1 服务器上的 broker-b-s.properties 为 Slave 角色的 Broker：

namesrvAddr=192.168.50.1:9876;192.168.50.2:9876
brokerClusterName=rocketMqCluster
brokerIP1=192.168.50.1
brokerName=broker-b
brokerId=1
deleteWhen=04
fileReservedTime=120
mapedFileSizeConsumeQueue=500000
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=10921
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
storePathRootDir=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b-s
storePathCommitLog=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b-s/commitlog
storePathConsumeQueue=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b-s/consumequeue
storePathIndex=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b-s/index
storeCheckpoint=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b-s/checkpoint
abortFile=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-b-s/abort

4) 修改 192.168.50.2 服务器上的 broker-a-s.properties 为 Slave 角色的 Broker：

namesrvAddr=192.168.50.1:9876;192.168.50.2:9876
brokerClusterName=rocketMqCluster
brokerIP1=192.168.50.2
brokerName=broker-a
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=10921
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
storePathRootDir=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a-s
storePathCommitLog=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a-s/commitlog
storePathConsumeQueue=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a-s/consumequeue
storePathIndex=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a-s/index
storeCheckpoint=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a-s/checkpoint
abortFile=/usr/local/rocketmq-all-4.3.2-bin-release/data/broker-a-s/abort

一台服务器上启动多个Broker 时，需指定不同的端口号，记得防火墙放开 NameServer 和 Broker 中用到的端口号哦～

分别启动四个 Broker:

nohup sh bin/mqbroker -c broker_config_file &

3. RocketMQ 可视化管理控制台：rocketmq-console

在服务器 192.168.50.1 上安装即可，无需集群

[[email&＃160;protected]153-215 local]# git clone https://github.com/apache/rocketmq-externals.git
Cloning into ‘rocketmq-externals‘...
remote: Enumerating objects: 10, done.
remote: Counting objects: 100% (10/10), done.
remote: Compressing objects: 100% (10/10), done.
remote: Total 9425 (delta 2), reused 1 (delta 0), pack-reused 9415
Receiving objects: 100% (9425/9425), 11.86 MiB | 232.00 KiB/s, done.
Resolving deltas: 100% (4235/4235), done.
[[email&＃160;protected]153-215 local]# cd rocketmq-externals/
[[email&＃160;protected]153-215 rocketmq-externals]# ls
dev README.md rocketmq-console rocketmq-docker rocketmq-flink rocketmq-flume rocketmq-hbase rocketmq-iot-bridge rocketmq-jms rocketmq-mysql rocketmq-php rocketmq-redis rocketmq-sentinel rocketmq-serializer rocketmq-spark
[[email&＃160;protected]153-215 rocketmq-externals]# git branch
* master
[[email&＃160;protected]153-215 rocketmq-externals]# git fetch origin release-rocketmq-console-1.0.0
From https://github.com/apache/rocketmq-externals
* branch release-rocketmq-console-1.0.0 -> FETCH_HEAD
[[email&＃160;protected]153-215 rocketmq-externals]# git checkout -b release-1.0.0 origin/release-rocketmq-console-1.0.0
Branch ‘release-1.0.0‘ set up to track remote branch ‘release-rocketmq-console-1.0.0‘ from ‘origin‘.
Switched to a new branch ‘release-1.0.0‘
[[email&＃160;protected]153-215 rocketmq-externals]# ls
README.md rocketmq-console
[[email&＃160;protected]153-215 rocketmq-externals]# ls rocketmq-console/
doc LICENSE NOTICE pom.xml README.md src style
[[email&＃160;protected]153-215 rocketmq-externals]# vim rocketmq-console/src/main/resources/application.properties

编辑 application.properties：

server.cOntextPath=/rocketmq
server.port=8080
#spring.application.index=true
spring.application.name=rocketmq-console
spring.http.encoding.charset=UTF-8
spring.http.encoding.enabled=true
spring.http.encoding.force=true
logging.config=classpath:logback.xml
#if this value is empty,use env value rocketmq.config.namesrvAddr NAMESRV_ADDR | now, you can set it in ops page.default localhost:9876
rocketmq.config.namesrvAddr=192.168.50.1:9876;192.168.50.2:9876
#if you use rocketmq version <3.5.8, rocketmq.config.isVIPChannel should be false.default true
rocketmq.config.isVIPChannel=
#rocketmq-console‘s data path:dashboard/monitor
rocketmq.config.dataPath=/tmp/rocketmq-console/data
#set it false if you don‘t want use dashboard.default true
rocketmq.config.enableDashBoardCollect=true

移动 rocketmq-console 所在目录，编译并启动 rocketmq-console：

[[email&＃160;protected]153-215 rocketmq-console]# mv /usr/local/rocketmq-externals/rocketmq-console /usr/local/rocketmq-console
[[email&＃160;protected]153-215 rocketmq-console]# cd /usr/local/rocketmq-console/
[[email&＃160;protected]153-215 rocketmq-console]# ls
doc LICENSE NOTICE pom.xml README.md src style
[[email&＃160;protected]153-215 rocketmq-console]# mvn clean package -Dmaven.test.skip=true
........
[INFO] Building jar: /usr/local/rocketmq-console/target/rocketmq-console-ng-1.0.0-sources.jar
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 02:54 min
[INFO] Finished at: 2019-01-11T17:02:34+08:00
[INFO] ------------------------------------------------------------------------
[[email&＃160;protected]153-215 rocketmq-console]# ls
doc LICENSE NOTICE pom.xml README.md src style target
[[email&＃160;protected]153-215 rocketmq-console]# ls target/
checkstyle-cachefile checkstyle-checker.xml checkstyle-result.xml classes generated-sources maven-archiver maven-status rocketmq-console-ng-1.0.0.jar rocketmq-console-ng-1.0.0.jar.original rocketmq-console-ng-1.0.0-sources.jar
[[email&＃160;protected]153-215 rocketmq-console]# java -jar target/rocketmq-console-ng-1.0.0.jar
.......
[2019-01-11 17:04:15.980] INFO Initializing ProtocolHandler ["http-nio-8080"]
[2019-01-11 17:04:15.991] INFO Starting ProtocolHandler [http-nio-8080]
[2019-01-11 17:04:16.232] INFO Using a shared selector for servlet write/read
[2019-01-11 17:04:16.251] INFO Tomcat started on port(s): 8080 (http)
[2019-01-11 17:04:16.257] INFO Started App in 6.594 seconds (JVM running for 7.239)

4. 集群测试

Producer 测试代码：

public class SyncProducerTest {
public static void main(String[] args) {
DefaultMQProducer producer = new DefaultMQProducer("producer_test_group");
producer.setNamesrvAddr("39.107.153.215:9876;39.107.157.89:9876");
try{
producer.start();
for(int i=0;i<100;i++){
Message message = new Message("topic_test", "tag_test", ("Hello World" + 1).getBytes("UTF-8"));
SendResult sendResult = producer.send(message);
System.out.println(JSON.toJSON(sendResult));
}
producer.shutdown();
}catch (Exception e){
e.printStackTrace();
}
}
}

Consumer 测试代码：

public class SyncConsumerTest {
public static void main(String[] args) {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_test_group");
consumer.setNamesrvAddr("39.107.153.215:9876;39.107.157.89:9876");
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
try {
consumer.subscribe("topic_test", "*");
consumer.registerMessageListener((MessageListenerConcurrently) (messageList, context) -> {
System.out.println(Thread.currentThread().getName() + " Receive New Message:" + messageList);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
}catch (Exception e){
e.printStackTrace();
}
}
}

SyncProducerTest 运行日志：

技术分享图片

SyncConsumerTest 运行日志：

技术分享图片

通过日志可以看到，消费者、生产者收发消息都是正常的，我们去可视化管理控制台查看下 http://192.168.50.1:8080/rocketmq：

技术分享图片

通过管控台可以看到，双 master 双 slave 的 broker 集群一切正常，并可进一步看到每个 broker 处理消息的情况。

推荐阅读

java
Java面试题解析

本文详细介绍了Java编程语言中的核心概念和常见面试问题，包括集合类、数据结构、线程处理、Java虚拟机（JVM）、HTTP协议以及Git操作等方面的内容。通过深入分析每个主题，帮助读者更好地理解Java的关键特性和最佳实践。 ... [详细]

蜡笔小新 2024-12-27 13:55:14
case
UNP 第9章：主机名与地址转换

本章探讨了用于在主机名和数值地址之间进行转换的函数，如gethostbyname和gethostbyaddr。此外，还介绍了getservbyname和getservbyport函数，用于在服务器名和端口号之间进行转换。 ... [详细]

蜡笔小新 2024-12-27 11:26:39
java
深入解析 HTTP 500 内部服务器错误

本文详细探讨了HTTP 500内部服务器错误的成因、解决方案及其在Web开发中的影响。通过对具体案例的分析，帮助读者理解并解决此类问题。 ... [详细]

蜡笔小新 2024-12-26 16:55:25
eval
PHP 5.5.0rc1 发布：深入解析 Zend OPcache

2013年5月9日，PHP官方发布了PHP 5.5.0rc1和PHP 5.4.15正式版，这两个版本均支持64位环境。本文将详细介绍Zend OPcache的功能及其在Windows环境下的配置与测试。 ... [详细]

蜡笔小新 2024-12-26 12:56:20
client
Linux下NFS客户端配置详解

NFS（Network File System）即网络文件系统，是一种分布式文件系统协议，主要用于Unix和类Unix系统之间的文件共享。本文详细介绍NFS的配置文件/etc/exports和相关服务配置，帮助读者理解如何在Linux环境中配置NFS客户端。 ... [详细]

蜡笔小新 2024-12-18 18:23:35
object
Windows服务与数据库交互问题解析

本文探讨了在Windows 10（64位）环境下开发的Windows服务，旨在定期向本地MS SQL Server (v.11)插入记录。尽管服务已成功安装并运行，但记录并未正确插入。我们将详细分析可能的原因及解决方案。 ... [详细]

蜡笔小新 2024-12-28 10:30:14
install
网络链路质量监控：Smokeping部署与配置

本文详细介绍了如何在Linux系统上安装和配置Smokeping，以实现对网络链路质量的实时监控。通过详细的步骤和必要的依赖包安装，确保用户能够顺利完成部署并优化其网络性能监控。 ... [详细]

蜡笔小新 2024-12-27 19:31:05
install
路由器配置与网络地址转换

本文介绍了如何在具备多个IP地址的FTP服务器环境中，通过动态地址端口复用和地址转换技术优化网络配置。重点讨论了2Mb/s DDN专线连接、Cisco 2611路由器及内部网络地址规划。 ... [详细]

蜡笔小新 2024-12-27 18:25:35
java
深入理解Cookie与Session会话管理

本文详细介绍了如何通过HTTP响应和请求处理浏览器的Cookie信息，以及如何创建、设置和管理Cookie。同时探讨了会话跟踪技术中的Session机制，解释其原理及应用场景。 ... [详细]

蜡笔小新 2024-12-27 18:20:43
java
深入理解 SQL 视图、存储过程与事务

本文详细介绍了SQL中的视图、存储过程和事务的概念及应用。视图为用户提供了一种灵活的数据查询方式，存储过程则封装了复杂的SQL逻辑，而事务确保了数据库操作的完整性和一致性。 ... [详细]

蜡笔小新 2024-12-27 17:40:42
install
DNN Community 和 Professional 版本的主要差异

本文详细解析了 DotNetNuke (DNN) 的两种主要版本：Community 和 Professional。通过对比两者的功能和附加组件，帮助用户选择最适合其需求的版本。 ... [详细]

蜡笔小新 2024-12-27 13:14:08
java
寻找满足特定条件的整数N的最大和(a+b)

本文探讨了如何在给定整数N的情况下，找到两个不同的整数a和b，使得它们的和最大，并且满足特定的数学条件。 ... [详细]

蜡笔小新 2024-12-26 19:26:18
hash
优化Kafka流状态存储查询的最佳实践

本文探讨了如何优化和正确配置Kafka Streams应用程序以确保准确的状态存储查询。通过调整配置参数和代码逻辑，可以有效解决数据不一致的问题。 ... [详细]

蜡笔小新 2024-12-26 18:17:14
hash
QT中QWebSocket与QTcpSocket的SSL加密通信配置

本文介绍如何在QT框架中使用QWebSocket和QTcpSocket实现SSL加密通信，涵盖单向认证设置。单向认证常见于Web通信场景，其中客户端验证服务端证书，而服务端不验证客户端证书。 ... [详细]

蜡笔小新 2024-12-26 17:40:23
eval
从零构建递归神经网络：仅用NumPy实现

尽管使用TensorFlow和PyTorch等成熟框架可以显著降低实现递归神经网络（RNN）的门槛，但对于初学者来说，理解其底层原理至关重要。本文将引导您使用NumPy从头构建一个用于自然语言处理（NLP）的RNN模型。 ... [详细]

蜡笔小新 2024-12-26 11:29:15

yoyokk99的秋天

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章