【项目实践04】【RocketMQ消息收发拦截器】

news/2024/4/24 17:00:09/文章来源:https://blog.csdn.net/qq_36882793/article/details/136526122

文章目录

  • 一、前言
  • 二、项目背景
  • 三、实现方案
    • 1. 关键接口
    • 2. 消息发送方
    • 3. 消息消费方
    • 4. 配置引入类
    • 5. 使用示例
  • 四、思路扩展
    • 1. 消费流程简介


一、前言

本系列用来记录一些在实际项目中的小东西,并记录在过程中想到一些小东西,因为是随笔记录,所以内容不会过于详细。

二、项目背景

原本的目的是想实现 MQ 消息之间的 TraceId 追踪。如下两个拦截器可以实现 MQ 消息之间 TraceId 的传递,不过项目后面转用 TLog 。(但是写都写了,不能浪费

三、实现方案

1. 关键接口

  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();}
    }
    
  2. 消息消费方拦截器,当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. 使用示例

  1. 编写两个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) {}
    }
    
  2. 创建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());}
    }
    
  3. 启用增强功能:服务配置中增加该属性

    oms.rocket.plus=true
    
  4. MQ 收发日志如下(我懒得开两个服务所以一个服务进行收发),收发日志如下:
    在这里插入图片描述
    可以看到:

    1. LogMqSendInterceptor 和 LogMqReceiveInterceptor 对MQ 收发日志进行了打印
    2. TraceMqSendInterceptor 和 TraceMqReceiveInterceptor 进行了 TracceId 传递:生产者由于没有获取到 TraceId 直接随机生成一个TraceId 传递,消费者接收到 TraceId 后存入上下文,当前线程再进行日志打印时携带了TraceId。

四、思路扩展

1. 消费流程简介

消息发送方的原理很简单,这里不再赘述。主要看消息接受方的原理,RocketMQ 消费者启动消费的大体流程:

  1. 服务器启动时 ListenerContainerConfiguration#afterSingletonsInstantiated 方法中会为所有被 @RocketMQMessageListener 注解修饰的对象创建一个 DefaultRocketMQListenerContainer 实例注册到容器中,并调用 DefaultRocketMQListenerContainer#start 方法启动
  2. DefaultRocketMQListenerContainer实例创建时会调用 afterPropertiesSet 方法初始化DefaultRocketMQListenerContainer#DefaultMQPushConsumer对象。在初始化的过程中会赋值messageListener 属性为 DefaultMessageListenerOrderly 或 DefaultMessageListenerConcurrently 对象。
  3. 当消费者进行消息消费时会调用 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());}});}}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.luyixian.cn/news_show_996849.aspx

如若内容造成侵权/违法违规/事实不符,请联系dt猫网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

学术论文GPT的源码解读与二次开发:从ChatPaper到gpt_academic

前言 本文的前两个部分最早是属于此旧文的《学术论文GPT的源码解读与微调&#xff1a;从ChatPaper到七月论文审稿GPT第1版》&#xff0c;但为了每一篇文章各自的内容更好的呈现&#xff0c;于是我今天做了以下三个改动 原来属于mamba第五部分的「Mamba近似工作之线性Transfor…

C++ 路径问题

目录 例1 例2 例3 例4 例5 例6 例1 62. 不同路径 1.初始化 2.当前位置的条数&#xff0c;就是上面位置的条数 &#xff0c;加上其左边位置的条数&#xff0c;dp[i][j] dp[i - 1][j] dp[i][j - 1]; 参考代码 class Solution { public:int uniquePaths(int m, int n) …

Material UI 5 学习01-按钮组件

Material UI 5 学习01-按钮组件 一、安装Material UI二、 组件1、Button组件1、基础按钮2、variant属性3、禁用按钮4、可跳转的按钮5、disableElevation属性6、按钮的点击事件onClick 2、Button按钮的颜色和尺寸1、Button按钮的颜色2、按钮自定义颜色3、Button按钮的尺寸 3、图…

VUE3中ArcGIS JsAPI 4.27 Map 隐藏地图黑色边框

问题&#xff1a; vue3中引入arcgis jsapi 地图加载后&#xff0c;点击地图会出现黑色边框&#xff0c;看起来很不协调 解决方案&#xff1a; 新建自定义CSS文件&#xff0c;输入一下样式内容&#xff0c;并在vue页面直接用import引入即可。 注意&#xff1a;直接写到vue页面…

Golang Channel 详细原理和使用技巧

1.简介 Channel(一般简写为 chan) 管道提供了一种机制:它在两个并发执行的协程之间进行同步&#xff0c;并通过传递与该管道元素类型相符的值来进行通信,它是Golang在语言层面提供的goroutine间的通信方式.通过Channel在不同的 goroutine中交换数据&#xff0c;在goroutine之间…

3Dmax中VR渲染太阳光渲染参数怎么设置?渲染100云渲染助力

我们用3Dmax建模时一些场景会用到太阳光&#xff0c;那么渲染参数是如何设置的呢&#xff1f; 我们一起来看看&#xff0c;直接上图 以上就是详细的参数设置&#xff0c;大家可以用做参考&#xff0c;如果本地渲染慢的朋友可以考虑使用云渲染100 机器多&#xff0c;渲染稳定不…

YOLOSHOW - YOLOv5 / YOLOv7 / YOLOv8 / YOLOv9 基于 Pyside6 的图形化界面

YOLOSHOW 是一个基于 PySide6&#xff08;Qt for Python&#xff09;开发的图形化界面应用程序&#xff0c;主要用于集成和可视化YOLO系列&#xff08;包括但不限于YOLOv5、YOLOv7、YOLOv8、YOLOv9&#xff09;的目标检测模型。YOLOSHOW 提供了一个用户友好的交互界面&#xff…

