RocketMQ的集群部署以及可视化监控界面
小编:管理员 250阅读 2022.08.03
RocketMQ is a distributed messaging and streaming platform with low latency, high performance and reliability, trillion-level capacity and flexible scalability. It is the third generation distributed messaging middleware open sourced by Alibaba in 2012. On November 21, 2016, Alibaba donated RocketMQ to the Apache Software Foundation. Next year, on February 20, the Apache Software Foundation announced Apache RocketMQ as a Top-Level Project. -- quote from Wikipedia
lg8YLt.png
RocketMQ是一个轻量级、高可用、低延时的消息中间件,能实现消息的存储,消息的失败重试,批量消息处理,延时消息处理等特性,在各种消息中间件中表现优异。
2.RocketMQ的组成部分RocketMQ一共包含四个部分:Name Server,Brokers,Producers,Consumers 。这几个部分都能单独的水平扩展。
- NameServer Cluster 轻量级的组件,主要用于服务的发现,能够读写路由信息以及全局的信息,支持高速的存储;
- Broker Cluster 轻量级的Topic与Queue的机制去管理数据的存储,方便实现容错需要2到3个数据分片的提供。客户端可以向Broker推送或者拉取数据,同时也支持灾难恢复和数据统计;
- Producer Cluster Producer能够进行分布式部署,推送消息到Brokers能进行均衡的分配,还具有高可用,低延时的特性;
- Consumer Cluster Consumer 也可以进行分布式的部署,推送和拉取消息,还能实时的订阅消息,从集群中消费消息,还能实现消息的广播。
服务器配置:
// 我们这里安装两台服务器,都是master,也就是两台master: 192.168.2.128 nameServer1,brokerServer1 192.168.2.135 nameServer2,brokerServer2复制
添加Host文件:
vim /etc/hosts // 添加如下配置 192.168.2.128 rocketmq-nameserver1 192.168.2.128 rocketmq-master1 192.168.2.135 rocketmq-nameserver2 192.168.2.135 rocketmq-master2 // 重启服务 service network restart复制
这里我们部署的是RocketMQ的3.2.6的版本,最新的版本可以参阅官网的地址。下载地址如下:
https://archive.apache.org/dist/rocketmq/3.2.6/rocketmq-all-3.2.6-source-release.zip复制
解压相关的包:
cd /usr/local wget https://archive.apache.org/dist/rocketmq/3.2.6/rocketmq-all-3.2.6-source-release.zip unzip rocketmq-all-3.2.6-source-release.zip mv rocketmq-all-3.2.6-source-release rocketmq-all-3.2.6 cd rocketmq-all-3.2.6/ // maven打包需要先配置maven环境 这里也可以在window的编译器中打包相关 mvn -Prelease-all -DskipTests clean install -U cd distribution/target/apache-rocketmq // 找到tar.gz tar -zxvf alibaba-rocketmq-3.2.6.tar.gz -C /usr/local mv alibaba-rocketmq alibaba-rocketmq-3.2.6 ln -s alibaba-rocketmq-3.2.6 rocketmq复制
创建存储路径:
mkdir /usr/local/rocketmq/store mkdir /usr/local/rocketmq/store/commitlog mkdir /usr/local/rocketmq/store/consumequeue mkdir /usr/local/rocketmq/store/index复制
RocketMq配置文件【两台机器】
vim /usr/local/rocketmq/conf/2m-noslave/broker-a.properties vim /usr/local/rocketmq/conf/2m-noslave/broker-b.properties复制
相关配置文件模板如下:
broker-a.properties:
#所属集群名字 brokerClusterName=rocketmq-cluster #broker名字,注意此处不同的配置文件填写的不一样 brokerName=broker-a|broker-b #0 表示 Master,>0 表示 Slave brokerId=0 #nameServer地址,分号分割 namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876 #在发送消息时,自动创建服务器不存在的topic,默认创建的队列数 defaultTopicQueueNums=4 #是否允许 Broker 自动创建Topic,建议线下开启,线上关闭 autoCreateTopicEnable=true #是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭 autoCreateSubscriptionGroup=true #Broker 对外服务的监听端口 listenPort=10911 #删除文件时间点,默认凌晨 4点 deleteWhen=04 #文件保留时间,默认 48 小时 fileReservedTime=120 #commitLog每个文件的大小默认1G mapedFileSizeCommitLog=1073741824 #ConsumeQueue每个文件默认存30W条,根据业务情况调整 mapedFileSizeConsumeQueue=300000 #destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #检测物理文件磁盘空间 diskMaxUsedSpaceRatio=88 #存储路径 storePathRootDir=/data/rocketmq/store #commitLog 存储路径 storePathCommitLog=/data/rocketmq/store/commitlog #消费队列存储路径存储路径 storePathConsumeQueue=/data/rocketmq/store/consumequeue #消息索引存储路径 storePathIndex=/data/rocketmq/store/index #checkpoint 文件存储路径 storeCheckpoint=/data/rocketmq/store/checkpoint #abort 文件存储路径 abortFile=/data/rocketmq/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 异步复制Master #- SYNC_MASTER 同步双写Master #- SLAVE brokerRole=ASYNC_MASTER #刷盘方式 #- ASYNC_FLUSH 异步刷盘 #- SYNC_FLUSH 同步刷盘 flushDiskType=ASYNC_FLUSH #checkTransactionMessageEnable=false #发消息线程池数量 #sendMessageThreadPoolNums=128 #拉消息线程池数量 #pullMessageThreadPoolNums=128复制
broker-b.properties:
#所属集群名字 brokerClusterName=rocketmq-cluster #broker名字,注意此处不同的配置文件填写的不一样 brokerName=broker-a|broker-b #0 表示 Master,>0 表示 Slave brokerId=0 #nameServer地址,分号分割 namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876 #在发送消息时,自动创建服务器不存在的topic,默认创建的队列数 defaultTopicQueueNums=4 #是否允许 Broker 自动创建Topic,建议线下开启,线上关闭 autoCreateTopicEnable=true #是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭 autoCreateSubscriptionGroup=true #Broker 对外服务的监听端口 listenPort=10911 #删除文件时间点,默认凌晨 4点 deleteWhen=04 #文件保留时间,默认 48 小时 fileReservedTime=120 #commitLog每个文件的大小默认1G mapedFileSizeCommitLog=1073741824 #ConsumeQueue每个文件默认存30W条,根据业务情况调整 mapedFileSizeConsumeQueue=300000 #destroyMapedFileIntervalForcibly=120000 #redeleteHangedFileInterval=120000 #检测物理文件磁盘空间 diskMaxUsedSpaceRatio=88 #存储路径 storePathRootDir=/data/rocketmq/store #commitLog 存储路径 storePathCommitLog=/data/rocketmq/store/commitlog #消费队列存储路径存储路径 storePathConsumeQueue=/data/rocketmq/store/consumequeue #消息索引存储路径 storePathIndex=/data/rocketmq/store/index #checkpoint 文件存储路径 storeCheckpoint=/data/rocketmq/store/checkpoint #abort 文件存储路径 abortFile=/data/rocketmq/store/abort #限制的消息大小 maxMessageSize=65536 #flushCommitLogLeastPages=4 #flushConsumeQueueLeastPages=2 #flushCommitLogThoroughInterval=10000 #flushConsumeQueueThoroughInterval=60000 #Broker 的角色 #- ASYNC_MASTER 异步复制Master #- SYNC_MASTER 同步双写Master #- SLAVE brokerRole=ASYNC_MASTER #刷盘方式 #- ASYNC_FLUSH 异步刷盘 #- SYNC_FLUSH 同步刷盘 flushDiskType=ASYNC_FLUSH #checkTransactionMessageEnable=false #发消息线程池数量 #sendMessageThreadPoolNums=128 #拉消息线程池数量 #pullMessageThreadPoolNums=128复制
修改日志配置文件【两台机器】
mkdir -p /usr/local/rocketmq/logs cd /usr/local/rocketmq/conf && sed -i 's#${user.home}#/usr/local/rocketmq#g' *.xml复制
修改启动NameServer【两台机器】
vim /usr/local/rocketmq/bin/runbroker.sh JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m - XX:PermSize=128m -XX:MaxPermSize=320m" vim /usr/local/rocketmq/bin/runserver.sh JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m - XX:PermSize=128m -XX:MaxPermSize=320m"复制
构建步骤如下:
①Start Name Server
cd /usr/local/rocketmq/bin nohup sh bin/mqnamesrv & tail -200f ~/logs/rocketmqlogs/namesrv.log复制
②Start Broker
// 启动BrokerServerA: cd /usr/local/rocketmq/bin nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-noslave/broker-a.properties & // 可用jps查看启动是否成功: jps // 启动BrokerServerB: cd /usr/local/rocketmq/bin nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-noslave/broker-b.properties &复制4.部署RocketMQ Console将rocketmq-web-console 部署到webapps目录中。
// 下载源码地址: https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console // 下载后需要修改config.properties rocketmq.namesrv.addr=192.168.2.128:9876;192.168.2.135:9876 // 使用maven打成war包 // 部署在tomcat如下目录: /usr/local/apache-tomcat-7.0.65/webapps/ // 启动tomcat,并访问: http://IP地址:8080/rocketmq-web-console/cluster/list.do // 既可以查看相应的集群状况复制5.Java调用RocketMQpom.xml:
复制Producer:org.apache.rocketmq rocketmq-client 4.6.0 com.alibaba.rocketmq rocketmq-all 3.0.10 pom
public class Producer { public static void main(String[] args) throws Exception { // 1.分组 DefaultMQProducer producer = new DefaultMQProducer("rmq-group"); // 2.服务器集群地址 producer.setNamesrvAddr("192.168.2.128:9876;192.168.2.135:9876"); producer.setInstanceName("producer"); producer.start(); for (int i = 0; i < 1; i++) { Thread.sleep(1000); Message message = new Message( "fbc_topic", "tagA", ("fbc--" + i).getBytes() ); SendResult sendResult = producer.send(message); System.out.println(sendResult.toString()); System.out.println("producer send:"+message); } producer.shutdown(); } }复制Consumer:
public class Consumer { private static Map复制map = new HashMap<>(); public static void main(String[] args) throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group"); consumer.setNamesrvAddr("192.168.2.128:9876;192.168.2.135:9876"); consumer.setInstanceName("consumer"); consumer.subscribe( "fbc_topic", "tagA" ); consumer.registerMessageListener( new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { for (MessageExt messageExt : list) { String msgId = messageExt.getMsgId(); byte[] body = messageExt.getBody(); System.out.println("msgId:" + msgId + ",body:" + new String(body)); } if(map.get("try")!=null){ System.out.println("重复消费"); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } System.out.println("开始报错"); try { int w = 1/0; } catch (Exception e) { e.printStackTrace(); map.put("try","1"); return ConsumeConcurrentlyStatus.RECONSUME_LATER; } // 消费状态 1:代表消费成功 2.代表消费失败 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } } ); consumer.start(); System.out.println("Consumer Started."); } }
相关推荐
- 【RocketMQ系列】RocketMQ集群,RocketMQ-on-DLedger集群 本文RocketMQ系列第四篇,主要介绍RocketMQ集群及如何部署自动容灾切换的 RocketMQ-on-DLedger Group。RocketMQ集群搭建ROcketMQ集群搭建有以下几种方案:「单Master模式」「多Master模式」「多Master多Slave模式-异步复制」「多Master多Slave模式-同步双写」其…
- 3DMAX提示和技巧 本主题标识使用 Civil View 的一些重要提示和技巧。常规使用屏幕分辨率至少为 1280x1024 的 Civil View。低于此分辨率时,一些面板将占用过多屏幕空间。 将视口设置为线框显示以达到最佳性能。 要尽可能简化用户界面,请在单个视口中工作并关闭 3ds Max 命令面…