RocketMQ系列之简单例子

缺乏、安全感 2021-09-15 05:20 518阅读 0赞

### 前言 ###

继我上一篇博客后  
[分布式消息队列RocketMQ学习教程①][RocketMQ]  
上一篇博客最主要介绍了几种常用的MQ，所以本博客再简单介绍一下RocketMQ的原理和简单的例子，基于Java实现，希望可以帮助学习者

### RoketMQ搭建Linux版 ###

“工于利其事，必先利其器”，所以我们首先需要搭建好RocketMQ，  
考虑到学习者不一定有Linux系统的服务器，所以本博客介绍一下Linux和Window系统的两种安装方法，以补充上一篇博客

因为阿里已经将RocketMQ捐给Apache了，所以现在我们需要去Apache官网下载  
[RocketMQ官网][RocketMQ 1]

注意RocketMQ是基于Java开发的，所以安装前必须安装JDK，Linux JDK安装的可以看[分布式消息队列RocketMQ学习教程①][RocketMQ]  
下载文件解压后，可以看到conf文件夹里有2m-noslave、2m-2s-async、2m-2s-sync文件夹

2m-noslave 两主,无从的配置

2m-2s-async 两主，两从，同步复制数据的配置

2m-2s-sync 两主，两从，异步复制数据的配置

我们找到2m-noslave的broker-a.properties文件，修改完善配置  
broker-a.properties

#所属集群名字  
    brokerClusterName=DefaultCluster
    #broker名字,注意此处不同的配置文件填写的不一样
    brokerName=broker-a
    #0 表示 Master,>0 表示 Slave
    brokerId=0
    #nameServer地址,分号分割
    namesrvAddr=127.0.0.1:9876
    
    #关键
    brokerIP1=127.0.0.1
    
    #在发送消息时,自动创建服务器不存在的topic,默认创建的队列数
    defaultTopicQueueNums=4
    #是否允许 Broker 自动创建Topic,建议线下开启,线上关闭
    autoCreateTopicEnable=true
    #是否允许 Broker 自动创建订阅组,建议线下开启,线上关闭
    autoCreateSubscriptionGroup=true
    #Broker 对外服务的监听端口
    listenPort=10911
    #删除文件时间点,默认凌晨 4点
    deleteWhen=04
    #文件保留时间,默认 48 小时
    fileReservedTime=48
    #commitLog每个文件的大小默认1G
    mapedFileSizeCommitLog=1073741824
    #ConsumeQueue每个文件默认存30W条,根据业务情况调整
    mapedFileSizeConsumeQueue=300000
    #destroyMapedFileIntervalForcibly=120000
    #redeleteHangedFileInterval=120000
    #检测物理文件磁盘空间
    diskMaxUsedSpaceRatio=88
    
    #这里是我的 日志配置
    #存储路径
    storePathRootDir=/usr/local/rocketmq/store
    #commitLog 存储路径
    storePathCommitLog=/usr/local/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/usr/local/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/usr/local/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/usr/local/rocketmq/store/abort
    
    
    #限制的消息大小  
    maxMessageSize=65536
    #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

先介绍一下linux系统的  
一般将压缩文件解压到/usr/local

cd /usr/local
    
    tar -xzf apache-rocketmq.tar.gz
    
    mv apache-rocketmq rocketmq
    
    mkdir /usr/rocketmq/logs

环境变量配置

vim /etc/profile

修改如下配置

export JAVA_HOME=/usr/java/jdk1.8.0_102
    export ROCKETMQ_HOME=/usr/local/rocketmq
    export PATH=$PATH:$JAVA_HOME/bin:/usr/local/src/redis-3.2.8/bin:$ROCKETMQ_HOME/bin:$PATH
    export CLASSPATH=.:$JAVA_HOME/lib

启动mqnamesrv

cd /usr/local/rocketmq/bin
    nohup sh /usr/local/rocketmq/bin/mqnamesrv >/usr/local/rocketmq/logs/mqnamesrv.log 2>&1 &

启动Broker

nohup sh /usr/local/rocketmq/bin/mqbroker  -c /usr/local/rocketmq/conf/2m-noslave/broker-a.properties > /usr/local/rocketmq/logs/mqbroker.log  2>&1 &

要设置自动创建Topic，需要加上  
autoCreateTopicEnable=true

关闭Broker服务  
sh mqshutdown broker

启动成功可以用jps查看  
![这里写图片描述][70]

### RocketMQ搭建Window版 ###

1、下载RocketMQ后，解压到D:\\alibaba-rocketmq

2、在D:\\alibaba-rocketmq，Ctrl+Shift，右键，打开dom界面，输入如下命令行  
start /b bin/mqnamesrv.exe >D:\\alibaba-rocketmq\\logs\\mqnamesrv.log  
查看nameserver是否启动  
jps -v

3、启动Broker

start /b bin/mqbroker.exe -n "127.0.0.1:9876" autoCreateTopicEnable=true >D:\alibaba-rocketmq\logs\mqbroker.log

Caused by: com.alibaba.rocketmq.client.exception.MQClientException: No route info of this topic, huang_1
    See http://docs.aliyun.com/cn#/pub/ons/faq/exceptions&topic_not_exist for further details.
    	at com.alibaba.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:525) ~[rocketmq-client-3.5.3.jar:na]
    	at com.alibaba.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1011) ~[rocketmq-client-3.5.3.jar:na]
    	at com.alibaba.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:970) ~[rocketmq-client-3.5.3.jar:na]
    	at com.alibaba.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:90) ~[rocketmq-client-3.5.3.jar:na]
    	at com.aliyun.openservices.ons.api.impl.rocketmq.ProducerImpl.send(ProducerImpl.java:107) ~[ons-client-1.2.3.jar:na]

