rocketmq学习(二) rocketmq集群部署与图形化控制台安装

时间:2024-01-25 16:21:20

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外,也能够从源码中学习到许多架构设计和编程实践相关的知识。

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