多master模式部署RocketMQ

向右看齐 2022-12-24 15:54 200阅读 0赞

前言：目前项目打算把一些操作都通过RocketMq改成异步的，现在系统是只有一台单点RocketMq，担心万一挂了就太影响使用了，所以打算部署成集群的，网上看到RocketMq有四种不同的部署模式，针对我这种只有两台电脑的，我打算部署成多master模式。供本人复习之用，不一定完全正确。

参考：

[https://blog.csdn.net/qq\_36147908/article/details/107844959][https_blog.csdn.net_qq_36147908_article_details_107844959]

[https://blog.csdn.net/leexide/article/details/80035468][https_blog.csdn.net_leexide_article_details_80035468]

**目录**

第一章 环境

第二章 部署

2.1 服务器端口

2.2 RocketMq的安装

2.2.1 安装jdk并下载rocket

2.2.2 修改启动内存

2.2.3 修改broker启动配置

2.3 操作指令

2.4 效果

第三章 代码测试

3.1 生产者

3.2 消费者

3.3 配置

3.4 发送消息

3.5 效果

--------------------

# 第一章 环境 #

172.31.44.112现在是已经单点部署了RocketMq，172.31.44.113需要再新安装RocketMq。

<table> 
 <thead> 
  <tr> 
   <th>序号</th> 
   <th>IP地址</th> 
   <th>主机名</th> 
   <th>用户</th> 
   <th>角色</th> 
   <th>模式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td>172.31.44.112</td> 
   <td>rocketmq01</td> 
   <td>root</td> 
   <td>nameServer1,brokerServer1</td> 
   <td>Master1</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td>172.31.44.113</td> 
   <td>rocketmq02</td> 
   <td>root</td> 
   <td>nameServer2,brokerServer2</td> 
   <td>Master2</td> 
  </tr> 
 </tbody> 
</table>

我为了图方便把防火墙都关了

# 第二章 部署 #

## 2.1 服务器端口 ##

rocketMq所在服务器 一般来说要开启四个端口 9876,10909,10911,10912

9876 nameserver的端口  10911 broker对外的接口

特意查了与10909与10912端口的作用

开放10912，否则master的消息将无法复制到slave节点

10909是VIP通道对应的端口，在JAVA中的消费者对象或者是生产者对象中关闭VIP通道即可无需开放10909端口

centos7 防火墙操作api：

查看防火墙状态:systemctl status firewalld.service

关闭防火墙：systemctl stop firewalld.service

停止防火墙服务（避免开机重启）：systemctl disable firewalld.service

## 2.2 RocketMq的安装 ##

### 2.2.1 安装jdk并下载rocket ###

下载地址：[http://rocketmq.apache.org/docs/quick-start/][http_rocketmq.apache.org_docs_quick-start]

首先安装jdk，这个就不用说了。

下载后并解压，这里我解压的位置为：

/home/zchen/app/rocketmq

### 2.2.2 修改启动内存 ###

cd /home/zchen/app/rocketmq/bin
    vim runbroker.sh
    #将JAVA_OPT修改为
    JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn64m"
    
    vim runserver.sh
    #将JAVA_OPT修改为
    JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx128m -Xmn64m"

### 2.2.3 修改broker启动配置 ###

这个根据mq的不同，每个把名字改一下就行了

#所属集群名字
    brokerClusterName=rocketmq-cluster
    #broker名字，注意此处不同的配置文件填写的不一样
    brokerName=broker-1
    #0 表示 Master，>0 表示 Slave
    brokerId=0
    #nameServer地址，分号分割
    namesrvAddr=172.31.44.112:9876;172.31.44.113: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=/home/zchen/app/rocketmq/store
    #commitLog 存储路径
    storePathCommitLog=/home/zchen/app/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/home/zchen/app/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/home/zchen/app/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/home/zchen/app/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/home/zchen/app/rocketmq/store/abort
    #限制的消息大小
    maxMessageSize=125536
    #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

## 2.3 操作指令 ##

上面配置完就可以启动RocketMQ了

启动nameserver：nohup sh mqnamesrv &
    启动broker：nohup sh mqbroker -c /home/zchen/app/rocketmq/conf/2m-noslave/broker-a.properties > /dev/null 2>&1 & netstat -ntlp
    停止nameserver：sh mqshutdown namesrv
    停止broker：sh mqshutdown broker
    
    
    下面是我自己装的rocketmq可视化界面的启动命令，没装的可以不用管
    cd /home/zchen/app/rocketmq-externals/rocketmq-externals-master/rocketmq-console/target
    可视化：nohup java -jar rocketmq-console-ng-2.0.0.jar  > /dev/null 2>&1 &

创建队列(下面的命令大体是对的，但是可能哪里标点空格标点有问题，复制不行可以手敲一遍)

sh mqadmin updateTopic -n 172.31.44.112:9876 -b 172.31.44.112:10911 -t faceTopic

sh mqadmin updateTopic -n 172.31.44.112:9876 -c rocketmq-cluster  -t faceRecognitionTopic

sh mqadmin updateTopic -n 172.31.44.112:9876;172.31.44.113:9876 -c rocketmq-cluster  -t faceTopic（这个总是报命令错误，不用管，我在这里记录一下，以后处理）

![20201129185118104.png][]

参考：

![1cb7f31289b330195d248f3b5e76202d.png][]

![2a73fce80a86ebbef8043dd70ab28c85.png][]

## 2.4 效果 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3E2MTAzNzY2ODE_size_16_color_FFFFFF_t_70][]

# 第三章 代码测试 #

## 3.1 生产者 ##

@Configuration
    public class ProducerConfiguration {
        @Value("${test.rocketmq.producerGroup}")
        private String producerGroup;
        /**
         * NameServer 地址
         */
        @Value("${rocketmq.namesrvaddr}")
        private String namesrvAddr;
    
        @Bean
        public DefaultMQProducer defaultMQProducer(){
            DefaultMQProducer producer =null;
            try {
                producer = new DefaultMQProducer(producerGroup);
                producer.setNamesrvAddr(namesrvAddr);
                producer.start();
    
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            return producer;
        }
    
    }

## 3.2 消费者 ##

@Component
    public class ConsumerTest {
        /**
         * 消费者的组名
         */
        @Value("${test.rocketmq.conumerGroup}")
        private String consumerGroup;
    
        /**
         * NameServer 地址
         */
        @Value("${rocketmq.namesrvaddr}")
        private String namesrvAddr;
    
        @PostConstruct
        public void consumer() {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.setNamesrvAddr(namesrvAddr);
            try {
                consumer.subscribe("user-topic", "user-tag");
                consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                    try {
                        for (MessageExt messageExt : list) {
                            Thread.sleep(10000);
                            System.err.println("消费消息: " + new String(messageExt.getBody()));//输出消息内容
    //                        Integer currentInt = Integer.valueOf(new String(messageExt.getBody()));
    //                        if(currentInt%2==0){
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
                            /*}else{
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }*/
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
                });
                consumer.start();
            } catch (Exception e) {
                e.printStackTrace();
            }
    
        }
    
    }

## 3.3 配置 ##

#RocketMQ
    # 生产者的组名
    test.rocketmq.producerGroup=test-group
    # 消费者的组名
    test.rocketmq.conumerGroup=test-group
    # NameServer地址
    rocketmq.namesrvaddr=172.31.44.112:9876;172.31.44.113:9876

## 3.4 发送消息 ##

@RestController
    public class MessageTest {
        @Autowired
        DefaultMQProducer defaultMQProducer;
    
        @RequestMapping("/pushMessage")
        public String pushMessage(String message){
            if(message==null){
                message="zhangchen";
            }
            try {
                Message mqMessage = new Message("user-topic", "user-tag", message.getBytes());
                SendResult result = defaultMQProducer.send(mqMessage);
                System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());
            } catch (Exception e) {
                e.printStackTrace();
            }
            return message;
        }
    
    }

## 3.5 效果 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3E2MTAzNzY2ODE_size_16_color_FFFFFF_t_70 1][]

[https_blog.csdn.net_qq_36147908_article_details_107844959]: https://blog.csdn.net/qq_36147908/article/details/107844959
[https_blog.csdn.net_leexide_article_details_80035468]: https://blog.csdn.net/leexide/article/details/80035468
[http_rocketmq.apache.org_docs_quick-start]: http://rocketmq.apache.org/docs/quick-start/
[20201129185118104.png]: /images/20221120/33ce6c5691904fdc912607e5570c466d.png
[1cb7f31289b330195d248f3b5e76202d.png]: /images/20221120/7e4380806a1a41a7bebdb974a4dc9241.png
[2a73fce80a86ebbef8043dd70ab28c85.png]: /images/20221120/9cd1a57df41e4b8c9f234d4f1af58f06.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3E2MTAzNzY2ODE_size_16_color_FFFFFF_t_70]: /images/20221120/5a303e543498471b9e98a55563a0c1ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3E2MTAzNzY2ODE_size_16_color_FFFFFF_t_70 1]: /images/20221120/dc69b560a6f04bd4973aeafe3e059ac2.png