在Docker下部署RocketMQ - broker高可用部署

矫情吗；* 2022-03-12 04:38 568阅读 0赞

# 方案 #

因为borker本身需要存储消息信息，所以高可用方案选择主从方式。  
主从方式可以通过主从同步热备或主从异步热备两种方式。  
同步热备需要主从都将消息信息处理完成以后才返回，所以性能要差一些。  
异步热备就是主结点成功处理消息完成就返回，从结点通过内部的同步机制主结点消息数据。  
**本案例采用主从异步热备模式**。在RocketMQ默认的配置文件路径下有**$\{ROCKET\_HOME\}/conf/2m-2s-async/**目录路径，在该路径下有两主两从异步热备配置的模板。本案例因为只考虑高可用，不考虑高性能，所以只配置一主一从。只需要使用基于broker-a.properties和broker-a-s.properties两个模板文件进行修改配置。  
如果要考虑高性能，只需要将broker-b.properties做类似修改就可以了。如果竟考虑高可用，又考虑高性能，则配置从个broker-x.properties和broker-x-s.properties配置就可以了。

# 一、生成镜像 #

通过dockerfile    及命令

docker build autumn/nameser .

生成镜像image

# nameserver镜像生成脚本 #

# Start from a Java image.
    FROM java:8
    ARG version
    # Rocketmq version
    ENV ROCKETMQ_VERSION 4.4.0
    # Rocketmq home
    ENV ROCKETMQ_HOME  /opt/rocketmq-${ROCKETMQ_VERSION}
    WORKDIR  ${ROCKETMQ_HOME}
    RUN mkdir -p \
            /opt/logs \
            /opt/store \
            /opt/conf
    ADD rocketmq-all-4.4.0-bin-release.zip rocketmq.zip
    RUN       unzip rocketmq.zip \
              && rm rocketmq.zip \
              && cp -r rocketmq-all-4.4.0-bin-release/* ./
    RUN chmod +x bin/mqnamesrv
    CMD cd ${ROCKETMQ_HOME}/bin && export JAVA_OPT=" -Duser.home=/opt" && sh mqnamesrv
    EXPOSE 9876
    VOLUME /opt/logs \
            /opt/store \
            /opt/conf

# broker镜像生成脚本 #

注意：由于测试环境，内存不够，所以更改了runbroker.sh启动脚本中的mxm内存配置

# Start from a Java image.
    FROM java:8
    
    # Rocketmq version
    ENV ROCKETMQ_VERSION 4.4.0
    
    # Rocketmq home
    ENV ROCKETMQ_HOME  /opt/rocketmq-${ROCKETMQ_VERSION}
    
    WORKDIR  ${ROCKETMQ_HOME}
    
    RUN mkdir -p \
            /opt/logs \
            /opt/store
    
    ADD rocketmq-all-4.4.0-bin-release.zip rocketmq.zip
    RUN       unzip rocketmq.zip \
              && rm rocketmq.zip \ 
              && cp -r rocketmq-all-4.4.0-bin-release/* ./
    
    RUN chmod +x bin/mqbroker
    
    CMD cd ${ROCKETMQ_HOME}/bin && export JAVA_OPT=" -Duser.home=/opt" && sh mqbroker -n namesrv:9876 -c /opt/conf/broker.properties 
    
    EXPOSE 10909 10911
    VOLUME /opt/logs \
            /opt/store \
            /opt/conf

# console镜像生成脚本 #

这里是在Apache开源后的，该控制台的代码管理地址  
[https://github.com/apache/incubator-rocketmq-externals/tree/master/rocketmq-console][https_github.com_apache_incubator-rocketmq-externals_tree_master_rocketmq-console]  
基于工程源代码编译，生成工程jar包rocketmq-console-ng-1.0.0.jar （该工程是spring-boot工程，jar中内嵌servlet容器。）

FROM java:8
    VOLUME /tmp
    ADD rocketmq-console-ng-1.0.0.jar app.jar
    RUN sh -c 'touch /app.jar'
    ENV JAVA_OPTS=""
    ENTRYPOINT [ "sh", "-c", "java $JAVA_OPTS -jar /app.jar" ]

# 生产方及消息方镜像 #

这两个镜像比较简单，工程本身就是以main运行的java工程。  
可以参考该文章：[https://blog.csdn.net/liulianglin/article/details/51787459][https_blog.csdn.net_liulianglin_article_details_51787459]

FROM java:8
    VOLUME /tmp
    #ADD mq-consumer.jar app.jar
    ADD mq-producer.jar app.jar
    RUN sh -c 'touch /app.jar'
    ENV JAVA_OPTS=""
    ENTRYPOINT [ "sh", "-c", "java $JAVA_OPTS -jar /app.jar" ]

# 二、配置文件 #

# borker主结点配置 #

broker-a-m:
        imag going/rocketmq-broker:4.2.1
        ports:
          - 10909:10909
          - 10911:10911
        volumes:
          - "/rocketmq/broker/a-m/logs:/opt/logs"
          - "/rocketmq/broker/a-m/stor/opt/store"
          - "/rocketmq/broker/a-m/conf:/opt/conf"
        links:
          - namesrv:namesrv

borker配置文件为

brokerClusterName=DefaultCluster
    brokerName=broker-a
    brokerId=0
    deleteWhen=04
    fileReservedTime=48
    brokerRole=ASYNC_MASTER
    flushDiskType=ASYNC_FLUSH

# borker从结点配置 #

broker-a-s:
        imag going/rocketmq-broker:4.2.1
        ports:
          - 11909:10909
          - 11911:10911
        volumes:
          - "/rocketmq/broker/a-s/logs:/opt/logs"
          - "/rocketmq/broker/a-s/stor/opt/store"
          - "/rocketmq/broker/a-s/conf:/opt/conf"
        links:
          - namesrv:namesrv

borker配置文件为

brokerClusterName=DefaultCluster
    brokerName=broker-a
    brokerId=1
    deleteWhen=04
    fileReservedTime=48
    brokerRole=SLAVE
    flushDiskType=ASYNC_FLUSH

# 三、docker-mopose #

version: '2.1'
    services:
      namesrv:
        image: autumn/rocketmq-namesrv:latest
        ports:
          - 9876:9876
        volumes:
          - "./logs/namesrv/master/logs:/opt/logs"
          - "./data/namesrv/master/stor:/opt/store"
      broker-a-m:
        image: autumn/rocketmq-broker:latest
        ports:
          - 10909:10909
          - 10911:10911
        volumes:
          - "./logs/broker/a-m/logs:/opt/logs"
          - "./data/broker/a-m/stor:/opt/store"
          - "./conf/broker/a-m/conf:/opt/conf"
        links:
          - namesrv:namesrv
      broker-a-s:
        image: autumn/rocketmq-broker:latest
        ports:
          - 11909:11909
          - 11911:11911
        volumes:
          - "./logs/broker/a-s/logs:/opt/logs"
          - "./data/broker/a-s/stor:/opt/store"
          - "./conf/broker/a-s/conf:/opt/conf"
        links:
          - namesrv:namesrv      
      console:
        image: autumn/rocketmq-console-ng:latest
        ports:
         - "8087:8080"
        links:
         - namesrv:namesrv
        environment:
         JAVA_OPTS: -Drocketmq.config.namesrvAddr=namesrv:9876 #-Dcom.rocketmq.sendMessageWithVIPChannel=false

通过docker-compose运行，打开console界面，查看执行情况  
http://127.0.0.1:8087/\#/

相关代码文件 [https://github.com/ffzhihua/rocketmq-docker-compose][https_github.com_ffzhihua_rocketmq-docker-compose]

# 四、在eclipse中开发测试 #

1.创建一个maven项目，其pom.xml的内容见pom.xml

2. 编写消息产生者Producer,见Producer.java

3.编写消息消费者Consumer

PushConsumer.java

package com.catt.rocketmq.example;
  
 import java.util.concurrent.TimeUnit;
  
 import com.alibaba.rocketmq.client.exception.MQClientException;
 import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
 import com.alibaba.rocketmq.client.producer.SendResult;
 import com.alibaba.rocketmq.common.message.Message;
  
 public class Producer {
   public static void main(String[] args) throws MQClientException,
       InterruptedException {
     /**
      * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例 
      * 注意：ProducerGroupName需要由应用来保证唯一 
      * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
      * 因为服务器会回查这个Group下的任意一个Producer
      */
     DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
     producer.setNamesrvAddr("192.168.230.128:9876");
     producer.setInstanceName("Producer");
  
     /**
      * Producer对象在使用之前必须要调用start初始化，初始化一次即可 
      * 注意：切记不可以在每次发送消息时，都调用start方法
      */
     producer.start();
  
     /**
      * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
      * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态， 
      * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高， 
      * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
      */
     for (int i = 0; i < 100; i++) {
       try {
         {
           Message msg = new Message("TopicTest1",// topic
               "TagA",// tag
               "OrderID001",// key
               ("Hello MetaQ").getBytes());// body
           SendResult sendResult = producer.send(msg);
           System.out.println(sendResult);
         }
         {
           Message msg = new Message("TopicTest2",// topic
               "TagB",// tag
               "OrderID0034",// key
               ("Hello MetaQ").getBytes());// body
           SendResult sendResult = producer.send(msg);
           System.out.println(sendResult);
         }
         {
           Message msg = new Message("TopicTest3",// topic
               "TagC",// tag
               "OrderID061",// key
               ("Hello MetaQ").getBytes());// body
           SendResult sendResult = producer.send(msg);
           System.out.println(sendResult);
         }
       } catch (Exception e) {
         e.printStackTrace();
       }
       TimeUnit.MILLISECONDS.sleep(1000);
     }
     /**
      * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
      * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
      */
     producer.shutdown();
   }
 }

