SpringBoot RocketMQ消息监听器(RocketMQMessageListener)后置注册_rocketmqlistener

一、问题说明

此篇文章会涉及到SpringBoot RocketMQ源码，后面会单独出一篇详细介绍源码

1. SpringBoot RocketMQ可以从源码中查看到实现了SmartInitializingSingleton接口，在所有Bean注册完成之后就会开始注册，注册使用的是main线程，就会导致如果消息队列积压大量消息时注册占用main线程时间过长，导致SpringBoot不能完全完成初始化（主要影响web无法对外提供接口访问）

• 源码注册入口：org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration#afterSingletonsInstantiated

2. 例如生产环境使用k8s集群部署，当消息积压过大时无法快速完成启动对外提供就绪检查接口，就会导致pod就绪检查失败，从而pod不停重启无法完成正常启动(实际上服务在正常处理消息，消费占用了初始化时长)

本次主要使用RocketMQ消息监听器后置注册方案，k8s层面也可以解决，但是不在本次讨论范围之内

RocketMQ消息监听器后置注册：就是在SpringBoot启动完成可以正常接收外部接口请求时再动态注册消息加监听器

本篇文章初衷如下

• 主要讨论RocketMQ如果在SpringBoot中后置注册，也就是我们自己控制注册顺序，而不是SpringBoot自动装配完成注册
• 讨论的方案不涉及改动源码，而是直接通过正常方式达到后置注册目的

主要解决

• 不希望在服务启动时就注册消息监听器，而是服务完全启动后再开始注册消息监听器

前置文章参考

• SpringBoot基础整合RocketMQ前面有发布过两篇文章，基础整合不会再过多说明，有需要可以参考下面的文章
• RocketMQ与SpringBoot整合、核心使用、多租户自动隔离、Java8时间支持 - 掘金 (juejin.cn)
• RocketMQ与SpringBoot整合进行生产级二次封装 - 掘金 (juejin.cn)

二、处理方案

1. 通过上面的源码可以看到，afterSingletonsInstantiated方法完成消息监听器的注册，所以此时就需要拦截消息监听器的注册，拦截注册有两个方法

• 改源码（不属于本篇文章讨论范围，同时也不建议，除非公司有专门维护的同事）
• 通过AOP切片，拦截方法注册

2. AOP切片原理

• 在SpringAOP中，我们可以通过 @Around 注解切入一个方法，只有我们自己调用ProceedingJoinPoint的proceed() 方法时才会触发方法本身的调用
• 所以原理就是通过@Around切入方法，但是不调用proceed()方法，直接返回，就可以完成拦截的目的

三、代码示例

3.1 AOP拦截RocketMQMessageListener注册

• 代码示例如下

import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

/**
 * RocketMQ注册拦截，自行注册，加快容器启动速度
 *
 * @author tianxincoord@163.com
 * @since 2022/8/15
 */
@Component
@Aspect
@Order(1)
@Slf4j
public class RocketMqSkipRegisterAop {
    /**
     * 通过切入afterSingletonsInstantiated拦截注册
     */
    @Around("execution(* org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.afterSingletonsInstantiated())")
    public Object init(ProceedingJoinPoint joinPoint) {
        log.info("RocketMQ开启代理，默认注册已跳过");
        /// 不调用原生执行方法，直接返回null，跳过方法调用
        // Object proceed = joinPoint.proceed();
        // return proceed
        return null;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

3.2 编写自定义注册类

由于ListenerContainerConfiguration注册类方法被切面拦截，所以可以复制一个这个类出来保留里面的方法，然后调整一下参数等信息，根据自己需要，也可以直接使用实例中的代码


示例类代码和ListenerContainerConfiguration基本一致，去掉继承关系和接口实现，当做普通类

import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.util.StringUtils;

import java.util.Collections;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * RocketMQ队列监听器
 *
 * @author tianxincoord@163.com
 * @since 2022/09/23
 */
public class RocketMqListenerRegistry {
    private final static Logger log = LoggerFactory.getLogger(
            org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.class);

    private final ConfigurableApplicationContext applicationContext;

    private final AtomicLong counter = new AtomicLong(0);

    private final StandardEnvironment environment;

    private final RocketMQProperties rocketMqProperties;

    private final RocketMQMessageConverter rocketMqMessageConverter;

    public RocketMqListenerRegistry(ConfigurableApplicationContext applicationContext,
                                    RocketMQMessageConverter rocketMqMessageConverter,
                                    StandardEnvironment environment,
                                    RocketMQProperties rocketMqProperties) {
        this.applicationContext = applicationContext;
        this.rocketMqMessageConverter = rocketMqMessageConverter;
        this.environment = environment;
        this.rocketMqProperties = rocketMqProperties;
    }


    public void afterSingletonsInstantiated() {
        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);
    }

    public void registerContainer(String beanName, Object bean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);

        if (RocketMQListener.class.isAssignableFrom(bean.getClass()) && RocketMQReplyListener.class.isAssignableFrom(
                bean.getClass())) {
            throw new IllegalStateException(
                    clazz + " cannot be both instance of " + RocketMQListener.class.getName() + " and " + RocketMQReplyListener.class.getName());
        }

        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());

