189 8069 5689
今天就跟大家聊聊有关RocketMQ怎么在springBoot中使用，可能很多人都不太了解，为了让大家更加了解，小编给大家总结了以下内容，希望大家根据这篇文章可以有所收获。
成都创新互联公司是一家专注于成都做网站、成都网站设计、成都外贸网站建设与策划设计,五常网站建设哪家好?成都创新互联公司做网站,专注于网站建设十载,网设计领域的专业建站公司;建站业务涵盖:五常等地区。五常做网站价格咨询:18982081108
pom 配置：
org.springframework.boot spring-boot-starter-parent 1.5.10.RELEASE org.apache.rocketmq rocketmq-client 4.2.0
application.properties 配置:
# 消费者的组名 apache.rocketmq.consumer.PushConsumer=PushConsumer # 生产者的组名 apache.rocketmq.producer.producerGroup=Producer # NameServer地址 apache.rocketmq.namesrvAddr=localhost:9876
java代码：
生产者
package test.config.rocketmq;
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;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;
import javax.annotation.PostConstruct;
@Component
public class RocketMQClient {
/**
* 生产者的组名
*/
@Value("${apache.rocketmq.producer.producerGroup}")
private String producerGroup;
/**
* NameServer 地址
*/
@Value("${apache.rocketmq.namesrvAddr}")
private String namesrvAddr;
@PostConstruct
public void defaultMQProducer() {
//生产者的组名
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
//指定NameServer地址，多个地址以 ; 隔开
producer.setNamesrvAddr(namesrvAddr);
producer.setVipChannelEnabled(false);
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 < 1; i++) {
SendResult result = producer.send(message);
System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());
}
stop.stop();
System.out.println("----------------发送一万条消息耗时：" + stop.getTotalTimeMillis());
} catch (Exception e) {
e.printStackTrace();
} finally {
producer.shutdown();
}
}
}消费者：
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;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
@Component
public class RocketMQServer {
/**
* 消费者的组名
*/
@Value("${apache.rocketmq.consumer.PushConsumer}")
private String consumerGroup;
/**
* NameServer 地址
*/
@Value("${apache.rocketmq.namesrvAddr}")
private String namesrvAddr;
@PostConstruct
public void defaultMQPushConsumer() {
//消费者的组名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
//指定NameServer地址，多个地址以 ; 隔开
consumer.setNamesrvAddr(namesrvAddr);
consumer.setVipChannelEnabled(false);
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();
}
}
}掉坑总结：
1.rocketMQ启动时，命令不是 mqbroker -n 127.0.0.1:9876
正确应该是：mqbroker -n 127.0.0.1:9876 butiautoCreateTopicEnable=true
否则会抛出：No route info of this topic, TopicTest
2.客户端连接时抛出异常
org.apache.rocketmq.client.exception.MQClientException:
Send [3] times, still failed, cost [3180]ms, Topic: TopicTest, BrokersSent: \
[WIN-93CGO0S5G25, WIN-93CGO0S5G25, WIN-93CGO0S5G25]
解决方式两种
1.producer.setVipChannelEnabled(false); 生产者和消费者添加这行代买。
2.降rocketmq版本,降成3.2.6
关于spring.rocketmq.name-server的坑
看下图：
注意：
如果你是SpringBoot2.0+的框架，或者是JDK10。
你需要将你自己的项目配置文件中的，spring.rocketmq.name-server改成
spring.rocketmq.nameServer。注意是nameServer。
不然就会报各种稀奇古怪的bug。
关于启动报内存不足的错
在安装启动Name Server和Broker的时候，一定要修改配置文件，不然内存会爆炸。
Native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory
看完上述内容，你们对RocketMQ怎么在springBoot中使用有进一步的了解吗？如果还想了解更多知识或者相关内容，请关注创新互联行业资讯频道，感谢大家的支持。