PushConsumer.java

package com.catt.rocketmq.example;
  
 import java.util.List;
  
 import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
 import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
 import com.alibaba.rocketmq.client.exception.MQClientException;
 import com.alibaba.rocketmq.common.message.MessageExt;
 public class PushConsumer {
  
   /**
    * 当前例子是PushConsumer用法，使用方式给用户感觉是消息从RocketMQ服务器推到了应用客户端。 
    * 但是实际PushConsumer内部是使用长轮询Pull方式从MetaQ服务器拉消息，然后再回调用户Listener方法 
    */
   public static void main(String[] args) throws InterruptedException,
       MQClientException {
     /**
      * 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例 
      * 注意：ConsumerGroupName需要由应用来保证唯一
      */
     DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
         "ConsumerGroupName");
     consumer.setNamesrvAddr("192.168.230.128:9876");
     consumer.setInstanceName("Consumber");
  
     /**
      * 订阅指定topic下tags分别等于TagA或TagC或TagD
      */
     consumer.subscribe("TopicTest1", "TagA || TagC || TagD");
     /**
      * 订阅指定topic下所有消息 
      * 注意：一个consumer对象可以订阅多个topic
      */
     consumer.subscribe("TopicTest2", "*");
  
     consumer.registerMessageListener(new MessageListenerConcurrently() {
  
       /**
        * 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
        */
       @Override
       public ConsumeConcurrentlyStatus consumeMessage(
           List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
         System.out.println(Thread.currentThread().getName()
             + " Receive New Messages: " + msgs.size());
  
         MessageExt msg = msgs.get(0);
         if (msg.getTopic().equals("TopicTest1")) {
           // 执行TopicTest1的消费逻辑
           if (msg.getTags() != null && msg.getTags().equals("TagA")) {
             // 执行TagA的消费
             System.out.println(new String(msg.getBody()));
           } else if (msg.getTags() != null
               && msg.getTags().equals("TagC")) {
             // 执行TagC的消费
           } else if (msg.getTags() != null
               && msg.getTags().equals("TagD")) {
             // 执行TagD的消费
           }
         } else if (msg.getTopic().equals("TopicTest2")) {
           System.out.println(new String(msg.getBody()));
         }
  
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
       }
     });
  
     /**
      * Consumer对象在使用之前必须要调用start初始化，初始化一次即可 
      */
     consumer.start();
  
     System.out.println("Consumer Started.");
   }
 }

[https_github.com_apache_incubator-rocketmq-externals_tree_master_rocketmq-console]: https://github.com/apache/incubator-rocketmq-externals/tree/master/rocketmq-console
[https_blog.csdn.net_liulianglin_article_details_51787459]: https://blog.csdn.net/liulianglin/article/details/51787459
[https_github.com_ffzhihua_rocketmq-docker-compose]: https://github.com/ffzhihua/rocketmq-docker-compose