springBoot整合RocketMQ及坑的示例代码

2019-11-26 09:32:01

字体：大中小

来源：转载

供稿：网友

版本：

JDK:1.8
springBoot:1.5.10
rocketMQ：4.2.0

pom 配置：

<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.10.RELEASE</version></parent><dependency>  <groupId>org.apache.rocketmq</groupId>  <artifactId>rocketmq-client</artifactId>  <version>4.2.0</version></dependency>

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;@Componentpublic 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;@Componentpublic 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的坑