RocketMQ入门手册
前言
继我上一篇博客后
分布式消息队列RocketMQ学习教程①
上一篇博客最主要介绍了几种常用的MQ,所以本博客再简单介绍一下RocketMQ的原理和简单的例子,基于Java实现,希望可以帮助学习者
RoketMQ搭建Linux版
“工于利其事,必先利其器”,所以我们首先需要搭建好RocketMQ,
考虑到学习者不一定有Linux系统的服务器,所以本博客介绍一下Linux和Window系统的两种安装方法,以补充上一篇博客
因为阿里已经将RocketMQ捐给Apache了,所以现在我们需要去Apache官网下载
RocketMQ官网
注意RocketMQ是基于Java开发的,所以安装前必须安装JDK,Linux JDK安装的可以看分布式消息队列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查看
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
编程实现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资源换取宝贵的网卡流量资源
启动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."); }
附录
- 【Oracle】Oracle 12c DB In-Memory入门实验手册(二)
- XML:XML轻松学习手册(4)XML快速入门
- Lync 2013快速入门手册之二:语音呼叫
- Elasticsearch2.x手册 - 第1章 入门指南 - 1.2 安装
- [转]BSD Socket 简易入门手册
- 使用nRF51822/nRF51422创建一个简单的BLE应用 ---入门实例手册(中文)之一
- Turtlebot3入门手册之二:硬件
- FusionChart完全入门手册5
- Docker的Online中文学习手册 Docker —— 从入门到实践
- XML轻松学习手册(2)XML快速入门之一
- 动效设计入门工具小手册(微交互)
- DirectX9.0 入门手册
- DirectX9.0 入门手册【转贴】
- 《Red5 用户参考手册》之二:入门第一章 FAQ
- 资源分享redis中文入门手册
- hadoop入门手册5:Hadoop【2.7.1】初级入门之命令:文件系统shell2
- VSS使用入门(VSS使用指南 VSS使用说明 VSS使用手册 2.01版)
- Node.js 入门手册:那些最流行的 Web 开发框架
- Express.js 中文入门指引手册
- V-Play入门手册4-如何获取最新的V-Play更新?