zoukankan      html  css  js  c++  java
  • rocketmq安装部署过程(4.0.0版本)

    • 准备工作

    • 3个虚拟机节点的构成如下

    • 安装步骤

    • 操作过程

    1、安装包已经上传至其中1个节点。

    2、解压缩安装包

    命令:unzip rocketmq-all-4.0.0-incubating-bin-release.zip

    解压缩之后如下:

    3、 我这里将解压缩之后的文件夹移动了位置,并修改了名字,以便后续操作。

    命令: mv /home/hadmin/software/apache-rocketmq-all/ /home/hadmin/rocketmq

    移动之后路径如下:

     4、修改配置文件

    我这里已经将配置文件提前准备好了,只呈现以下配置文件的结果。

    默认配置可以参考源码:

    https://github.com/apache/rocketmq/blob/master/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java

    Tips1:

    autoCreateTopicEnable,建议线下开启,线上关闭

    Tips2:

    同一个机器上有多个Broker时,各个Broker的文件路径要独立。

    ■节点1(192.168.6.3)配置文件:

    《broker-a-m.properties》

    brokerClusterName=post
    brokerName=broker-a
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=0
    listenPort=10911
    brokerIP1=192.168.6.3
    deleteWhen=04
    fileReservedTime=72
    brokerRole=ASYNC_MASTER
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-a-m
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-a-m
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

    《broker-c-s.properties》 

    brokerClusterName=post
    brokerName=broker-c
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=1
    listenPort=10920
    brokerIP1=192.168.6.3
    deleteWhen=04
    fileReservedTime=72
    brokerRole=SLAVE
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-c-s
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-c-s
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

    ■节点2(192.168.6.4)配置文件:

    《broker-a-s.properties》

    brokerClusterName=post
    brokerName=broker-a
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=1
    listenPort=10920
    deleteWhen=04
    brokerIP1=192.168.6.4
    fileReservedTime=72
    brokerRole=SLAVE
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-a-s
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-a-s
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

    《broker-b-m.properties》

    brokerClusterName=post
    brokerName=broker-b
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=0
    listenPort=10911
    brokerIP1=192.168.6.4
    deleteWhen=04
    fileReservedTime=72
    brokerRole=ASYNC_MASTER
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-b-m
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-b-m
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

    ■节点3(192.168.6.5)配置文件:

    《broker-b-s.properties》 

    brokerClusterName=post
    brokerName=broker-b
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=1
    listenPort=10920
    brokerIP1=192.168.6.5
    deleteWhen=04
    fileReservedTime=72
    brokerRole=SLAVE
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-b-s
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-b-s
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

    《broker-c-m.properties》 

    brokerClusterName=post
    brokerName=broker-c
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=0
    listenPort=10911
    brokerIP1=192.168.6.5
    deleteWhen=04
    fileReservedTime=72
    brokerRole=ASYNC_MASTER
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-c-m
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-c-m
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

    5、启动nameserver

    启动命令:nohup sh /home/hadmin/rocketmq/bin/mqnamesrv >/home/hadmin/rocketmq/logs/mqnamesrv.log 2>&1 &

    注意:我这里将启动日志重定义到了logs路径下,需要提前手动创建logs文件夹,以便于统一管理日志,方便查看。

    创建文件夹命令:mkdir /home/hadmin/rocketmq/logs

    下图中展示了启动命令,日志中可以看到NameServer成功启动的日志。

    ■NameServer - 节点1

    ■NameServer = 节点2

    6、启动Broker-a(Master位于节点1、Slave位于节点2)

    broker-a分为Master和Slave,分别位于节点1和节点2上,需要分别启动。

    注意:需要根据启动角色,为broker指定一个配置文件。

    ■broker-a的master - 节点1

    命令: nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-a-m.properties >/home/hadmin/rocketmq/logs/broker-a-m.log 2>&1 &

     

    ■broker-a的slave - 节点2

    命令:nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-a-s.properties >/home/hadmin/rocketmq/logs/broker-a-s.log 2>&1 &

    ■验证broker-a:

    broker-a启动结束,这时候可以使用命令查看一下rocketmq集群状态。

    命令:sh /home/hadmin/rocketmq/bin/mqadmin clusterList -n 192.168.6.3:9876

    7、启动broker-b(Master位于节点2,、Slave位于节点3)

    ■broker-b的Master - 节点2

    命令:nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-b-m.properties >/home/hadmin/rocketmq/logs/broker-b-m.log 2>&1 &

    ■broker-b的Slave - 节点3

    命令:nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-b-s.properties >/home/hadmin/rocketmq/logs/broker-b-s.log 2>&1 &

    ■验证broker-b

    命令:sh /home/hadmin/rocketmq/bin/mqadmin clusterList -n 192.168.6.3:9876

    8、 启动broker-b(Master位于节点3,、Slave位于节点1)

    ■broker-c的Master - 节点3

    命令:nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-c-m.properties >/home/hadmin/rocketmq/logs/broker-c-m.log 2>&1 &

    ■broker-c的Slave - 节点1

    命令:nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-c-s.properties >/home/hadmin/rocketmq/logs/broker-c-s.log 2>&1 &

    ■验证broker-c

    命令:sh /home/hadmin/rocketmq/bin/mqadmin clusterList -n 192.168.6.3:9876

    • 问题1:

    启动broker的时候提示内存不够的错误。 

    解决方法:由于个人电脑配置不够,无法为虚拟机申请更大的内存。所以,采用修改broker启动内存的方式解决了。

    修改文件路径:{ROCKET_HOME}/bin/runbroker.sh

    修改前:JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"

    修改后:JAVA_OPT="${JAVA_OPT} -server -Xms1024m -Xmx1536m -Xmn1546m"

    如图所示:

    重新启动broker,日志中没有错误,jps进程中也可以看到broker正常启动。

    • 问题2:

    在同一台机器上启动多个broker的时候提示如下错误。

    问题原因:

    按照本文最开始的额图片所示,同一台机器上会存在两个Broker,如果不进行特殊指定,broker的默认端口是10911。

    所以一台机器上启动两个broker时,第二个broker就会出现端口被占用的错误。

    解决办法:

    修改rocketmq的配置文件,增加listenPort配置。配置之后如下所示:

    →同一台机器的配置文件

    →broker-a-m.properties

    →broker-c-s.properties

    使用jps查看一下进程是否有问题

    在查看一下启动日志是否有问题

    最后使用clusterList命令来验证一下集群健康状态。

    命令:sh bin/mqadmin clusterList -n 192.168.6.3:9876

      成功实现了,如本文最开始图片所示的安装部署。

    • 问题3:

    同一台机器上启动多个broker,在启动第二个broker的时候报错

    RocketMq Lock failed,MQ already started;

    解决办法:

    路径冲突了,修改各种路径即可。我这版修改了一下5个路径之后,解决。

    storePathRootDir=/home/radmin/data/rocketmq/rootdir-b-m
    storePathCommitLog=/home/radmin/data/rocketmq/commitlog-b-m
    storePathConsumerQueue=/home/radmin/data/rocketmq/consumequeue-b-m
    storePathIndex=/home/radmin/data/rocketmq/index-b-m
    storeCheckpoint=/home/radmin/data/rocketmq/checkpoint-b-m
    • 使用程序测试

    编写了java程序代码尝试向集群中生产消息,程序代码如下:

    需要注意的是,如果autoCreateTopicEnable=false的时候,需要自己手动创建Topic。

    命令样例:sh $MQ_HOME/bin/mqadmin updateTopic -c post -n "192.168.1.80:9876;192.168.1.81:9876" -t QCH2 -r 4 -w 4

    package post;
    
    import org.apache.rocketmq.client.exception.MQBrokerException;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.exception.RemotingException;
    
    import java.util.UUID;
    
    public class ProducerTest {
        private static DefaultMQProducer producer = null;
    
        public static void main(String[] args) {
            System.out.print("[----------]Start");
            boolean result = false;
            try {
                ProducerStart();
                SendMessage("qch_20170706", "hello rocketmq!");
            }finally {
                producer.shutdown();
            }
            System.out.print("[----------]Succeed");
        }
    
        private static boolean ProducerStart() {
            producer = new DefaultMQProducer("pro_qch_test");
            producer.setNamesrvAddr("192.168.6.3:9876;192.168.6.4:9876");
            producer.setInstanceName(UUID.randomUUID().toString());
            producer.setVipChannelEnabled(false);
            try {
                producer.start();
            } catch (MQClientException e) {
                e.printStackTrace();
                return false;
            }
            return true;
        }
    
        private static boolean SendMessage(String topic, String str) {
            Message msg = new Message(topic, str.getBytes());
            try {
                producer.send(msg);
            } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
                e.printStackTrace();
                return false;
            }
            return true;
        }
    }
    

    运行之后,日志中提示下面的错误:

    问题原因:

    因为broker部署在虚拟机,并且虚拟双网卡,client无法正常连接服务端。

    解决方法:

    可以在broker的配置文件中配置brokerIP1(本机IP)属性。

    修改后配置文件如下图所示:

    修改之后,重新启动rocketmq集群,运行生产者程序,确认正常结束。

    然后,有尝试这编写消费者代码,验证是否可以正常消费。结果正常,这里贴一下代码及结果日志。

    package post;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.MessageExt;
    
    import java.util.List;
    import java.util.UUID;
    
    public class ConsumerTest {
        public static void main(String[] args) {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("con_qch_test");
            consumer.setInstanceName(UUID.randomUUID().toString());
            consumer.setConsumeMessageBatchMaxSize(32);
            consumer.setNamesrvAddr("192.168.6.3:9876;192.168.6.4:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(
                        List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for(MessageExt me : list) {
                        System.out.print(new String(me.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            try {
                consumer.subscribe("qch_20170706", "*");
                consumer.start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    

    样例代码上传到了git上了,有需要的可以去参考。

    https://github.com/quchunhui/rocketmq_sample

    --END--

  • 相关阅读:
    Digital Square(hdu4394)搜索
    Substrings(hdu1238)字符串匹配
    Graveyard(poj3154)
    数据库的范式理论
    模和除
    You can Solve a Geometry Problem too (hdu1086)几何,判断两线段相交
    对vector等STL标准容器进行排序操作(转!)
    0 or 1(hdu2608)数学题
    LintCode 4.丑数
    LintCode 9.Fizz Buzz问题
  • 原文地址:https://www.cnblogs.com/quchunhui/p/7126868.html
Copyright © 2011-2022 走看看