java RocketMQ快速入门基础知识

如何使用

1、引入 rocketmq-client

<dependency>
<groupid>org.apache.rocketmq</groupid>
<artifactid>rocketmq-client</artifactid>
<version>4.1.0-incubating</version>
</dependency>

2、编写producer

defaultmqproducer producer = new defaultmqproducer("producer_demo");
//指定nameserver地址
producer.setnamesrvaddr("192.168.116.115:9876;192.168.116.116:9876"); //修改为自己的
/**
* producer对象在使用之前必须要调用start初始化，初始化一次即可
* 注意：切记不可以在每次发送消息时，都调用start方法
*/
producer.start();

for (int i = 0; i < 997892; i++) {
try {
//构建消息
message msg = new message("topictest" /* topic */,
"taga" /* tag */,
("测试rocketmq" + 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();

3、编写consumer

/**
* consumer group,非常重要的概念，后续会慢慢补充
*/
defaultmqpushconsumer consumer = new defaultmqpushconsumer("consumer_demo");
//指定nameserver地址，多个地址以 ; 隔开
consumer.setnamesrvaddr("192.168.116.115:9876;192.168.116.116:9876"); //修改为自己的
/**
* 设置consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
* 如果非第一次启动，那么按照上次消费的位置继续消费
*/
consumer.setconsumefromwhere(consumefromwhere.consume_from_first_offset);
consumer.subscribe("topictest", "*");
consumer.registermessagelistener(new messagelistenerconcurrently() {
@override
public consumeconcurrentlystatus consumemessage(list<messageext> msgs,
consumeconcurrentlycontext context) {
try {
for(messageext msg:msgs){
string msgbody = new string(msg.getbody(), "utf-8");
system.out.println(" messagebody: "+ msgbody);//输出消息内容
}
} catch (exception e) {
e.printstacktrace();
return consumeconcurrentlystatus.reconsume_later; //稍后再试
}
return consumeconcurrentlystatus.consume_success; //消费成功
}
});
consumer.start();
system.out.printf("consumer started.%n");

4、说明

各位根据自己的环境，修改namesrvaddr的值，我的集群请参考：rocketmq集群部署配置。稍后通过rocketmq管控台就可以看到之前搭建的多master多slave模式，异步复制集群模式。

5、通过rocketmq管控台

rocketmq-console-ng获取方式为：rocketmq-console-ng，之后通过mavne进行编译获取jar，命令如下：

mvn clean package -dmaven.test.skip=true
java -jar target/rocketmq-console-ng-1.0.0.jar

得到rocketmq-console-ng-1.0.0.jar之后，找到rocketmq-console-ng-1.0.0.jar\boot-inf\classes\application.properties文件，根据自己的namesrvaddr进行修改rocketmq.config.namesrvaddr的值。

直接启动：

java -jar rocketmq-console-ng-1.0.0.jar

管控台是基于springboot的，的确springboot非常方便和非常火了，所以有必要去学习下springboot了（其实还是spring系列，所以spring也必要深入学习下），稍后通过管控台进行观察运行。

6、运行观察

一个好的习惯是先运行consumer，之后在运行producer，之后通过rocketmq-console-ng管控台观察

运行完成之后，的确broker-a的数据加上broker-b的数据量就等于我们发送的数据量，而且slave的数量也master的数量也是一致的，效果如下：

查看发送这些数据，2台机器的磁盘情况如下：

到目前位置，关于rocketmq快速入门就结束了。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。