        boolean listenerEnabled = (boolean) rocketMqProperties.getConsumer().getListeners()
                .getOrDefault(consumerGroup, Collections.EMPTY_MAP).getOrDefault(topic, true);

        if (!listenerEnabled) {
            log.debug(
                    "Consumer Listener (group:{},topic:{}) is not enabled by configuration, will ignore initialization.",
                    consumerGroup, topic);
            return;
        }
        validate(annotation);

        String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
                counter.incrementAndGet());
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;

        genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
                () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
        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);
    }

    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()));
        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;
    }

    private void validate(RocketMQMessageListener annotation) {
        if (annotation.consumeMode() == ConsumeMode.ORDERLY && annotation.messageModel() == MessageModel.BROADCASTING) {
            throw new BeanDefinitionValidationException(
                    "Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

3.3 启动类启动自定义注册

在启动类中单独在SpringApplication.run(xxx.class, args);之后开始注册。SpringApplication.run执行完成才表示整个SpringBoot服务真正意义上的启动完成

import com.codecoord.rocketmq.config.RocketMqListenerRegistry;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.StandardEnvironment;

/**
 * RocketMQ启动类
 */
@Slf4j
@SpringBootApplication
public class RocketMqApplication {
    public static void main(String[] args) {
        /*
         * 指定使用的日志框架，否则将会报错
         * RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.InternalThreadLocalMap).
         * RocketMQLog:WARN Please initialize the logger system properly.
         */
        System.setProperty("rocketmq.client.logUseSlf4j", "true");
        ConfigurableApplicationContext run = SpringApplication.run(RocketMqApplication.class, args);
        log.info("SpringBoot RocketMQ服务启动完成");
        ConfigurableEnvironment environment = run.getEnvironment();
        registerRocketMq(environment, run);
    }

    private static void registerRocketMq(ConfigurableEnvironment environment,
                                         ConfigurableApplicationContext applicationContext) {
        try {
            log.info("开始注册RocketMQ");
            RocketMQMessageConverter messageConverter = applicationContext.getBean(RocketMQMessageConverter.class);
            RocketMQProperties mqProperties = applicationContext.getBean(RocketMQProperties.class);
            RocketMqListenerRegistry registry = new RocketMqListenerRegistry(applicationContext, messageConverter, (StandardEnvironment) environment, mqProperties);
            // 手工调用此方法完成原来应该完成的注册调用
            registry.afterSingletonsInstantiated();
            log.info("RocketMQ注册成功");
        } catch (Exception e) {
            // 此处可以整合消息通知，当注册失败时发送企业微信、钉钉、邮件等告警
            log.error("RocketMQ注册异常", e);
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

• 上述注册也可以单独拆成的组件类，这样甚至可以通过controller接收命令的方式完成注册，美滋滋

四、代码测试

1. 启动代码观察注册过程

2. 通过以上，可以保证服务快速启动，尽快提供web服务来告诉外界服务正常

五、特殊说明

1. 如果项目是web和任务是分离的，那么上面的问题不存在，因为任务处理服务不需要处理web服务
2. 欢迎大家留言探讨