zoukankan      html  css  js  c++  java
  • rocketmq学习(二) rocketmq集群部署与图形化控制台安装

    1.rocketmq图形化控制台安装

      虽然rocketmq为用户提供了使用命令行管理主题、消费组以及broker配置的功能,但对于不够熟练的非运维人员来说,命令行的管理界面还是较难使用的。为此,我们可以使用图形化的管理界面来简化管理操作。

      rocketmq官方推荐的图形化控制台目前还处在不成熟的孵化阶段。仓库地址为(https://github.com/apache/rocketmq-externals),其中包含了rocketmq相关拓展的、属于孵化期的各种项目。下载源码之后,找到rocketmq-console文件夹,这就是rocketmq官方推荐的图形化控制台项目,基于springboot和angularJS。

      打开application.properties,能看到一些重要参数的配置,例如端口,nameServer地址,登录权限控制等等。对于启动参数的设置,可以选择直接在配置文件中修改;也可在启动项目时通过命令行指定。

      

      为部署项目,先执行maven的打包命令(mvn clean package),生成jar包。

      然后执行java -jar rocketmq-console-ng-1.0.1.jar --rocketmq.config.namesrvAddr=localhost:9876(nameServer地址) --server.port=8080(启动端口)。

      通过浏览器访问项目启动的ip/port,即可看到以下管理界面(右上角可以中英文切换)。

       至此,rocketmq图形化控制台安装成功。

    2.rocketmq集群部署

      rocketmq的单机部署虽然简单方便,却存在着单点故障的问题。通过集群部署nameServer和broker可以实现rocketmq服务端的高可用。

      下面介绍rocketmq的集群部署,以在两台机器上搭建一个双主双从的高可用rocketmq集群为例子,这两台机器(linux环境)的IP地址分别是192.168.32.130,和192.168.32.131。

    2.1 nameServer集群部署

      由于nameServer的集群节点之间互不通信,所以不需要额外的配置。

      在两台机器的rocketmq安装的根路径下分别执行"sh bin/mqnamesrv",各启动一个nameServer,使用其默认的端口9876。

      现在,192.168.32.130:9876和192.168.32.131:9876上都各运行着一个nameServer服务。

    2.2 broker集群部署

      broker作为rocketmq的核心,其运行的稳定性至关重要。前面提到的双主双从(2-master-2-slave)实际上指代的是broker的集群运行模式,从broker作为主broker的备份,负责和主broker保持数据同步,可读不可写。

      rocketmq通过赋予broker名称来区别不同角色的broker。我们把当前两个角色的broker分别命名为broker-abroker-b。

      在192.168.32.130 部署broker-a的主和broker-b的从,在192.168.32.131 部署broker-b的主和broker-a的从,同一角色的主从分别部署在不同的机器上。这样,即使任意一台机器挂掉,由于从broker的存在,broker-a和broker-b依然可以对外提供服务。

      启动broker时,可以通过指定配置文件的方式为broker设置一系列参数。不同角色的broker,主从broker之间的配置文件参数内容各不相同。

    broker配置文件中参数介绍:

      brokerName:broker名称,互为主从的broker名称保持一致。

      namesrvAddr:关联的nameserver地址,多个用";"隔开。

      listenPort:broker监听端口,同一机器部署多个broker不能监听端口不能相同,避免冲突

      storePathRootDir:broker存储数据的根目录

      brokerClusterName:broker集群名称,相同集群的master能互相识别

      brokerId:0代表master,大于0代表不同的slave-broker

      deleteWhen:删除过时消息的时间,04代表凌晨4点

      fileReservedTime:落盘数据文件保存的时长,单位小时

      brokerRole:brokerRole有三种类型,SYNC_MASTERASYNC_MASTERSLAVE,SYNC_M和ASYNC_M都代表主broker,区别在于主从之间进行数据同步的方式不同。SYNC代表主从数据同步完成,才向客户端返回消息发送成功结果;反之ASYNC代表主库收到消息后立即返回发送消息成功结果。

      可以看到,ASYNC_MASTER的效率更高,但是当MASTER出现故障时,可能出现消息丢失的问题。需要用户进行效率与可靠性之间的取舍。

      flushDiskType:flushDiskType有两种类型,SYNC_FLUSH(同步刷盘)ASYNC_FLUSH(异步刷盘),用于指定broker在接收到消息后,返回消息发送结果和数据落盘处理的策略。当选择同步刷盘时,只有当消息数据真正的写入磁盘持久化时,才返回消息发送成功。选择异步刷盘时,消息数据写入本地虚拟内存映射后,就直接返回。

      broker的本地落盘策略和主从同步策略的选择类似,都需要在效率与可靠性、一致性之间进行取舍。

      比较推荐的一种配置方案是主从同步策略选择SYNC_MASTER而本地落盘策略选择SYNC_FLUSH。从可靠性的角度来看,只要主从broker没有同时挂掉(避免了单点故障),消息将不会丢失;从效率的角度来看,由于主从broker都是异步落盘,执行效率也有一定的保障,是一个优秀的折中方案。

    broker配置文件详情:

    broker-a主broker  配置文件(broker-a.properties):

    brokerName=broker-a
    namesrvAddr=192.168.32.130:9876;192.168.32.131:9876
    listenPort=10911
    storePathRootDir=/root/rocketmq/data/store/store-a
    brokerClusterName=DefaultCluster
    brokerId=0
    deleteWhen=04
    fileReservedTime=48
    brokerRole=SYNC_MASTER
    flushDiskType=ASYNC_FLUSH

    broker-b主broker  配置文件(broker-b.properties):

    brokerName=broker-b
    namesrvAddr=192.168.32.130:9876;192.168.32.131:9876
    listenPort=10911
    storePathRootDir=/root/rocketmq/data/store/store-b
    brokerClusterName=DefaultCluster
    brokerId=0
    deleteWhen=04
    fileReservedTime=48
    brokerRole=SYNC_MASTER
    flushDiskType=ASYNC_FLUSH

    broker-a从broker  配置文件(broker-a-s.properties):

    brokerName=broker-a
    namesrvAddr=192.168.32.130:9876;192.168.32.131:9876
    listenPort=11011
    storePathRootDir=/root/rocketmq/data/store/store-a
    brokerClusterName=DefaultCluster
    brokerId=1
    deleteWhen=04
    fileReservedTime=48
    brokerRole=SLAVE
    flushDiskType=ASYNC_FLUSH

    broker-b从broker  配置文件(broker-b-s.properties):

    brokerName=broker-b
    namesrvAddr=192.168.32.130:9876;192.168.32.131:9876
    listenPort=11011
    storePathRootDir=/root/rocketmq/data/store/store-b
    brokerClusterName=DefaultCluster
    brokerId=1
    deleteWhen=04
    fileReservedTime=48
    brokerRole=SLAVE
    flushDiskType=ASYNC_FLUSH

       在对应机器依次执行以下命令,依次启动broker(先启动主broker,后启动从broker)。

      192.168.32.130执行:sh bin/mqbroker -c [配置文件路径 eg: rocketmq/data/conf/broker-a.properties]

      192.168.32.131执行:sh bin/mqbroker -c [配置文件路径 eg: rocketmq/data/conf/broker-b.properties]

      192.168.32.131执行:sh bin/mqbroker -c [配置文件路径 eg: rocketmq/data/conf/broker-a-s.properties]

      192.168.32.130执行:sh bin/mqbroker -c [配置文件路径 eg: rocketmq/data/conf/broker-b-s.properties]

      此时,rocketmq双主双从的broker集群已经搭建完毕。启动图形控制台,指定命令行参数namesrvAddr=192.168.32.130:9876;192.168.32.131:9876,可看到以下信息:

    3. 使用java客户端收发消息

      前面我们通过命令行的方式进行了rocketmq收发消息的测试。但在实际使用过程中,还是需要sdk客户端来进行收发消息。这里,我们使用rocketmq提供的java sdk来进行rocketmq的消息收发实验。

      先通过图形控制台创建一个主题用于测试,主题名称为"TopicTest"(随便取的)。

      接着启动一个java项目,加入rocketmq-client的依赖。

    maven坐标:

    <!-- 原生rocketmq client -->
    <dependency>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-client</artifactId>
         <version>4.4.0</version>
    </dependency>

    生产者producer示例代码:

    public class Producer {
        public static void main(String[] args) throws Exception {
            final DefaultMQProducer producer = new DefaultMQProducer("test_producer_group");
            // 设置nameServer地址
            producer.setNamesrvAddr("192.168.32.130:9876;192.168.32.131:9876");
            producer.start();
            for (int i = 0; i < 1000; i++) {
                try {
                    // 构造消息对象,topic=TopicTest,tag=TagA
                    Message msg = new Message("TopicTest", "TagA" ,
                        ("Hello RocketMQ TopicTest" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // 发送消息
                    SendResult sendResult = producer.send(msg);
                    System.out.printf("%s%n", sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                    Thread.sleep(1000);
                }
            }
            producer.shutdown();
        }
    }

    消费者consumer示例代码:

    public class Consumer {
        public static void main(String[] args) throws Exception {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test_consumer_group");
            // 设置nameServer地址
            consumer.setNamesrvAddr("192.168.32.130:9876;192.168.32.131:9876");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            // 订阅主题 tag="*"代表订阅TopicTest主题下所有子主题消息
            consumer.subscribe("TopicTest", "*");
            // 注册消息监听回调函数
            consumer.registerMessageListener((MessageListenerConcurrently)(msgs, context) -> {
                for(MessageExt messageExt : msgs){
                    String strMsg = new String(messageExt.getBody());
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), strMsg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }
    }

      生产者和消费者都是简单的main方法启动,先启动producer发送消息,然后启动consumer接受消息,控制台上将会看到接受时消息的日志。可以试试主动关闭一个master-broker,看看broker集群的消息收发是否正常。

      至此,通过java客户端使用rocketmq的测试告一段落。

    总结

      本篇博客介绍了rocketmq的集群部署,图形化界面的安装以及如何使用java客户端与rocketmq进行交互。rocketmq还有着许多好用,强大的功能,后续的博客将结合着rocketmq的源码来介绍它们。

      去阅读并理解源码,可以在解决问题时能看得更深、更远。通过阅读rocketmq的源码,除了更好地掌握rocketmq外,也能够从源码中学习到许多架构设计和编程实践相关的知识。

      如有理解不到位的地方,请多多指教。

  • 相关阅读:
    Hibernate知识回顾
    Lucene搜索过程的核心类
    JSP知识回顾
    Java Swing 知识回顾
    Spring知识回顾
    javaDocking 学习
    Lucene搜索范围
    MINA 学习
    java 阿拉伯人民币转换为中文
    oracle提高查询效率方法
  • 原文地址:https://www.cnblogs.com/xiaoxiongcanguan/p/11579300.html
Copyright © 2011-2022 走看看