Spring Boot 整合 RocketMq

官方有專門的springboot-starter:https://github.com/apache/rocketmq-externals/blob/master/rocketmq-spring-boot-starter/README_zh_CN.md

Spring Boot 整合 RocketMq

1.pom.xml添加RocketMq依賴

<dependency>  
     <groupId>com.alibaba.rocketmq</groupId>  
     <artifactId>rocketmq-client</artifactId>  
     <version>3.2.6</version>  
</dependency>

2.application.properties中添加如下信息

###producer
#該應(yīng)用是否啟用生產(chǎn)者
rocketmq.producer.isOnOff=on
#發(fā)送同一類消息的設(shè)置為同一個group，保證唯一,默認(rèn)不需要設(shè)置，rocketmq會使用ip@pid(pid代表jvm名字)作為唯一標(biāo)示
rocketmq.producer.groupName=${spring.application.name}
#mq的nameserver地址
rocketmq.producer.namesrvAddr=127.0.0.1:9876
#消息最大長度 默認(rèn)1024*4(4M)
rocketmq.producer.maxMessageSize=4096
#發(fā)送消息超時時間,默認(rèn)3000
rocketmq.producer.sendMsgTimeout=3000
#發(fā)送消息失敗重試次數(shù)，默認(rèn)2
rocketmq.producer.retryTimesWhenSendFailed=2

###consumer
##該應(yīng)用是否啟用消費者
rocketmq.consumer.isOnOff=on
rocketmq.consumer.groupName=${spring.application.name}
#mq的nameserver地址
rocketmq.consumer.namesrvAddr=127.0.0.1:9876
#該消費者訂閱的主題和tags("*"號表示訂閱該主題下所有的tags),格式：topic~tag1||tag2||tag3;topic2~*;
rocketmq.consumer.topics=DemoTopic~*;
rocketmq.consumer.consumeThreadMin=20
rocketmq.consumer.consumeThreadMax=64
#設(shè)置一次消費消息的條數(shù)，默認(rèn)為1條
rocketmq.consumer.consumeMessageBatchMaxSize=1

3.生產(chǎn)者Bean配置

package com.clouds.common.rocketmq.producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.util.StringUtils;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.clouds.common.rocketmq.constants.RocketMQErrorEnum;
import com.clouds.common.rocketmq.exception.RocketMQException;

/**
 * 生產(chǎn)者配置
 * .<br/>
 * 
 * Copyright: Copyright (c) 2017  zteits
 * 
 * @ClassName: MQProducerConfiguration
 * @Description: 
 * @version: v1.0.0
 * @author: zhaowg
 * @date: 2018年3月2日 下午11:44:36
 * Modification History:
 * Date             Author          Version            Description
 *---------------------------------------------------------*
 * 2018年3月2日      zhaowg           v1.0.0               創(chuàng)建
 */
@SpringBootConfiguration
public class MQProducerConfiguration {
    public static final Logger LOGGER = LoggerFactory.getLogger(MQProducerConfiguration.class);
    /**
     * 發(fā)送同一類消息的設(shè)置為同一個group，保證唯一,默認(rèn)不需要設(shè)置，rocketmq會使用ip@pid(pid代表jvm名字)作為唯一標(biāo)示
     */
    @Value("${rocketmq.producer.groupName}")
    private String groupName;
    @Value("${rocketmq.producer.namesrvAddr}")
    private String namesrvAddr;
    /**
     * 消息最大大小，默認(rèn)4M
     */
    @Value("${rocketmq.producer.maxMessageSize}")
    private Integer maxMessageSize ;
    /**
     * 消息發(fā)送超時時間，默認(rèn)3秒
     */
    @Value("${rocketmq.producer.sendMsgTimeout}")
    private Integer sendMsgTimeout;
    /**
     * 消息發(fā)送失敗重試次數(shù)，默認(rèn)2次
     */
    @Value("${rocketmq.producer.retryTimesWhenSendFailed}")
    private Integer retryTimesWhenSendFailed;

