参考资料

windows下配置rocketMQ

解压缩

系统环境变量配置

变量名：ROCKETMQ_HOME

变量值：MQ解压路径MQ文件夹名

启动NAMESERVER

Cmd命令框执行进入至‘MQ文件夹bin’下，然后执行‘start mqnamesrv.cmd’，启动NAMESERVER。成功后会弹出提示框，此框勿关闭。

D:

cd D:softwarerocketmq-all-4.2bin

start mqnamesrv.cmd

结果：

暂时忽略警告。

启动BROKER

Cmd命令框执行进入至‘MQ文件夹bin’下，然后执行‘start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true’，启动BROKER。成功后会弹出提示框，此框勿关闭。

start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

建立项目测试生产者消费者(测试用代码)

该代码只是单纯用于测试，要用于实际环境还需要有相当的规则和编写规范，下一章将补充这个

项目结构如下：

各个文件内容：

pom.xml

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

net.funfunle

TestMQ

1.0-SNAPSHOT

org.apache.maven.plugins

maven-compiler-plugin

3.3

1.8

1.8

org.apache.rocketmq

rocketmq-common

4.2.0

org.apache.rocketmq

rocketmq-client

4.2.0

mq.properties

# 消费者的组名

apache.rocketmq.consumer.PushConsumer=PushConsumer

# 生产者的组名

apache.rocketmq.producer.producerGroup=Producer

# NameServer地址

apache.rocketmq.namesrvAddr=localhost:9876

MQConfig.java

package net.funfunle.TestMQ.config;

import org.apache.commons.lang3.StringUtils;

import java.io.InputStream;

import java.io.InputStreamReader;

import java.util.Properties;

/**

* Created by DGDL-08 on 2017/3/24.

*/

public class MQConfig {

private static Properties config;

private static final String producerGroup;

private static final String namesrvAddr;

private static final String consumerGroup;

static {

//InputStream inputStream = MyClass.class.getClassLoader().getResourceAsStream("com/john/basis/conf.properties");

config=new Properties();

InputStream inputStream=null;

try{

InputStream in = ClassLoader.getSystemResourceAsStream("mq.properties");

InputStreamReader is=new InputStreamReader(in,"utf-8");

config.load(is);

is.close();

in.close();

// config.load(inputStream);

}

catch (Exception ed){

ed.printStackTrace();

}

consumerGroup=config.getProperty("apache.rocketmq.consumer.PushConsumer");

producerGroup=config.getProperty("apache.rocketmq.producer.producerGroup");

namesrvAddr=config.getProperty("apache.rocketmq.namesrvAddr");

}

public static String getProducerGroup() {

return producerGroup;

}

public static String getNamesrvAddr() {

return namesrvAddr;

}

public static String getConsumerGroup() {

return consumerGroup;

}

}

RocketMQClient.java

package net.funfunle.TestMQ;

import net.funfunle.TestMQ.config.MQConfig;

import org.apache.commons.lang3.time.StopWatch;

import org.apache.rocketmq.client.producer.DefaultMQProducer;

import org.apache.rocketmq.client.producer.SendResult;

import org.apache.rocketmq.common.message.Message;

import org.apache.rocketmq.remoting.common.RemotingHelper;

public class RocketMQClient {

/**

* 生产者的组名

*/

private String producerGroup= MQConfig.getProducerGroup();

/**

* NameServer 地址

*/

private String namesrvAddr=MQConfig.getNamesrvAddr();

public void defaultMQProducer() {

//生产者的组名

DefaultMQProducer producer = new DefaultMQProducer(producerGroup);

//指定NameServer地址，多个地址以 ; 隔开

producer.setNamesrvAddr(namesrvAddr);

try {

/**

* Producer对象在使用之前必须要调用start初始化，初始化一次即可

* 注意：切记不可以在每次发送消息时，都调用start方法

*/

producer.start();

//创建一个消息实例，包含 topic、tag 和 消息体

//如下：topic 为 "TopicTest"，tag 为 "push"

Message message = new Message("TopicTest", "push", "发送消息----zhisheng-----".getBytes(RemotingHelper.DEFAULT_CHARSET));

StopWatch stop = new StopWatch();

stop.start();

for (int i = 0; i < 10000; i++) {

SendResult result = producer.send(message);

System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());

}

stop.stop();

System.out.println("----------------发送一万条消息耗时：" + stop.getTime());

} catch (Exception e) {

e.printStackTrace();

} finally {

producer.shutdown();

}

}

public static void main(String[] args){

new RocketMQClient().defaultMQProducer();

}

}

RocketMQServer.java

package net.funfunle.TestMQ;

import net.funfunle.TestMQ.config.MQConfig;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;

import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;

import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import org.apache.rocketmq.common.message.MessageExt;

import org.apache.rocketmq.remoting.common.RemotingHelper;

/**

* Created by zhisheng_tian on 2018/2/6

*/

public class RocketMQServer {

/**

* 消费者的组名

*/

private String consumerGroup= MQConfig.getConsumerGroup();

/**

* NameServer 地址

*/

private String namesrvAddr=MQConfig.getNamesrvAddr();

public void defaultMQPushConsumer() {

//消费者的组名

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

//指定NameServer地址，多个地址以 ; 隔开

consumer.setNamesrvAddr(namesrvAddr);

try {

//订阅PushTopic下Tag为push的消息

consumer.subscribe("TopicTest", "push");

//设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费

//如果非第一次启动，那么按照上次消费的位置继续消费

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {

try {

for (MessageExt messageExt : list) {

System.out.println("messageExt: " + messageExt);//输出消息内容

String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);

System.out.println("消费响应：msgId : " + messageExt.getMsgId() + ", msgBody : " + messageBody);//输出消息内容

}

} catch (Exception e) {

e.printStackTrace();

return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试

}

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功

});

consumer.start();

} catch (Exception e) {

e.printStackTrace();

}

}

public static void main(String[] args){

new RocketMQServer().defaultMQPushConsumer();

}

}

分别执行生产者和消费者，可以得到结果：