springBoot整合rocketMq

　　springBoot整合rocketMq
　　因写了一个rocketMQ 发送，有一个消息发送一次，进行一下producer创建，后修改为项目启动创建生产者。
　　一、项目结构
　　二、maven相关包：     com.alibaba.rocketmq    rocketmq-client    3.5.8       com.alibaba.rocketmq    rocketmq-all    3.5.8    pom 
　　三、配置文件
　　application.ymlrockerMq:  #Name Server 地址，因为是集群部署 所以有多个用 分号 隔开 nameServer: ip:9876 #主题名称 topic: dddd #一般发送同样消息的Producer，归为同一个Group，应用必须设置，并保证命名唯一 producerGroup: dd #tag 临时值 标签 tags: mm
　　四、生产者部分
　　1、ErrorCodeimport 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、MQConsumerConfigurationimport 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、MQConsumeMsgListenerProcessorimport 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; 	} }