    @Bean
    public DefaultMQProducer getRocketMQProducer() throws RocketMQException {
        if (StringUtils.isEmpty(this.groupName)) {
            throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"groupName is blank",false);
        }
        if (StringUtils.isEmpty(this.namesrvAddr)) {
            throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"nameServerAddr is blank",false);
        }
        DefaultMQProducer producer;
        producer = new DefaultMQProducer(this.groupName);
        producer.setNamesrvAddr(this.namesrvAddr);
        //如果需要同一個jvm中不同的producer往不同的mq集群發(fā)送消息，需要設(shè)置不同的instanceName
        //producer.setInstanceName(instanceName);
        if(this.maxMessageSize!=null){
            producer.setMaxMessageSize(this.maxMessageSize);
        }
        if(this.sendMsgTimeout!=null){
            producer.setSendMsgTimeout(this.sendMsgTimeout);
        }
        //如果發(fā)送消息失敗，設(shè)置重試次數(shù)，默認(rèn)為2次
        if(this.retryTimesWhenSendFailed!=null){
            producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
        }
        
        try {
            producer.start();
            
            LOGGER.info(String.format("producer is start ! groupName:[%s],namesrvAddr:[%s]"
                    , this.groupName, this.namesrvAddr));
        } catch (MQClientException e) {
            LOGGER.error(String.format("producer is error {}"
                    , e.getMessage(),e));
            throw new RocketMQException(e);
        }
        return producer;
    }
}

4.消費者Bean配置

package com.clouds.common.rocketmq.consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.util.StringUtils;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.clouds.common.rocketmq.constants.RocketMQErrorEnum;
import com.clouds.common.rocketmq.consumer.processor.MQConsumeMsgListenerProcessor;
import com.clouds.common.rocketmq.exception.RocketMQException;


/**
 * 消費者Bean配置
 * .<br/>
 * 
 * Copyright: Copyright (c) 2017  zteits
 * 
 * @ClassName: MQConsumerConfiguration
 * @Description: 
 * @version: v1.0.0
 * @author: zhaowg
 * @date: 2018年3月2日 下午11:48:32
 * Modification History:
 * Date             Author          Version            Description
 *---------------------------------------------------------*
 * 2018年3月2日      zhaowg           v1.0.0               創(chuàng)建
 */
@SpringBootConfiguration
public class MQConsumerConfiguration {
    public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerConfiguration.class);
    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;
    @Value("${rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;
    @Value("${rocketmq.consumer.topics}")
    private String topics;
    @Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
    private int consumeMessageBatchMaxSize;
    @Autowired
    private MQConsumeMsgListenerProcessor mqMessageListenerProcessor;
    
    @Bean
    public DefaultMQPushConsumer getRocketMQConsumer() throws RocketMQException {
        if (StringUtils.isEmpty(groupName)){
            throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"groupName is null !!!",false);
        }
        if (StringUtils.isEmpty(namesrvAddr)){
            throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"namesrvAddr is null !!!",false);
        }
        if(StringUtils.isEmpty(topics)){
            throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL,"topics is null !!!",false);
        }
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.registerMessageListener(mqMessageListenerProcessor);
        /**
         * 設(shè)置Consumer第一次啟動是從隊列頭部開始消費還是隊列尾部開始消費
         * 如果非第一次啟動，那么按照上次消費的位置繼續(xù)消費
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        /**
         * 設(shè)置消費模型，集群還是廣播，默認(rèn)為集群
         */
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        /**
         * 設(shè)置一次消費消息的條數(shù)，默認(rèn)為1條
         */
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
        try {
            /**
             * 設(shè)置該消費者訂閱的主題和tag，如果是訂閱該主題下的所有tag，則tag使用*；如果需要指定訂閱該主題下的某些tag，則使用||分割，例如tag1||tag2||tag3
             */
            String[] topicTagsArr = topics.split(";");
            for (String topicTags : topicTagsArr) {
                String[] topicTag = topicTags.split("~");
                consumer.subscribe(topicTag[0],topicTag[1]);
            }
            consumer.start();
            LOGGER.info("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr);
        }catch (MQClientException e){
            LOGGER.error("consumer is start !!! groupName:{},topics:{},namesrvAddr:{}",groupName,topics,namesrvAddr,e);
            throw new RocketMQException(e);
        }
        return consumer;
    }
}

5.消費者消息監(jiān)聽處理器

package com.clouds.common.rocketmq.consumer.processor;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
/**
 * 消費者消費消息路由
 * .<br/>
 * 
 * Copyright: Copyright (c) 2017  zteits
 * 
 * @ClassName: RocketMQMessageListenerConcurrentlyProcessor
 * @Description: 
 * @version: v1.0.0
 * @author: zhaowg
 * @date: 2018年2月28日 上午11:12:32
 * Modification History:
 * Date             Author          Version            Description
 *---------------------------------------------------------*
 * 2018年2月28日      zhaowg           v1.0.0               創(chuàng)建
 */
