springBoot整合rocketMq
因写了一个rocketMQ 发送，有一个消息发送一次，进行一下producer创建，后修改为项目启动创建生产者。
一、项目结构
二、maven相关包：
com.alibaba.rocketmq
rocketmq-client
3.5.8
com.alibaba.rocketmq
rocketmq-all
3.5.8
pom
三、配置文件
application.yml
rockerMq:
#Name Server 地址，因为是集群部署 所以有多个用 分号 隔开
nameServer: ip:9876
#主题名称
topic: dddd
#一般发送同样消息的Producer，归为同一个Group，应用必须设置，并保证命名唯一
producerGroup: dd
#tag 临时值 标签
tags: mm四、生产者部分
1、ErrorCode
import java.io.Serializable;
/**
* @author
* @description:
* @param:
* @return:
* @date：2021/9/22
*/
public interface ErrorCode extends Serializable {
/**
* 错误码
* @return
*/
String getCode();
/**
* 错误信息
* @return
*/
String getMsg();
}2、RocketMQErrorEnum
/**
* @author
* @description:
* @param:
* @return:
* @date：2021/9/22
*/
public enum RocketMQErrorEnum implements ErrorCode{
/********公共********/
PARAMM_NULL("MQ_001","参数为空"),
/********生产者*******/
/********消费者*******/
NOT_FOUND_CONSUMESERVICE("MQ_100","根据topic和tag没有找到对应的消费服务"),
HANDLE_RESULT_NULL("MQ_101","消费方法返回值为空"),
CONSUME_FAIL("MQ_102","消费失败")
;
private String code;
private String msg;
private RocketMQErrorEnum(String code, String msg) {
this.code = code;
this.msg = msg;
}
@Override
public String getCode() {
return this.code;
}
@Override
public String getMsg() {
return this.msg;
}
}3、AppException
/**
* @author
* @description:
* @param:
* @return:
* @date：2021/9/22
*/
public class AppException extends RuntimeException {
private static final long serialVersionUID = 1L;
/**
* 错误编码
*/
protected ErrorCode errCode;
/**
* 错误信息
*/
protected String errMsg;
/**
* 无参构造函数
*/
public AppException() {
super();
}
public AppException(Throwable e) {
super(e);
}
public AppException(ErrorCode errCode, String... errMsg) {
super(errCode.getMsg());
this.errCode = errCode;
setErrMsg(errMsg,true);
}
public AppException(ErrorCode errCode, String errMsg,Boolean isTransfer) {
super(errMsg);
this.errCode = errCode;
setErrMsg(new String[]{errMsg},isTransfer);
}
/**
* 构造函数
*
* @param cause 异常
*/
public AppException(ErrorCode errCode, Throwable cause, String... errMsg) {
super(errCode.getCode() + errCode.getMsg(), cause);
this.errCode = errCode;
setErrMsg(errMsg,true);
}
public ErrorCode getErrCode() {
return errCode;
}
public void setErrCode(ErrorCode errCode) {
this.errCode = errCode;
}
public String getErrMsg() {
return this.errMsg;
}
public void setErrMsg(String[] errMsg,Boolean isTransfer) {
if (null != errMsg &&errMsg.length>0) {
if(errCode.getMsg().contains("%s") && isTransfer){
this.errMsg = String.format(errCode.getMsg(), errMsg);
}else{
StringBuffer sf = new StringBuffer();
for (String msg : errMsg) {
sf.append(msg+";");
}
this.errMsg = sf.toString();
}
}else{
this.errMsg = errCode.getMsg();
}
}
public static void main(String[] args) {
String str = "ERRCode:1004--对象不存在:[%s]";
if (str.contains("%s")){
System.out.println("包含");
}
}
}4.RocketMQException
/**
* @author
* @description:
* @param:
* @return:
* @date：2021/9/22
*/
public class RocketMQException extends AppException{
private static final long serialVersionUID = 1L;
/**
* 无参构造函数
*/
public RocketMQException() {
super();
}
public RocketMQException(Throwable e) {
super(e);
}
public RocketMQException(ErrorCode errorType) {
super(errorType);
}
public RocketMQException(ErrorCode errorCode, String ... errMsg) {
super(errorCode, errMsg);
}
/**
* 封装异常
* @param errorCode
* @param errMsg
* @param isTransfer 是否转换异常信息，如果为false,则直接使用errMsg信息
*/
public RocketMQException(ErrorCode errorCode, String errMsg,Boolean isTransfer) {
super(errorCode, errMsg,isTransfer);
}
public RocketMQException(ErrorCode errCode, Throwable cause,String ... errMsg) {
super(errCode,cause, errMsg);
}
}5、MQProducerConfiguration
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author
* @description:
* @param:
* @return:
* @date：2021/9/22
*/
@Slf4j
@Configuration
public class MQProducerConfiguration {
/**
* 发送同一类消息的设置为同一个group，保证唯一,默认不需要设置，rocketmq会使用ip@pid(pid代表jvm名字)作为唯一标示
*/
@Value("${config.rockerMq.producerGroup}")
private String groupName;
@Value("${config.rockerMq.nameServer}")
private String namesrvAddr;
@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);
producer.setVipChannelEnabled(false);
//如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
//producer.setInstanceName(instanceName);
try {
producer.start();
log.info(String.format("producer is start ! groupName:[%s],namesrvAddr:[%s]"
, this.groupName, this.namesrvAddr));
} catch (MQClientException e) {
log.error(String.format("producer is error {}"
, e.getMessage(),e));
throw new RocketMQException(e);
}
return producer;
}
}6、SendRocketMqMsg
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.client.producer.SendStatus;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.alibaba.rocketmq.common.message.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
* @author taochui
* @description:
* @param:
* @return:
* @date：2021/9/22
*/
@Slf4j
@Service
public class SendRocketMqMsg {
/**使用RocketMq的生产者*/
@Autowired
private DefaultMQProducer defaultMQProducer;
@Resource
private Environment environment;
/**
* 发送消息
*
* 2018年3月3日 tc
* @throws InterruptedException
* @throws MQBrokerException
* @throws RemotingException
* @throws MQClientException
*/
public boolean send(String msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
String topic = environment.getProperty("config.rockerMq.topic").trim();
String tags = environment.getProperty("config.rockerMq.tags").trim();
Message sendMsg = new Message(topic,tags,msg.getBytes());
//默认3秒超时
SendResult sendResult = defaultMQProducer.send(sendMsg);
log.info("发送响应信息："+sendResult.toString());
SendStatus sendStatus = sendResult.getSendStatus();
log.info("sendStatus:" + sendStatus);
if(sendStatus != null && "SEND_OK".equals(sendStatus.toString())){
return true;
}else{
return false;
}
}
}7、使用测试，发送消息
public class test {
@Autowired
SendRocketMqMsg sendRocketMqMsg;
public void sendTest() {
String msg = "test";
boolean status = sendRocketMqMsg.send(msg);
}
}之前的代码部分新增consumer包。
1、MQConsumerConfiguration
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.context.annotation.Configuration;
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;
/**
*@Description: 消费者Bean配置
*@Param:
*@return:
*@Author:
*@date: 2021/10/11
**/
@Configuration
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);
/**
* 设置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 RocketMQException(e);
}
return consumer;
}
}2、MQConsumeMsgListenerProcessor
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;
/**
*@Description: 消费者消费消息路由
*@Param:
*@return:
*@Author:
*@date: 2021/10/11
**/
@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;
}
}
页面更新：2024-03-03
本站资料均由网友自行发布提供，仅用于学习交流。如有版权问题，请与我联系，QQ：4156828
© CopyRight 2008-2024 All Rights Reserved. Powered By bs178.com 闽ICP备11008920号-3
闽公网安备35020302034844号