一文详解:Open SSL

Open SSL是 SSL (传输层安全)和 TLS (传输层安全)协议的健壮的开源实现。这些加密协议被广泛用于保护计算机网络上的通信&#xff0c;通过在两个通信应用程序之间提供隐私和数据完整性。从更实际的角度来说&#xff0c;OpenSSL 是一个工具包&#xff0c;其中包含各种命令行实用…

JavaScript中call和apply函数方法

看下下面这个代码示例&#xff1a; javascript const lufthansa {airline: Lufthansa,iataCode: LH,bookings: [],book(flightNum, name) {console.log(${name} booked a seat on ${this.airline} flight ${this.iataCode}${flightNum});}, };lufthansa.book(239, ‘IT知识一…

力扣--动态规划152.乘积最大子数组

思路分析&#xff1a; 使用动态规划&#xff0c;定义一个二维数组dp&#xff0c;其中dp[i][0]表示以第i个元素结尾的乘积最大子数组的乘积&#xff0c;dp[i][1]表示以第i个元素结尾的乘积最小子数组的乘积。初始化dp数组的第一个元素为数组的第一个元素。遍历数组&#xff0c;…

Unity背景模糊图片高斯模糊高性能的实现方案

环境&#xff1a; unity2021.3.x 效果&#xff1a; 模糊前&#xff1a; 模糊后&#xff1a; 模糊前&#xff1a; 模糊后&#xff1a; 实现核心思路(shader)&#xff1a; SubShader {CGINCLUDE#include "UnityCG.cginc"sampler2D _MainTex; // 主纹理half4 _MainTe…

持续集成(CICD)- Jenkins安装插件

文章目录 Jenkins 检查自己是否有此插件安装插件&#xff1a; 以Git 插件举例&#xff08;其他插件类似&#xff09;&#xff1a; Jenkins 检查自己是否有此插件 检查自己的jenkins是否有git插件&#xff1a;进入Manage Jenkins - 往下滑动找到Global Tool Configuration - 如…

Android使用OpenGL和FreeType绘制文字

Open GL主要是渲染图形的&#xff0c;有时候需要绘制文字&#xff0c;网上搜了一下&#xff0c;基本思路都是把文字转成位图&#xff0c;再使用Open GL纹理进行渲染。为了保证灵活性&#xff0c;我们把每个可能用到的字符都生成一个位图加载到纹理缓冲区&#xff0c;绘制字符串…

未来AI发展趋势

引言 如何实现一个乌托邦式的社会呢&#xff1f;如果直接将人类社会分为两个阵营&#xff0c;一个是以旧人类为首的阵营&#xff0c;一个是以AI抚养的新人类阵营&#xff0c;其中新人类阵营的所有资源皆由旧人类阵营提供&#xff0c;且旧人类阵营的人类都经过了化学阉割无法生…

Linux CentOS系统安装Spug并结合内网穿透实现远程访问本地运维平台

目录 前言 1. Docker安装Spug 2 . 本地访问测试 3. Linux 安装cpolar 4. 配置Spug公网访问地址 5. 公网远程访问Spug管理界面 6. 固定Spug公网地址 结语 作者简介&#xff1a; 懒大王敲代码&#xff0c;计算机专业应届生 今天给大家聊聊Linux CentOS系统安装Spug并结合…

【实验】学习实验debug,以及经验感悟

记录两次独立解决问题的过程&#xff1a; 目前来看&#xff0c;问题分为几种&#xff1a; 抄代码的时候抄错了&#xff0c;比如dim1写成dim0这种 逻辑错误&#xff0c;如果两份代码没什么差别的话&#xff0c;那么肯定是逻辑错误。 下面的两个问题都是逻辑错误&#xff0c;因为…

2024最新算法:斑翠鸟优化算法(Pied Kingfisher Optimizer ,PKO)求解23个基准函数

一、斑翠鸟优化算法 斑翠鸟优化算法&#xff08;Pied Kingfisher Optimizer ,PKO&#xff09;&#xff0c;是由Abdelazim Hussien于2024年提出的一种基于群体的新型元启发式算法&#xff0c;它从自然界中观察到的斑翠鸟独特的狩猎行为和共生关系中汲取灵感。PKO 算法围绕三个不…

FPGA- RGB_TFT显示屏原理及驱动逻辑

下图是TFT显示屏的显示效果 该显示屏共分为 2 个版本&#xff0c;4.3 寸版本的 TFT4.3’’_V3.0 和 5.0 寸版本的 TFT5.0’’_V3.0。 两者 PCB 背板电路完全相同&#xff0c;接口脚位定义完全相同&#xff0c;接口时序完全相同&#xff0c;仅使用的显示屏 模组尺寸不同。设计两…

uniapp直接连接wifi(含有ios和安卓的注意事项)

前言 小程序中直接连接wifi-----微信小程序 代码 启动 //启动wifistartWifi() {return new Promise((resolve, reject) > {uni.startWifi({success: (res) > {console.log(启动wifi 成功, res)resolve(true)},fail: (err) > {console.error(启动wifi 失败, err)uni.s…

【Web安全】XSS攻击与绕过

【Web安全】XSS攻击与绕过 【Web安全靶场】xss-labs-master 1-20 文章目录 【Web安全】XSS攻击与绕过1. XSS攻击是啥&#xff1f;2. XSS如何发生&#xff1f;3. XSS分类3.1. 反射型3.2. 存储型3.3. DOM型 4. XSS攻击方式1. script标签2. img标签3. input标签4. details标签5.…