@Component
public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently{
    private static final Logger logger = LoggerFactory.getLogger(MQConsumeMsgListenerProcessor.class);
    
    /**
     *  默認(rèn)msgs里只有一條消息，可以通過設(shè)置consumeMessageBatchMaxSize參數(shù)來批量接收消息<br/>
     *  不要拋異常，如果沒有return CONSUME_SUCCESS ，consumer會重新消費該消息，直到return CONSUME_SUCCESS
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        if(CollectionUtils.isEmpty(msgs)){
            logger.info("接受到的消息為空，不處理，直接返回成功");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = msgs.get(0);
        logger.info("接受到的消息為："+messageExt.toString());
        if(messageExt.getTopic().equals("你的Topic")){
            if(messageExt.getTags().equals("你的Tag")){
                //TODO 判斷該消息是否重復(fù)消費（RocketMQ不保證消息不重復(fù)，如果你的業(yè)務(wù)需要保證嚴(yán)格的不重復(fù)消息，需要你自己在業(yè)務(wù)端去重）
                //TODO 獲取該消息重試次數(shù)
                int reconsume = messageExt.getReconsumeTimes();
                if(reconsume ==3){//消息已經(jīng)重試了3次，如果不需要再次消費，則返回成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                //TODO 處理對應(yīng)的業(yè)務(wù)邏輯
            }
        }
        // 如果沒有return success ，consumer會重新消費該消息，直到return success
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

至此RocketMq配置已經(jīng)全部完成，現(xiàn)在編寫測試用例測試下。

6.生產(chǎn)者Test

package com.clouds.common.test;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

@RunWith(SpringRunner.class)
@SpringBootTest
public class DefaultProductTest {
    private static final Logger logger = LoggerFactory.getLogger(DefaultProductTest.class);
    
    /**使用RocketMq的生產(chǎn)者*/
    @Autowired
    private DefaultMQProducer defaultMQProducer;
    
    /**
     * 發(fā)送消息
     * 
     * 2018年3月3日 zhaowg
     * @throws InterruptedException 
     * @throws MQBrokerException 
     * @throws RemotingException 
     * @throws MQClientException 
     */
    @Test
    public void send() throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
        String msg = "demo msg test";
        logger.info("開始發(fā)送消息："+msg);
        Message sendMsg = new Message("DemoTopic","DemoTag",msg.getBytes());
        //默認(rèn)3秒超時
        SendResult sendResult = defaultMQProducer.send(sendMsg);
        logger.info("消息發(fā)送響應(yīng)信息："+sendResult.toString());
    }
}

現(xiàn)在啟動SpringBootRocketMqApplication.java.
運行DefaultProductTest類的send方法發(fā)送消息，可以看到在消費者監(jiān)聽器中已經(jīng)獲取到消息了。

注意：發(fā)送的topic和tag，如果希望該應(yīng)用收到消息，要現(xiàn)在application.properties中配置好rocketmq.consumer.topics值。

打印信息如下：

2018-03-03 00:52:39.785  INFO 14592 --- [           main] c.clouds.common.test.DefaultProductTest  : 開始發(fā)送消息：demo msg test
2018-03-03 00:52:39.836  INFO 14592 --- [           main] c.clouds.common.test.DefaultProductTest  : 消息發(fā)送響應(yīng)信息：SendResult [sendStatus=SEND_OK, msgId=2F61081600002A9F00000000000A35F6, messageQueue=MessageQueue [topic=DemoTopic, brokerName=iZbp1g7kmh3b0jp1moarjoZ, queueId=0], queueOffset=2]
2018-03-03 00:52:39.853  INFO 14592 --- [       Thread-3] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@3e694b3f: startup date [Sat Mar 03 00:52:37 CST 2018]; root of context hierarchy
2018-03-03 00:52:39.862  INFO 14592 --- [MessageThread_2] .c.c.r.c.p.MQConsumeMsgListenerProcessor : 接受到的消息為：MessageExt [queueId=0, storeSize=136, queueOffset=2, sysFlag=0, bornTimestamp=1520009559787, bornHost=/222.35.171.182:60888, storeTimestamp=1520009560327, storeHost=/47.97.8.22:10911, msgId=2F61081600002A9F00000000000A35F6, commitLogOffset=669174, bodyCRC=1945576729, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=DemoTopic, flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=3, WAIT=true, TAGS=DemoTag}, body=13]]

7.項目源代碼

https://gitee.com/zhaowg3/springboot-rocketmq/tree/simple/

參考：

  1. 分布式開放消息系統(tǒng)(RocketMQ)的原理與實踐

后續(xù)：消息監(jiān)聽優(yōu)化

如果一個消費者訂閱了多個topic和tag,在上面的消息監(jiān)聽程序中將會出現(xiàn)大量的if..else，每次新增一個訂閱topic，則需要修改消息監(jiān)聽器類；下一篇文章將通過自定義注解消除消息監(jiān)聽器中的if..else。

下一篇：RocketMq消息監(jiān)聽程序消除大量的if..else

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容