前言

继我上一篇博客后

分布式消息队列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入门手册-LMLPHP

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

RocketMQ入门手册-LMLPHP

编程实现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资源换取宝贵的网卡流量资源

RocketMQ入门手册-LMLPHP

启动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实例

阿里RocketMQ Quick Start

RocketMQ集群安装

rocketMq监控平台rocketmq-console搭建

05-11 22:15