//根据Topic分组

            Map<String, List<MessageExt>> topicGroups = msgs.stream().collect(Collectors.groupingBy(MessageExt::getTopic));

            for (Map.Entry<String, List<MessageExt>> topicEntry : topicGroups.entrySet()) {

                String topic = topicEntry.getKey();

                //根据tags分组

                List<MessageExt> messageExts=topicEntry.getValue();

                for(MessageExt messageExt:messageExts){

                    String tag= messageExt.getTags();

                    if(StringUtils.isEmpty(tag)){

                        messageExt.setTags("*");

                    }

                }

                Map<String, List<MessageExt>> tagGroups = topicEntry.getValue().stream().collect(Collectors.groupingBy(MessageExt::getTags));

                for (Map.Entry<String, List<MessageExt>> tagEntry : tagGroups.entrySet()) {

                    String tag = tagEntry.getKey();

                    //消费某个主题下，tag的消息

                    this.consumeMsgForTag(topic,tag,tagEntry.getValue());

                }

            }

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

 xmlns:aop="http://www.springframework.org/schema/aop"

 xmlns:context="http://www.springframework.org/schema/context"

 xmlns:p="http://www.springframework.org/schema/p"

 xmlns:tx="http://www.springframework.org/schema/tx"

 xmlns:cache="http://www.springframework.org/schema/cache"

 xsi:schemaLocation="http://www.springframework.org/schema/beans

  http://www.springframework.org/schema/beans/spring-beans.xsd

  http://www.springframework.org/schema/aop

  http://www.springframework.org/schema/aop/spring-aop-3.0.xsd

  http://www.springframework.org/schema/context

  http://www.springframework.org/schema/context/spring-context-3.0.xsd

  http://www.springframework.org/schema/tx

  http://www.springframework.org/schema/tx/spring-tx-3.0.xsd

  http://www.springframework.org/schema/cache

  http://www.springframework.org/schema/cache/spring-cache.xsd">   

    <cache:annotation-driven/>

    <bean id="messageListeners" class="com.rongdu.cashloan.cl.mq.MQConsumeMsgListenerProcessor"></bean>

    <bean id="rocketmqConsumer" class="org.apache.rocketmq.client.consumer.DefaultMQPushConsumer" init-method="start" destroy-method="shutdown">

        <property name="consumerGroup" value="${rocketmq.consumer.groupName}"/>

        <property name="namesrvAddr" value="${rocketmq.consumer.namesrvAddr}"/>

        <property name="messageListener" ref="messageListeners"/>

        <property name="subscription">

            <map>

                <entry key="${rocketmq.consumer.topics}" value="${rocketmq.consumer.tags}" />

            </map>

        </property>

        <property name="consumeMessageBatchMaxSize" value="${rocketmq.consumer.consumeMessageBatchMaxSize}"></property>

    </bean>

    <bean id="rocketMQProducer" class="org.apache.rocketmq.client.producer.DefaultMQProducer" init-method="start" destroy-method="shutdown">

        <property name="producerGroup" value="${rocketmq.producer.groupName}"/>

        <property name="namesrvAddr" value="${rocketmq.producer.namesrvAddr}"/>

        <!-- 失败重试次数 -->

        <property name="retryTimesWhenSendFailed" value="${rocketmq.producer.retryTimesWhenSendFailed}" />

    </bean>

</beans> 

#该应用是否启用生产者

rocketmq.producer.isOnOff=on

#发送同一类消息的设置为同一个group，保证唯一,默认不需要设置，rocketmq会使用ip@pid(pid代表jvm名字)作为唯一标示

rocketmq.producer.groupName=cashloan-producer

#mq的nameserver地址

rocketmq.producer.namesrvAddr=172.20.102.150:

rocketmq.producer.topics=cashloan_risk_control_req

#消息最大长度 默认1024*(4M)

rocketmq.producer.maxMessageSize=

#发送消息超时时间,默认3000

rocketmq.producer.sendMsgTimeout=

#发送消息失败重试次数，默认2

rocketmq.producer.retryTimesWhenSendFailed=

###consumer

##该应用是否启用消费者

rocketmq.consumer.isOnOff=on

rocketmq.consumer.groupName=cashloan-consumer

#mq的nameserver地址

rocketmq.consumer.namesrvAddr=172.20.102.150:

#该消费者订阅的主题和tags("*"号表示订阅该主题下所有的tags),格式：topic~tag1||tag2||tag3;topic2~*;

rocketmq.consumer.topics=cashloan_risk_control_result

rocketmq.consumer.tags=*

rocketmq.consumer.consumeThreadMin=

rocketmq.consumer.consumeThreadMax=

#设置一次消费消息的条数，默认为1条

rocketmq.consumer.consumeMessageBatchMaxSize=

Message msg = new Message("cashloan_risk_control_req" /* Topic */,

                    "TagA" /* Tag */,

                    (String.valueOf(borrowId)).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */

            );

            SendResult sendResult = producer.send(msg);

@Component

public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently {

    private static final Logger logger = LoggerFactory.getLogger(MQConsumeMsgListenerProcessor.class);

    @Override

    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

        if(CollectionUtils.isEmpty(msgs)){

            logger.info("接受到的消息为空，不处理，直接返回成功");

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

        }

        MessageExt messageExt = msgs.get();

        logger.info("接受到的消息为："+messageExt.toString());

        if(messageExt.getTopic().equals("cashloan_risk_control")){

            if(messageExt.getTags().equals("borrowId")){

                //TODO 判断该消息是否重复消费（RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重）

                //TODO 获取该消息重试次数

                int reconsume = messageExt.getReconsumeTimes();

                if(reconsume ==){//消息已经重试了3次，如果不需要再次消费，则返回成功

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

                }

                //TODO 处理对应的业务逻辑

            }

        }

        // 如果没有return success ，consumer会重新消费该消息，直到return success

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

    }

}