springBoot整合RocketMQ及坑的示例代码

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

# 消费者的组名

apache.rocketmq.consumer.pushconsumer=pushconsumer

# 生产者的组名

apache.rocketmq.producer.producergroup=producer

# nameserver地址

apache.rocketmq.namesrvaddr=localhost: 9876

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();

     }

   }

}