文章目录
- 一、前言
- 二、项目背景
- 三、实现方案
- 1. 关键接口
- 2. 消息发送方
- 3. 消息消费方
- 4. 配置引入类
- 5. 使用示例
- 四、思路扩展
- 1. 消费流程简介
一、前言
本系列用来记录一些在实际项目中的小东西，并记录在过程中想到一些小东西，因为是随笔记录，所以内容不会过于详细。
二、项目背景
原本的目的是想实现 MQ 消息之间的 TraceId 追踪。如下两个拦截器可以实现 MQ 消息之间 TraceId 的传递，不过项目后面转用 TLog 。（但是写都写了，不能浪费 ）
三、实现方案
1. 关键接口
-
消息发送方拦截器，当RocketMQ 消息发送的时候会调用对应的方法进行拦截，实现该接口可进行消息发送前后的扩展。
public interface MqSendInterceptor extends Ordered, Comparable<MqSendInterceptor> {/*** 消息发送前处理** @param message*/default void preHandle(Message message) {}/*** 消息发送后处理** @param destination* @param payload* @param e*/default void afterHandle(String destination, org.springframework.messaging.Message<?> payload, Exception e) {}/*** 拦截器优先级** @return*/@Overridedefault int getOrder() {return Integer.MIN_VALUE;}/*** 拦截器排序** @param o* @return*/@Overridedefault int compareTo(MqSendInterceptor o) {return getOrder() - o.getOrder();} } -
消息消费方拦截器，当RocketMQ 消息接收消费的时候会调用对应的方法进行拦截，实现该接口可进行消息消费前后的扩展。
public interface MqReceiveInterceptor extends Ordered, Comparable<MqReceiveInterceptor> {/*** 消息预处理** @param conusmerGroup* @param messageExt* @return*/void preHandle(String conusmerGroup, MessageExt messageExt);/*** 消息后处理** @param consumerGroup* @param messageExt*/void afterHandle(String consumerGroup, MessageExt messageExt, Exception exception);/*** 拦截器优先级** @return*/@Overridedefault int getOrder() {return Integer.MIN_VALUE;}/*** 拦截器排序** @param o* @return*/@Overridedefault int compareTo(MqReceiveInterceptor o) {return getOrder() - o.getOrder();} }
2. 消息发送方
消息发送的扩展逻辑实现其实很简单，直接继承的了基本的 RocketMQTemplate ，对一些方法进行了重写。重写的逻辑是在消息发送前后会调用 MqSendInterceptor 的对应方法即可。具体代码如下
@Slf4j
public class RocketMQPlusTemplate extends RocketMQTemplate implements InitializingBean {@Autowired(required = false)private List<MqSendInterceptor> mqSendInterceptors;@Overridepublic void afterPropertiesSet() throws Exception {if (mqSendInterceptors == null) {mqSendInterceptors = Lists.newArrayListWithCapacity(0);}Collections.sort(mqSendInterceptors);super.afterPropertiesSet();}@Overridepublic <T> T sendAndReceive(String destination, Message<?> message, Type type,String hashKey, long timeout, int delayLevel) {return handleMsgSupplier(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("send request message failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);if (delayLevel > 0) {rocketMsg.setDelayTimeLevel(delayLevel);}MessageExt replyMessage;if (Objects.isNull(hashKey) || hashKey.isEmpty()) {replyMessage = (MessageExt) getProducer().request(rocketMsg, timeout);} else {replyMessage = (MessageExt) getProducer().request(rocketMsg, getMessageQueueSelector(), hashKey, timeout);}return replyMessage != null ? (T) doConvertMessage(replyMessage, type) : null;} catch (Exception e) {log.error("send request message failed. destination:{}, message:{} ", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic void sendAndReceive(String destination, Message<?> message,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long t, int delayLevel) {handleMsgRunnable(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("send request message failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {long timeout = t;org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);if (delayLevel > 0) {rocketMsg.setDelayTimeLevel(delayLevel);}if (timeout <= 0) {timeout = getProducer().getSendMsgTimeout();}RequestCallback requestCallback = null;if (rocketMQLocalRequestCallback != null) {requestCallback = new RequestCallback() {@Overridepublic void onSuccess(org.apache.rocketmq.common.message.Message message) {rocketMQLocalRequestCallback.onSuccess(doConvertMessage((MessageExt) message, getMessageType(rocketMQLocalRequestCallback)));}@Overridepublic void onException(Throwable e) {rocketMQLocalRequestCallback.onException(e);}};}if (Objects.isNull(hashKey) || hashKey.isEmpty()) {getProducer().request(rocketMsg, requestCallback, timeout);} else {getProducer().request(rocketMsg, getMessageQueueSelector(), hashKey, requestCallback, timeout);}} catch (Exception e) {log.error("send request message failed. destination:{}, message:{} ", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic <T extends Message> SendResult syncSend(String destination, Collection<T> messages, long timeout) {return handleMsgsSupplier(() -> {if (Objects.isNull(messages) || messages.size() == 0) {log.error("syncSend with batch failed. destination:{}, messages is empty ", destination);throw new IllegalArgumentException("`messages` can not be empty");}try {long now = System.currentTimeMillis();Collection<org.apache.rocketmq.common.message.Message> rmqMsgs = new ArrayList<>();for (Message msg : messages) {if (Objects.isNull(msg) || Objects.isNull(msg.getPayload())) {log.warn("Found a message empty in the batch, skip it");continue;}rmqMsgs.add(this.createRocketMqMessage(destination, msg));}SendResult sendResult = getProducer().send(rmqMsgs, timeout);long costTime = System.currentTimeMillis() - now;if (log.isDebugEnabled()) {log.debug("send messages cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());}return sendResult;} catch (Exception e) {log.error("syncSend with batch failed. destination:{}, messages.size:{} ", destination, messages.size());throw new MessagingException(e.getMessage(), e);}}, destination, messages);}@Overridepublic SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout) {return handleMsgSupplier(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("syncSendOrderly failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {long now = System.currentTimeMillis();org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);SendResult sendResult = getProducer().send(rocketMsg, getMessageQueueSelector(), hashKey, timeout);long costTime = System.currentTimeMillis() - now;if (log.isDebugEnabled()) {log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());}return sendResult;} catch (Exception e) {log.error("syncSendOrderly failed. destination:{}, message:{} ", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic SendResult syncSend(String destination, Message<?> message, long timeout, int delayLevel) {return handleMsgSupplier(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("syncSend failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {long now = System.currentTimeMillis();org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);if (delayLevel > 0) {rocketMsg.setDelayTimeLevel(delayLevel);}SendResult sendResult = getProducer().send(rocketMsg, timeout);long costTime = System.currentTimeMillis() - now;if (log.isDebugEnabled()) {log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());}return sendResult;} catch (Exception e) {log.error("syncSend failed. destination:{}, message:{} ", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout,int delayLevel) {handleMsgRunnable(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("asyncSend failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);if (delayLevel > 0) {rocketMsg.setDelayTimeLevel(delayLevel);}getProducer().send(rocketMsg, sendCallback, timeout);} catch (Exception e) {log.info("asyncSend failed. destination:{}, message:{} ", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic void asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback,long timeout) {handleMsgRunnable(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("asyncSendOrderly failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);getProducer().send(rocketMsg, getMessageQueueSelector(), hashKey, sendCallback, timeout);} catch (Exception e) {log.error("asyncSendOrderly failed. destination:{}, message:{} ", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic void sendOneWay(String destination, Message<?> message) {handleMsgRunnable(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("sendOneWay failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);getProducer().sendOneway(rocketMsg);} catch (Exception e) {log.error("sendOneWay failed. destination:{}, message:{} ", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic void sendOneWayOrderly(String destination, Message<?> message, String hashKey) {handleMsgRunnable(() -> {if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {log.error("sendOneWayOrderly failed. destination:{}, message is null ", destination);throw new IllegalArgumentException("`message` and `message.payload` cannot be null");}try {org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);getProducer().sendOneway(rocketMsg, getMessageQueueSelector(), hashKey);} catch (Exception e) {log.error("sendOneWayOrderly failed. destination:{}, message:{}", destination, message);throw new MessagingException(e.getMessage(), e);}}, destination, message);}@Overridepublic TransactionSendResult sendMessageInTransaction(final String destination,final Message<?> message, final Object arg) throws MessagingException {return handleMsgSupplier(() -> {try {if (((TransactionMQProducer) getProducer()).getTransactionListener() == null) {throw new IllegalStateException("The rocketMQTemplate does not exist TransactionListener");}org.apache.rocketmq.common.message.Message rocketMsg = this.createRocketMqMessage(destination, message);return getProducer().sendMessageInTransaction(rocketMsg, arg);} catch (MQClientException e) {throw RocketMQUtil.convert(e);}}, destination, message);}protected org.apache.rocketmq.common.message.Message createRocketMqMessage(String destination, Message<?> message) {Message<?> msg = this.doConvert(message.getPayload(), message.getHeaders(), null);final org.apache.rocketmq.common.message.Message rocketMessage =RocketMQUtil.convertToRocketMessage(getMessageConverter(), getCharset(),destination, msg);preHandle(rocketMessage);return rocketMessage;}private Object doConvertMessage(MessageExt messageExt, Type type) {if (Objects.equals(type, MessageExt.class)) {return messageExt;} else if (Objects.equals(type, byte[].class)) {return messageExt.getBody();} else {String str = new String(messageExt.getBody(), Charset.forName(getCharset()));if (Objects.equals(type, String.class)) {return str;} else {// If msgType not string, use objectMapper change it.try {if (type instanceof Class) {//if the messageType has not Generic Parameterreturn this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) type);} else {//if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third parameter "conversionHint".//we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) type).getRawType(), null);}} catch (Exception e) {log.error("convert failed. str:{}, msgType:{}", str, type);throw new RuntimeException("cannot convert message to " + type, e);}}}}private Type getMessageType(RocketMQLocalRequestCallback rocketMQLocalRequestCallback) {Class<?> targetClass = AopProxyUtils.ultimateTargetClass(rocketMQLocalRequestCallback);Type matchedGenericInterface = null;while (Objects.nonNull(targetClass)) {Type[] interfaces = targetClass.getGenericInterfaces();if (Objects.nonNull(interfaces)) {for (Type type : interfaces) {if (type instanceof ParameterizedType && (Objects.equals(((ParameterizedType) type).getRawType(), RocketMQLocalRequestCallback.class))) {matchedGenericInterface = type;break;}}}targetClass = targetClass.getSuperclass();}if (Objects.isNull(matchedGenericInterface)) {return Object.class;}Type[] actualTypeArguments = ((ParameterizedType) matchedGenericInterface).getActualTypeArguments();if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {return actualTypeArguments[0];}return Object.class;}/*** 消息** @param runnable* @param destination* @param message*/private void handleMsgRunnable(Runnable runnable, String destination, Message<?> message) {Exception exception = null;try {runnable.run();} catch (Exception e) {exception = e;throw e;} finally {afterHandle(destination, message, exception);}}/*** 消息处理** @param supplier* @param destination* @param message* @return*/private <R> R handleMsgSupplier(Supplier<R> supplier, String destination, Message<?> message) {Exception exception = null;try {return supplier.get();} catch (Exception e) {exception = e;throw e;} finally {afterHandle(destination, message, exception);}}/*** 消息** @param supplier* @param destination* @param messages* @param <T>* @param <P>* @return*/private <T, P extends Message<?>> T handleMsgsSupplier(Supplier<T> supplier, String destination, Collection<P> messages) {Exception exception = null;try {return supplier.get();} catch (Exception e) {exception = e;throw e;} finally {for (Message<?> message : messages) {afterHandle(destination, message, exception);}}}/*** 消息前置处理** @param message*/protected void preHandle(org.apache.rocketmq.common.message.Message message) {if (!Boolean.TRUE.toString().equalsIgnoreCase(message.getUserProperty(MqContants.NOT_INTERCEPT))) {for (MqSendInterceptor mqSendInterceptor : mqSendInterceptors) {mqSendInterceptor.preHandle(message);}}}/*** 消息后置处理** @param destination* @param message* @param exception*/protected void afterHandle(String destination, Message<?> message, Exception exception) {if (!Boolean.TRUE.equals(message.getHeaders().get(MqContants.NOT_INTERCEPT))) {for (MqSendInterceptor mqSendInterceptor : mqSendInterceptors) {mqSendInterceptor.afterHandle(destination, message, exception);}}log.warn("[mq发送][topic = {}, payload = {}, exception = {}]", destination, message.getPayload(), exception);}
}
3. 消息消费方
消息消费方的实现会更加复杂一点，通过 BeanPostProcessor 接口在 DefaultRocketMQListenerContainer 对象创建的时候为其创建代理对象。如下：
public class RocketMqBeanPostProcessor implements InstantiationAwareBeanPostProcessor, InitializingBean {@Autowired(required = false)private List<MqReceiveInterceptor> mqReceiveInterceptors;/*** 获取消费者组** @param container* @return*/private static String getConusmerGroup(DefaultRocketMQListenerContainer container) {String consumerGroup = null;RocketMQListener<?> rocketMQListener = container.getRocketMQListener();if (rocketMQListener != null) {final RocketMQMessageListener annotation =rocketMQListener.getClass().getAnnotation(RocketMQMessageListener.class);if (annotation != null) {consumerGroup = annotation.consumerGroup();}} else {RocketMQReplyListener<?, ?> rocketMQReplyListener = container.getRocketMQReplyListener();if (rocketMQReplyListener != null) {final RocketMQMessageListener annotation =rocketMQReplyListener.getClass().getAnnotation(RocketMQMessageListener.class);if (annotation != null) {consumerGroup = annotation.consumerGroup();}}}return consumerGroup;}@Overridepublic void afterPropertiesSet() {if (mqReceiveInterceptors == null) {mqReceiveInterceptors = Lists.newArrayListWithCapacity(0);}Collections.sort(mqReceiveInterceptors);}@Overridepublic Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {// 对 DefaultRocketMQListenerContainer 处理，代理 messageListener 对象// 引入RocketMq 对消息的处理是交由 DefaultRocketMQListenerContainer 中的 messageListener 对象的if (bean instanceof DefaultRocketMQListenerContainer) {final DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;final DefaultMQPushConsumer consumer = container.getConsumer();if (consumer != null) {final MessageListener messageListener = consumer.getMessageListener();if (messageListener instanceof MessageListenerOrderly|| messageListener instanceof MessageListenerConcurrently) {// 重新设置代理后的对象consumer.setMessageListener((MessageListener)createProxy(getConusmerGroup(container), messageListener));}}}return bean;}/*** 为 MessageListener 创建代理对象** @param consumerGroup* @param messageListener* @return*/private Object createProxy(String consumerGroup, MessageListener messageListener) {return Proxy.newProxyInstance(MessageListener.class.getClassLoader(), messageListener.getClass().getInterfaces(),(proxy, method, args) -> {// 代理对象判断如果调用的是 consumeMessage 方法则进行增强，调用前后调用消息拦截器。if ("consumeMessage".equals(method.getName())) {Exception exception = null;try {for (MessageExt message : (List<MessageExt>) args[0]) {if (!Boolean.TRUE.toString().equalsIgnoreCase(message.getUserProperty(MqContants.NOT_INTERCEPT))) {for (MqReceiveInterceptor mqReceiveInterceptor : mqReceiveInterceptors) {mqReceiveInterceptor.preHandle(consumerGroup, message);}}}return method.invoke(messageListener, args);} catch (Exception e) {exception = e;throw e;} finally {for (MessageExt message : (List<MessageExt>) args[0]) {if (!Boolean.TRUE.toString().equalsIgnoreCase(message.getUserProperty(MqContants.NOT_INTERCEPT))) {for (MqReceiveInterceptor mqReceiveInterceptor : mqReceiveInterceptors) {mqReceiveInterceptor.afterHandle(consumerGroup, message, exception);}}}}}return method.invoke(messageListener, args);});}}
4. 配置引入类
项目实际上是做了一个 spring-boot-starter 的jar 包供其他服务引用的因此，这里是通过自动装配将 RocketMQAutoConfig 引入，然后 RocketMQAutoConfig 通过 @Import 引入了 RocketmqConfig 类。
@Import(RocketmqConfig.class)
public class RocketMQAutoConfig {
}
其中 RocketmqConfig 的实现如下:
@Configuration
@Configuration
public class RocketmqConfig implements ApplicationContextAware {public static final String PRODUCER_BEAN_NAME = "defaultMQProducer";public static final String CONSUMER_BEAN_NAME = "defaultLitePullConsumer";private ApplicationContext applicationContext;@Value("${rocketmq.name-server}")private String nameServer;@Value("${rocketmq.producer.group:default}")private String producerGroup;@Value("${rocketmq.producer.sendMessageTimeout}")private int sendMessageTimeout;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}/*** DefaultMQProducer 注入* @return*/@Bean("defaultMQProducer")public DefaultMQProducer defaultMQProducer() {DefaultMQProducer defaultMQProducer = new DefaultMQProducer();defaultMQProducer.setProducerGroup(producerGroup);defaultMQProducer.setNamesrvAddr(nameServer);defaultMQProducer.setVipChannelEnabled(false);defaultMQProducer.setSendMsgTimeout(sendMessageTimeout);return defaultMQProducer;}/*** MQ 消息转换器* @return*/@Bean@ConditionalOnProperty(prefix = "oms", name = "rocket.bak", havingValue = "true")@ConditionalOnMissingBean(RocketMQMessageConverter.class)public RocketMQMessageConverter rocketmqMessageConverter() {return new RocketMQMessageConverter();}/*** mq 扩展，支持发送前后拦截* 这里直接参考 org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration#rocketMQTemplate(org.apache.rocketmq.spring.support.RocketMQMessageConverter) 的实现** @param rocketMQMessageConverter* @return*/@Bean(destroyMethod = "destroy")@ConditionalOnProperty(prefix = "oms", name = "rocket.plus", havingValue = "true")public RocketMQTemplate rocketMQTemplate(RocketMQMessageConverter rocketMQMessageConverter) {RocketMQTemplate rocketMQTemplate = new RocketMQPlusTemplate();if (applicationContext.containsBean(PRODUCER_BEAN_NAME)) {rocketMQTemplate.setProducer((DefaultMQProducer) applicationContext.getBean(PRODUCER_BEAN_NAME));}if (applicationContext.containsBean(CONSUMER_BEAN_NAME)) {rocketMQTemplate.setConsumer((DefaultLitePullConsumer) applicationContext.getBean(CONSUMER_BEAN_NAME));}rocketMQTemplate.setMessageConverter(rocketMQMessageConverter.getMessageConverter());return rocketMQTemplate;}/*** mq 扩展，支持接收前后拦截** @return*/@Bean@ConditionalOnProperty(prefix = "oms", name = "rocket.plus", havingValue = "true")public RocketMqBeanPostProcessor rocketMqBeanPostProcessor() {return new RocketMqBeanPostProcessor();}}
这里为 RocketMQ 收发增强功能做了一个配置项，如果需要开启增强功能则需要再服务的配置项中增加如下配置 :
# mq 扩展功能开启
oms.rocket.plus=true
5. 使用示例
-
编写两个MQ 日志打印拦截器
// 日志打印的两个拦截器 @Slf4j public class LogMqSendInterceptor implements MqSendInterceptor {@Overridepublic void afterHandle(String destination, Message<?> payload, Exception e) {try {log.info("[{} 发送消息][payLoad = {}, exception = {}]",destination, JSON.toJSONString(payload.getPayload()), e);} catch (Exception ex) {log.debug("[{} 日志打印错误]", destination, ex);}} }@Slf4j public class LogMqReceiveInterceptor implements MqReceiveInterceptor {@Overridepublic void preHandle(String conusmerGroup, MessageExt messageExt) {try {log.info("[{} 收到消息][conusmerGroup = {}, message = {}]",messageExt.getTopic(), conusmerGroup, new String(messageExt.getBody()));} catch (Exception e) {log.debug("[{} 日志打印错误]", messageExt.getTopic(), e);}}@Overridepublic void afterHandle(String consumerGroup, MessageExt messageExt, Exception exception) {} }/******************************************/ // 日志追踪的两个拦截器 @Slf4j public class TraceMqSendInterceptor implements MqSendInterceptor {@Value("${spring.application.name:''}")private String applicationName;@Overridepublic void preHandle(Message message) {String traceId = MDC.get(RpcConstants.MDC_TRACE_ID);if (StringUtils.isBlank(traceId)) {traceId = UUID.randomUUID().toString();log.info("[发送到Mq 消息][traceId 为空,随机生成 traceId = {}]", traceId);}message.putUserProperty(RpcConstants.MDC_TRACE_ID, traceId);if (StringUtils.isNotBlank(applicationName)) {message.putUserProperty(RpcConstants.APPLICATION_NAME, applicationName);}}@Overridepublic void afterHandle(String destination, org.springframework.messaging.Message<?> message, Exception e) {}}@Slf4j public class TraceMqReceiveInterceptor implements MqReceiveInterceptor {@Overridepublic void preHandle(String conusmerGroup, MessageExt messageExt) {String traceId = messageExt.getUserProperty(RpcConstants.MDC_TRACE_ID);final String applicationName = messageExt.getUserProperty(RpcConstants.APPLICATION_NAME);log.info("[接收到Mq 消息][messageId = {}, traceId = {}, sendAppName = {}]",messageExt.getMsgId(), traceId, applicationName);if (StringUtils.isBlank(traceId)) {traceId = UUID.randomUUID().toString();log.info("[接收到Mq 消息][traceId 为空, messageId = {}, 随机生成 traceId = {}]",messageExt.getMsgId(), traceId);}MDC.put(RpcConstants.MDC_TRACE_ID, traceId);}@Overridepublic void afterHandle(String consumerGroup, MessageExt messageExt, Exception exception) {} } -
创建MQ消息收发者
// 消费者 @Slf4j @Component @RocketMQMessageListener(topic = "topic01", consumerGroup = "cg01") public class DemoMQListener implements RocketMQListener<String> {@Autowiredprivate ApplicationEventPublisher applicationEventPublisher;@Overridepublic void onMessage(String message) {log.info("收到消息 {}", message);}}// 生产者 @Component public class DemoRunner implements ApplicationRunner {@Autowiredprivate RocketMQTemplate rocketMQTemplate;@Overridepublic void run(ApplicationArguments args) throws Exception {rocketMQTemplate.send("topic01", MessageBuilder.withPayload("hello world").build());} } -
启用增强功能：服务配置中增加该属性
oms.rocket.plus=true -
MQ 收发日志如下（我懒得开两个服务所以一个服务进行收发）,收发日志如下：
可以看到：- LogMqSendInterceptor 和 LogMqReceiveInterceptor 对MQ 收发日志进行了打印
- TraceMqSendInterceptor 和 TraceMqReceiveInterceptor 进行了 TracceId 传递：生产者由于没有获取到 TraceId 直接随机生成一个TraceId 传递，消费者接收到 TraceId 后存入上下文，当前线程再进行日志打印时携带了TraceId。
四、思路扩展
1. 消费流程简介
消息发送方的原理很简单，这里不再赘述。主要看消息接受方的原理，RocketMQ 消费者启动消费的大体流程：
- 服务器启动时 ListenerContainerConfiguration#afterSingletonsInstantiated 方法中会为所有被 @RocketMQMessageListener 注解修饰的对象创建一个 DefaultRocketMQListenerContainer 实例注册到容器中，并调用 DefaultRocketMQListenerContainer#start 方法启动
- DefaultRocketMQListenerContainer实例创建时会调用 afterPropertiesSet 方法初始化DefaultRocketMQListenerContainer#DefaultMQPushConsumer对象。在初始化的过程中会赋值messageListener 属性为 DefaultMessageListenerOrderly 或 DefaultMessageListenerConcurrently 对象。
- 当消费者进行消息消费时会调用 DefaultMessageListenerOrderly 或 DefaultMessageListenerConcurrently 的 consumeMessage 方法，而在该方法中会去调用我们自己注入的被 @RocketMQMessageListener 注解修饰的类的 onMessage 方法。
其中具体代码如下：
@Overridepublic void afterSingletonsInstantiated() {// 扫描容器中所有被 RocketMQMessageListener 注解修饰的类Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class).entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey())).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));// 对消费方进行注册beans.forEach(this::registerContainer);}private void registerContainer(String beanName, Object bean) {... if (!RocketMQListener.class.isAssignableFrom(bean.getClass()) && !RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName() + " or " + RocketMQReplyListener.class.getName());}// 获取并解析注解信息RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);String consumerGroup = this.environment.resolvePlaceholders(annotation.consumerGroup());String topic = this.environment.resolvePlaceholders(annotation.topic());... // 注解校验validate(annotation);// 获取当前对象在 容器中的beanName,其中通过 AtomicLong 的自增来确保 BeanName 的唯一性String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),counter.incrementAndGet());GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;// 创建 DefaultRocketMQListenerContainer 并注册到容器中。genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,() -> createRocketMQListenerContainer(containerBeanName, bean, annotation));// 获取容器中刚注册的 DefaultRocketMQListenerContainer 判断其如果没有启动则启动。DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,DefaultRocketMQListenerContainer.class);if (!container.isRunning()) {try {container.start();} catch (Exception e) {log.error("Started container failed. {}", container, e);throw new RuntimeException(e);}}log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);}// 创建 DefaultRocketMQListenerContainer private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean,RocketMQMessageListener annotation) {DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();container.setRocketMQMessageListener(annotation);String nameServer = environment.resolvePlaceholders(annotation.nameServer());nameServer = StringUtils.isEmpty(nameServer) ? rocketMQProperties.getNameServer() : nameServer;String accessChannel = environment.resolvePlaceholders(annotation.accessChannel());container.setNameServer(nameServer);if (!StringUtils.isEmpty(accessChannel)) {container.setAccessChannel(AccessChannel.valueOf(accessChannel));}container.setTopic(environment.resolvePlaceholders(annotation.topic()));String tags = environment.resolvePlaceholders(annotation.selectorExpression());if (!StringUtils.isEmpty(tags)) {container.setSelectorExpression(tags);}container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));// 将实际的 Bean 保存到 DefaultRocketMQListenerContainer#rocketMQListener 或 rocketMQReplyListener 属性中if (RocketMQListener.class.isAssignableFrom(bean.getClass())) {container.setRocketMQListener((RocketMQListener) bean);} else if (RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {container.setRocketMQReplyListener((RocketMQReplyListener) bean);}container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());container.setName(name);return container;}
DefaultRocketMQListenerContainer 实现了 InitializingBean 接口，因此在其实例创建的时候会调用 afterPropertiesSet 方法，在启动会初始化 DefaultRocketMQListenerContainer #DefaultMQPushConsumer 对象，这里我们主要关注 DefaultMessageListenerOrderly 和 DefaultMessageListenerConcurrently 两个类。
@Overridepublic void afterPropertiesSet() throws Exception {initRocketMQPushConsumer();this.messageType = getMessageType();this.methodParameter = getMethodParameter();log.debug("RocketMQ messageType: {}", messageType);}//初始化 consumer 对象private void initRocketMQPushConsumer() throws MQClientException {...// 消费模式：顺序还是并发，根据不同的模式赋值不同的 MessageListenerswitch (consumeMode) {case ORDERLY:consumer.setMessageListener(new DefaultMessageListenerOrderly());break;case CONCURRENTLY:consumer.setMessageListener(new DefaultMessageListenerConcurrently());break;default:throw new IllegalArgumentException("Property 'consumeMode' was wrong.");}... }
而在 DefaultMessageListenerConcurrently 和 DefaultMessageListenerOrderly中消息的处理是交由 rocketMQListener#onMessage 来处理。
public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {@SuppressWarnings("unchecked")@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt messageExt : msgs) {log.debug("received msg: {}", messageExt);try {long now = System.currentTimeMillis();handleMessage(messageExt);long costTime = System.currentTimeMillis() - now;log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);} catch (Exception e) {log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e);context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);return ConsumeConcurrentlyStatus.RECONSUME_LATER;}}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}}public class DefaultMessageListenerOrderly implements MessageListenerOrderly {@SuppressWarnings("unchecked")@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {for (MessageExt messageExt : msgs) {log.debug("received msg: {}", messageExt);try {long now = System.currentTimeMillis();handleMessage(messageExt);long costTime = System.currentTimeMillis() - now;log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);} catch (Exception e) {log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e);context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;}}return ConsumeOrderlyStatus.SUCCESS;}}private void handleMessage(MessageExt messageExt) throws MQClientException, RemotingException, InterruptedException {if (rocketMQListener != null) {rocketMQListener.onMessage(doConvertMessage(messageExt));} else if (rocketMQReplyListener != null) {Object replyContent = rocketMQReplyListener.onMessage(doConvertMessage(messageExt));Message<?> message = MessageBuilder.withPayload(replyContent).build();org.apache.rocketmq.common.message.Message replyMessage = MessageUtil.createReplyMessage(messageExt, convertToBytes(message));consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer().send(replyMessage, new SendCallback() {@Override public void onSuccess(SendResult sendResult) {if (sendResult.getSendStatus() != SendStatus.SEND_OK) {log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());} else {log.debug("Consumer replies message success.");}}@Override public void onException(Throwable e) {log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());}});}}
![[HackMyVM]靶场 Wild](https://img-blog.csdnimg.cn/direct/b4c47be59eff48bd8e18ff09c60609e3.png)
