import com.alibaba.rocketmq.client.exception.MQClientException;

import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;

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.util.StringUtils;

@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 testRocketMQConsumer() throws Exception {

if (StringUtils.isEmpty(groupName)){

throw new Exception(“groupName is null !!!”);

}

if (StringUtils.isEmpty(namesrvAddr)){

throw new Exception(“namesrvAddr is null !!!”);

}

if(StringUtils.isEmpty(topics)){

throw new Exception(“topics is null !!!”);

}

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);

consumer.setNamesrvAddr(namesrvAddr);

consumer.setConsumeThreadMin(consumeThreadMin);

consumer.setConsumeThreadMax(consumeThreadMax);

consumer.registerMessageListener(mqMessageListenerProcessor);

/**

• 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费

• 如果非第一次启动，那么按照上次消费的位置继续消费

*/

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

/**

• 设置消费模型，集群还是广播，默认为集群

*/

//consumer.setMessageModel(MessageModel.CLUSTERING);

/**

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

*/

consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);

try {

/**

• 设置该消费者订阅的主题和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 Exception(e);

}

return consumer;

}

}

5.新建处理类

---

package cn.baocl.rocketmq.processor;

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;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Component;

import org.springframework.util.CollectionUtils;

import java.util.List;

@Component

public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently {

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

/**

• 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
  

• 不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS

*/

@Override

public ConsumeConcurrentlyStatus consumeMessage(List 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 判断该消息是否重复消费（RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重）

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

int reconsume = messageExt.getReconsumeTimes();

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

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

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

}

}

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

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

}

6.接口调用

---

package cn.baocl.rocketmq.controllor;

import cn.baocl.rocketmq.entity.TestVo;

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.SendCallback;

import com.alibaba.rocketmq.client.producer.SendResult;

import com.alibaba.rocketmq.common.message.Message;

import com.alibaba.rocketmq.remoting.exception.RemotingException;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年Java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上Java开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加V获取：vip1024b （备注Java）

最后

由于篇幅限制，小编在此截出几张知识讲解的图解

一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

712805051785)]

[外链图片转存中…(img-wumYaoE6-1712805051785)]

[外链图片转存中…(img-XyGKx4wQ-1712805051785)]

[外链图片转存中…(img-77GTefII-1712805051786)]

[外链图片转存中…(img-1ex16839-1712805051786)]

一个人可以走的很快，但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎扫码加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
[外链图片转存中…(img-mSI1tNBT-1712805051786)]