springBoot整合RocketMQ及坑的示例代码

springBoot整合RocketMQ及坑的示例代码

版本：

JDK:1.8

springBoot:1.5.10

rocketMQ：4.2.0

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.DefaultMQPushConhttp://sumer;

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 nahttp://mesrvAddr;

@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

将下面的配置文件根据你的需要改

我这里以前默认是Xms4g，都是g，我修改到m就行了。

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。