出现以上异常启动时添加autoCreateTopicEnable=true

4、查看topic命令：mqadmin topicList -n “127.0.0.1:9876”

cd 到bin目录，执行下面命令  
mqadmin updateTopic -t test\_1 -b “127.0.0.1:10911” -n “127.0.0.1:9876”  
添加如下参数到eclipse启动工程的VM参数里  
\-Drocketmq.namesrv.addr=127.0.0.1:9876

### RocketMq监控平台搭建 ###

需要去github下载，下载链接  
[rocketmq-console][]

下载后在rocketmq-console文件夹里，ctrl+shift，右键，在此处打开命令窗口，打开cmd窗口，主要要先搭建好maven环境

mvn clean package -Dmaven.test.skip=true

打包完成之后，我们去target文件夹找到rocketmq-console-ng-1.0.0.jar  
然后

mkdir rocketmq-console
    cd /usr/local/rocketmq-console

使用xftp上传rocketmq-console-ng-1.0.0.jar到/usr/local/rocketmq-console

nohup java -jar rocketmq-console-ng-1.0.0.jar --server.port=12581 --rocketmq.config.namesrvAddr=127.0.0.1:9876 >/usr/local/rocketmq-console/run.log 2>&1 &

端口检查

netstat -anp|grep 12581

部署成功，打开http://服务器IP:12581

![这里写图片描述][70 1]

### 编程实现MQ实例 ###

maven加入配置

<dependency>
                <groupId>com.alibaba.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>3.0.10</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.rocketmq</groupId>
                <artifactId>rocketmq-all</artifactId>
                <version>3.0.10</version>
                <type>pom</type>
            </dependency>

消息队列消费者消费消息实例

package com.mq.test;
    
    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;
    
    import java.util.List;
    
    public class MQConsumer {
        public static void main(String[] args) throws MQClientException {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
                    "mq-group");
    
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.setInstanceName("consumer");
            consumer.subscribe("TopicA-test", "TagA");
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(
                        List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.println(new String(msg.getBody()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.println("RocketMQ Consumer Started...");
        }
    }

消息队列生产者产生消息实例

package com.mq.test;
    
    import java.util.Date;
    
    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 MQProducer {
        public static void main(String[] args) throws MQClientException {
            DefaultMQProducer producer = new DefaultMQProducer("mq-group");
    //        producer.setNamesrvAddr("123.207.63.192:9876");
            producer.setNamesrvAddr("127.0.0.1:9876");
            producer.setInstanceName("producer");
            producer.start();
            try {
                for (int i = 0; i < 10; i++) {
                    Thread.sleep(1000);  //MQ每隔一秒发送一条消息
                    Message msg = new Message("TopicA-test",// topic
                            "TagA",// tag
                            ("RocketMQ message"+i)
                                    .getBytes()// body
                    );
                    SendResult sendResult = producer.send(msg);//发送消息
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            producer.shutdown();//关闭消息生产者
        }
    }

下面是来自github wiki的学习例子

Filter网络架构，以CPU资源换取宝贵的网卡流量资源

![screenshot][]

启动Broker时，增加以下配置，可以自动加载Filter Server进程

filterServerNums=1

Filter样本（Consumer仅负责将代码上传到Filter Server，由Filter Server编译后执行）

package com.alibaba.rocketmq.example.filter;
    
    import com.alibaba.rocketmq.common.filter.MessageFilter;
    import com.alibaba.rocketmq.common.message.MessageExt;
    
    
    public class MessageFilterImpl implements MessageFilter { 
    
        @Override
        public boolean match(MessageExt msg) { 
            String property = msg.getUserProperty("SequenceId");
            if (property != null) { 
                int id = Integer.parseInt(property);
                if ((id % 3) == 0 && (id > 10)) { 
                    return true;
                }
            }
    
            return false;
        }
    }

Consumer例子

public static void main(String[] args) throws InterruptedException, MQClientException { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupNamecc4");
            
            String filterCode = MixAll.file2String("/home/admin/MessageFilterImpl.java");
            consumer.subscribe("TopicFilter7", "com.alibaba.rocketmq.example.filter.MessageFilterImpl",
                filterCode);
    
            consumer.registerMessageListener(new MessageListenerConcurrently() { 
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                        ConsumeConcurrentlyContext context) { 
                    System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            <br>
            consumer.start();
    
            System.out.println("Consumer Started.");
        }

附录

[RocketMQ原理与安装教程][RocketMQ 2]

[RocketMQ实例][RocketMQ 3]

[阿里RocketMQ Quick Start][RocketMQ Quick Start]

[RocketMQ集群安装][RocketMQ 4]

[rocketMq监控平台rocketmq-console搭建][rocketMq_rocketmq-console]

[RocketMQ]: https://blog.csdn.net/u014427391/article/details/78343163
[RocketMQ 1]: http://rocketmq.apache.org/
[70]: /images/20210813/dc8ee59c962f4bc5ac93664b5a42b284.png
[rocketmq-console]: https://github.com/apache/incubator-rocketmq-externals/tree/master/rocketmq-console
[70 1]: /images/20210813/dd899c2d84954913a223a99dc397e741.png
[screenshot]: /images/20210813/79ddba68a14e4502b41e16aafd55a60f.png
[RocketMQ 2]: http://www.cnblogs.com/zyguo/p/4959530.html
[RocketMQ 3]: https://blog.csdn.net/coffeesweet/article/details/52451860
[RocketMQ Quick Start]: https://blog.csdn.net/a19881029/article/details/34446629
[RocketMQ 4]: https://blog.csdn.net/xxzblog/article/details/78574796
[rocketMq_rocketmq-console]: https://blog.csdn.net/www_wangzheguilai/article/details/76571868