使用starter的时候需要自定义全局拦截器在 listener onMessage 之前处理 MessageExt

texousliu commented 1 year ago

FEATURE REQUEST

使用starter的时候需要自定义全局拦截器在 listener onMessage 之前处理 MessageExt
Provide any additional detail on your proposed use case for this feature.

添加message处理器


package org.apache.rocketmq.spring.support;

import org.apache.rocketmq.common.message.MessageExt;

/**

listener message customizer
@author texousliu
@since 2023-08-18 */ public interface RocketMQListenerMessageCustomizer {

void customize(final MessageExt messageExt);

}

4. ListenerContainerConfiguration 注入自定义处理器
```java
    private List<RocketMQListenerMessageCustomizer> rocketMQListenerMessageCustomizers;

    public ListenerContainerConfiguration(RocketMQMessageConverter rocketMQMessageConverter,
                                          ConfigurableEnvironment environment, RocketMQProperties rocketMQProperties,
                                          @Autowired(required = false) List<RocketMQListenerMessageCustomizer> rocketMQListenerMessageCustomizers) {
        this.rocketMQMessageConverter = rocketMQMessageConverter;
        this.environment = environment;
        this.rocketMQProperties = rocketMQProperties;
        this.rocketMQListenerMessageCustomizers = rocketMQListenerMessageCustomizers;
    }

DefaultRocketMQListenerContainer 添加

private List<RocketMQListenerMessageCustomizer> rocketMQListenerMessageCustomizers;
public List<RocketMQListenerMessageCustomizer> getRocketMQListenerMessageCustomizers() {
    return rocketMQListenerMessageCustomizers;
}

public void setRocketMQListenerMessageCustomizers(List<RocketMQListenerMessageCustomizer> rocketMQListenerMessageCustomizers) {
    this.rocketMQListenerMessageCustomizers = rocketMQListenerMessageCustomizers;
}

ListenerContainerConfiguration 配置 DefaultRocketMQListenerContainer

container.setRocketMQListenerMessageCustomizers(rocketMQListenerMessageCustomizers);

处理消息的地方添加调用

private void handleMessage(
    MessageExt messageExt) throws MQClientException, RemotingException, InterruptedException {
    if (rocketMQListenerMessageCustomizers != null) {
        for (RocketMQListenerMessageCustomizer customizer : rocketMQListenerMessageCustomizers) {
            customizer.customize(messageExt);
        }
    }
   // ...... other code
}

Indicate the importance of this issue to you (blocker, must-have, should-have, nice-to-have). Are you currently using any workarounds to address this issue?

texousliu commented 1 year ago

我觉得这是一个比较重要的功能，在获取消息之后能够全局拦截并预先处理。比如说：需要全链路追踪的时候，可以通过这个扩展拦截器处理额外的信息到 context 中

texousliu commented 1 year ago

因为我在 starter 的源代码中没有看到相关的扩展接口，如果需要强制实现的话需要重写 ListenerContainerConfiguration，我认为这是不太友好的方式

mufengCc commented 1 year ago

写的不错，很有想法。

个人认为，作者没有实现扩展接口，可能另有考虑。

另外，如果要针对rocketMQ实现链路追踪，需要生产者在发送消息时，通过properties增加对应值。在消费者端，使用aop进行处理，也很方便

下面是我的代码示例

`@Slf4j @Aspect @Component public class TraceAspect {

private static final String TRACE_ID = "traceId";

@SneakyThrows
@Around("@annotation(mqTraceID)")
public Object before2(ProceedingJoinPoint joinPoint, MqTraceID mqTraceID) {

    Object[] args = joinPoint.getArgs();

    for (Object arg : args) {
        if (arg instanceof MessageExt messageExt) {
            Map<String, String> properties = messageExt.getProperties();
            String traceId = properties.get(TRACE_ID);
            if (StringUtils.isBlank(traceId)) {
                traceId = generateTraceId();
            }
            MDC.put(TRACE_ID, traceId);
            break;
        }
    }

    try {
        return joinPoint.proceed();
    } finally {
        MDC.clear();
    }
}

public static String generateTraceId() {
    return UUID.randomUUID().toString().replace("-", "").toLowerCase();
}

}`

消费者

Component @RocketMQMessageListener(topic = "test-topic", consumerGroup = "test-consumer-group") public class BuyRenewCloudDiskMq implements RocketMQListener{ @MqTraceID @SneakyThrows @Override public void onMessage(MessageExt messageExt) {

}