mvn -Prelease-all -DskipTests clean packager -U

<!-- MQ Begin -->
<dependency>
 <groupId>org.apache.rocketmq</groupId>
 <artifactId>rocketmq-client</artifactId>
 <version>${rocketmq.version}</version>
</dependency>
<!-- MQ End -->

@ConfigurationProperties(prefix = "rocketmq")
public class RocketMQProperties {
 private boolean isEnable = false;
 private String namesrvAddr = "localhost:9876";
 private String groupName = "default";
 private int producerMaxMessageSize = 1024;
 private int producerSendMsgTimeout = 2000;
 private int producerRetryTimesWhenSendFailed = 2;
 private int consumerConsumeThreadMin = 5;
 private int consumerConsumeThreadMax = 30;
 private int consumerConsumeMessageBatchMaxSize = 1;
 //省略get set
}

#發(fā)送同一類消息的設(shè)置為同一個(gè)group，保證唯一,默認(rèn)不需要設(shè)置，rocketmq會(huì)使用ip@pid(pid代表jvm名字)作為唯一標(biāo)示
rocketmq.groupName=please_rename_unique_group_name
#是否開啟自動(dòng)配置
rocketmq.isEnable=true
#mq的nameserver地址
rocketmq.namesrvAddr=127.0.0.1:9876
#消息最大長度 默認(rèn)1024*4(4M)
rocketmq.producer.maxMessageSize=4096
#發(fā)送消息超時(shí)時(shí)間,默認(rèn)3000
rocketmq.producer.sendMsgTimeout=3000
#發(fā)送消息失敗重試次數(shù)，默認(rèn)2
rocketmq.producer.retryTimesWhenSendFailed=2
#消費(fèi)者線程數(shù)量
rocketmq.consumer.consumeThreadMin=5
rocketmq.consumer.consumeThreadMax=32
#設(shè)置一次消費(fèi)消息的條數(shù)，默認(rèn)為1條
rocketmq.consumer.consumeMessageBatchMaxSize=1

/**
 * 消費(fèi)者接口
 * 
 * @author SimpleWu
 *
 */
public interface RocketConsumer {

/**
 * 初始化消費(fèi)者
 */
 public abstract void init();

 /**
 * 注冊(cè)監(jiān)聽
 * 
 * @param messageListener
 */
 public void registerMessageListener(MessageListener messageListener);

}

/**
 * 消費(fèi)者基本信息
 * 
 * @author SimpelWu
 */
public abstract class AbstractRocketConsumer implements RocketConsumer {

 protected String topics;
 protected String tags;
 protected MessageListener messageListener;
 protected String consumerTitel;
 protected MQPushConsumer mqPushConsumer;

 /**
 * 必要的信息
 * 
 * @param topics
 * @param tags
 * @param consumerTitel
 */
 public void necessary(String topics, String tags, String consumerTitel) {
 this.topics = topics;
 this.tags = tags;
 this.consumerTitel = consumerTitel;
 }

 public abstract void init();

 @Override
 public void registerMessageListener(MessageListener messageListener) {
 this.messageListener = messageListener;
 }
 
}

/**
 * mq配置
 * 
 * @author SimpleWu
 */
@Configuration
@EnableConfigurationProperties({ RocketMQProperties.class })
@ConditionalOnProperty(prefix = "rocketmq", value = "isEnable", havingValue = "true")
public class RocketMQConfiguation {

 private RocketMQProperties properties;

 private ApplicationContext applicationContext;

 private Logger log = LoggerFactory.getLogger(RocketMQConfiguation.class);

 public RocketMQConfiguation(RocketMQProperties properties, ApplicationContext applicationContext) {
 this.properties = properties;
 this.applicationContext = applicationContext;
 }

 /**
 * 注入一個(gè)默認(rèn)的消費(fèi)者
 * @return
 * @throws MQClientException
 */
 @Bean
 public DefaultMQProducer getRocketMQProducer() throws MQClientException {
 if (StringUtils.isEmpty(properties.getGroupName())) {
  throw new MQClientException(-1, "groupName is blank");
 }

 if (StringUtils.isEmpty(properties.getNamesrvAddr())) {
  throw new MQClientException(-1, "nameServerAddr is blank");
 }
 DefaultMQProducer producer;
 producer = new DefaultMQProducer(properties.getGroupName());

 producer.setNamesrvAddr(properties.getNamesrvAddr());
 // producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");

 // 如果需要同一個(gè)jvm中不同的producer往不同的mq集群發(fā)送消息，需要設(shè)置不同的instanceName
 // producer.setInstanceName(instanceName);
 producer.setMaxMessageSize(properties.getProducerMaxMessageSize());
 producer.setSendMsgTimeout(properties.getProducerSendMsgTimeout());
 // 如果發(fā)送消息失敗，設(shè)置重試次數(shù)，默認(rèn)為2次
 producer.setRetryTimesWhenSendFailed(properties.getProducerRetryTimesWhenSendFailed());

 try {
  producer.start();
  log.info("producer is start ! groupName:{},namesrvAddr:{}", properties.getGroupName(),
   properties.getNamesrvAddr());
 } catch (MQClientException e) {
  log.error(String.format("producer is error {}", e.getMessage(), e));
  throw e;
 }
 return producer;

 }

