这篇文章主要介绍了springBoot整合RocketMQ及坑的示例代码，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
版本：
- JDK:1.8
- springBoot:1.5.10
- rocketMQ：4.2.0
pom 配置：
org.springframework.bootspring-boot-starter-parent1.5.10.RELEASE org.apache.rocketmqrocketmq-client4.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 defaultMQ<mark style="color:transparent">来源gaodaimacom搞#^代%!码&网</mark>PushConsumer() { //消费者的组名 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”
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持gaodaima搞代码网。
以上就是springBoot整合RocketMQ及坑的示例代码的详细内容，更多请关注gaodaima搞代码网其它相关文章！
转载请注明原文链接：springBoot整合RocketMQ及坑的示例代码
