springBoot整合RocketMQ及坑的示例代码

版本：

• 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;

@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 

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

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

java_opt="${java_opt} -server -xms256m -xmx256m -xmn128m -xx:metaspacesize=128m -xx:maxmetaspacesize=320m"

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