 /**
 * SpringBoot啟動(dòng)時(shí)加載所有消費(fèi)者
 */
 @PostConstruct
 public void initConsumer() {
 Map<String, AbstractRocketConsumer> consumers = applicationContext.getBeansOfType(AbstractRocketConsumer.class);
 if (consumers == null || consumers.size() == 0) {
  log.info("init rocket consumer 0");
 }
 Iterator<String> beans = consumers.keySet().iterator();
 while (beans.hasNext()) {
  String beanName = (String) beans.next();
  AbstractRocketConsumer consumer = consumers.get(beanName);
  consumer.init();
  createConsumer(consumer);
  log.info("init success consumer title {} , toips {} , tags {}", consumer.consumerTitel, consumer.tags,
   consumer.topics);
 }
 }

 /**
 * 通過消費(fèi)者信心創(chuàng)建消費(fèi)者
 * 
 * @param consumerPojo
 */
 public void createConsumer(AbstractRocketConsumer arc) {
 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.properties.getGroupName());
 consumer.setNamesrvAddr(this.properties.getNamesrvAddr());
 consumer.setConsumeThreadMin(this.properties.getConsumerConsumeThreadMin());
 consumer.setConsumeThreadMax(this.properties.getConsumerConsumeThreadMax());
 consumer.registerMessageListener(arc.messageListenerConcurrently);
 /**
  * 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開始消費(fèi)還是隊(duì)列尾部開始消費(fèi) 如果非第一次啟動(dòng)，那么按照上次消費(fèi)的位置繼續(xù)消費(fèi)
  */
 // consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 /**
  * 設(shè)置消費(fèi)模型，集群還是廣播，默認(rèn)為集群
  */
 // consumer.setMessageModel(MessageModel.CLUSTERING);

 /**
  * 設(shè)置一次消費(fèi)消息的條數(shù)，默認(rèn)為1條
  */
 consumer.setConsumeMessageBatchMaxSize(this.properties.getConsumerConsumeMessageBatchMaxSize());
 try {
  consumer.subscribe(arc.topics, arc.tags);
  consumer.start();
  arc.mqPushConsumer=consumer;
 } catch (MQClientException e) {
  log.error("info consumer title {}", arc.consumerTitel, e);
 }

 }

}

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.xcloud.config.rocketmq.RocketMQConfiguation

@Component
public class DefaultConsumerMQ extends AbstractRocketConsumer {
 /**
 * 初始化消費(fèi)者
 */
 @Override
 public void init() {
 // 設(shè)置主題,標(biāo)簽與消費(fèi)者標(biāo)題
 super.necessary("TopicTest", "*", "這是標(biāo)題");
 //消費(fèi)者具體執(zhí)行邏輯
 registerMessageListener(new MessageListenerConcurrently() {
  @Override
  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
  msgs.forEach(msg -> {
   System.out.printf("consumer message boyd %s %n", new String(msg.getBody()));
  });
  // 標(biāo)記該消息已經(jīng)被成功消費(fèi)
  return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
  }
 });
 }
}

@Override
public String qmtest(@PathVariable("name")String name) throws MQClientException, RemotingException, MQBrokerException, InterruptedException, UnsupportedEncodingException {
 Message msg = new Message("TopicTest", "tags1", name.getBytes(RemotingHelper.DEFAULT_CHARSET));
 // 發(fā)送消息到一個(gè)Broker
 SendResult sendResult = defaultMQProducer.send(msg);
 // 通過sendResult返回消息是否成功送達(dá)
 System.out.printf("%s%n", sendResult);
 return null;
}

http://localhost:10001/demo/base/mq/hello consumer message boyd hello 
http://localhost:10001/demo/base/mq/嘿嘿嘿嘿嘿 consumer message boyd 嘿嘿嘿嘿嘿