Spring整合RocketMQ

曾经终败给现在 2022-05-29 10:35 229阅读 0赞

RocketMQ在使用的时候需要定义一些列的信息，然后注册一个MessageListener，不是很方便。能不能通过Spring整合RocketMQ，像整合RabbitMQ，Kafka这样通过注解@RabbitListener，@KafkaListener的标识在方法上的形式直接生成一个客户端，简单方便。

有了这样的想法，稍微构思了下花了一天多的时间写了一个初版，具备了基本的功能，通过注解@RocketListener标识方法的形式直接生成一个消费者客户端，项目运行后能订阅指定的Broker消费消息。花的时间不多，写的仓促，如果以后有大的调整或者改进会重新放出。

下面看整合的代码结构：  
![这里写图片描述][SouthEast]

红框内是整合的代码，另外两个类和一个配置文件是使用的形式。

下面逐个的看类的结构和相关的说明：

**EnableRocket**：引入RocketMQ整合框架的注解，在配置类上添加此注解就启动了整合解析，最主要的作用就是引入了整合的配置类RocketBootstrapConfiguration。

import java.lang.annotation.Documented;
    import java.lang.annotation.ElementType;
    import java.lang.annotation.Retention;
    import java.lang.annotation.RetentionPolicy;
    import java.lang.annotation.Target;
    
    import com.bob.intergrate.rocket.config.RocketBootstrapConfiguration;
    import org.springframework.context.annotation.Import;
    
    /** * RocketMQ整合注解 * * @author wb-jjb318191 * @create 2018-03-20 9:23 */
    @Documented
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ ElementType.TYPE, ElementType.ANNOTATION_TYPE })
    @Import(RocketBootstrapConfiguration.class)
    public @interface EnableRocket { 
    }

**RocketListener** ：标识方法的注解，将指定方法标识为一个RocketMQ的消费器，框架启动时将此方法封装成一个MessageListener，配合注解上的信息和环境配置生成一个Consumer。注解上的可配置选项可以根据实际需要自行添加，在解析时相应加上就可以了，看流程很容易明白的。

import java.lang.annotation.Documented;
    import java.lang.annotation.ElementType;
    import java.lang.annotation.Retention;
    import java.lang.annotation.RetentionPolicy;
    import java.lang.annotation.Target;
    
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    
    import static org.apache.rocketmq.common.consumer.ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;
    
    /** * RocketMQ消费者监听 * * @author wb-jjb318191 * @create 2018-03-20 9:19 */
    @Documented
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ElementType.METHOD})
    public @interface RocketListener { 
    
        /** * 消费者组 * * @return */
        String consumerGroup();
    
        /** * @return */
        String topic();
    
        /** * @return */
        String tag() default "*";
    
        /** * @return */
        String namesrvAddr() default "";
    
        /** * 从哪里开始消费 * * @return */
        ConsumeFromWhere consumeFromWhere() default CONSUME_FROM_LAST_OFFSET;
    
    }

**RocketBootstrapConfiguration**：Spring整合RocketMQ的配置类，定义了3个Bean，在Spring启动过程中依次发挥作用。

import org.springframework.beans.factory.config.BeanDefinition;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Role;
    
    /** * Rocket整合配置类 * * @author wb-jjb318191 * @create 2018-03-20 9:24 */
    @Configuration
    public class RocketBootstrapConfiguration { 
    
        @Bean
        @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
        public RocketListenerAnnotationPostProcessor rocketListenerAnnotationProcessor() {
            return new RocketListenerAnnotationPostProcessor();
        }
    
        @Bean
        @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
        public RocketListenerAnnotationBeanPostProcessor rocketListenerAnnotationBeanPostProcessor() {
            return new RocketListenerAnnotationBeanPostProcessor();
        }
    
        @Bean
        @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
        public RocketConsumerLifecycleProcessor rocketConsumerLifecycleProcessor() {
            return new RocketConsumerLifecycleProcessor();
        }
    
    }

**RocketListenerAnnotationPostProcessor**：检测BeanClass中是否含有@RocketListener标识的方法，若有则注册一个DefaultMQPushConsumer类型的BeanDefinition，提取注解上的信息放入其中，以备后续实例化Bean时使用。

import java.lang.reflect.Method;
    import java.util.HashSet;
    import java.util.Map;
    import java.util.Set;
    
    import com.bob.intergrate.rocket.ann.RocketListener;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.MutablePropertyValues;
    import org.springframework.beans.factory.BeanDefinitionStoreException;
    import org.springframework.beans.factory.config.BeanDefinition;
    import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
    import org.springframework.beans.factory.support.BeanDefinitionRegistry;
    import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
    import org.springframework.beans.factory.support.RootBeanDefinition;
    import org.springframework.context.annotation.Bean;
    import org.springframework.core.MethodIntrospector;
    import org.springframework.core.MethodIntrospector.MetadataLookup;
    import org.springframework.util.Assert;
    import org.springframework.util.ReflectionUtils;
    
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUMER_GROUP;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUME_BEAN_NAME;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUME_FROM_WHERE;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUME_METHOD;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.NAMESRV_ADDR;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.ROCKETMQ_CONSUMER_BEAN_NAME_SUFFIX;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.TAG;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.TOPIC;
    
    /** * {@link RocketListener}注解解析器 * * @author wb-jjb318191 * @create 2018-03-20 9:25 */
    public class RocketListenerAnnotationPostProcessor implements BeanDefinitionRegistryPostProcessor {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RocketListenerAnnotationPostProcessor.class);
    
        @Override
        public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry beanDefinitionRegistry) throws BeansException {
            String[] beanDefinitionNames = beanDefinitionRegistry.getBeanDefinitionNames();
            for (String name : beanDefinitionNames) {
                BeanDefinition beanDefinition = beanDefinitionRegistry.getBeanDefinition(name);
                Class<?> beanClass;
                try {
                    beanClass = resolveBeanClass(beanDefinition, beanDefinitionRegistry);
                } catch (ClassNotFoundException e) {
                    throw new BeanDefinitionStoreException(String.format("解析[%s]BeanDefinition出现异常", beanDefinition.toString()), e);
                }
                Map<Method, RocketListener> annotatedMethods = introspectorRocketAnnotatedMethod(beanClass);
                if (annotatedMethods.isEmpty()) {
                    continue;
                }
                for (Map.Entry<Method, RocketListener> entry : annotatedMethods.entrySet()) {
                    RocketListener listener = entry.getValue();
                    BeanDefinition rocketConsumer = new RootBeanDefinition(DefaultMQPushConsumer.class);
                    //让@RocketListener的定义Bean先实例化
                    rocketConsumer.setDependsOn(beanDefinition.getFactoryBeanName());
                    MutablePropertyValues mpv = rocketConsumer.getPropertyValues();
                    mpv.add(CONSUMER_GROUP, listener.consumerGroup());
                    mpv.add(TOPIC, listener.topic());
                    mpv.add(TAG, listener.tag());
                    mpv.add(NAMESRV_ADDR, listener.namesrvAddr());
                    mpv.add(CONSUME_BEAN_NAME, name);
                    mpv.add(CONSUME_METHOD, entry.getKey());
                    mpv.add(CONSUME_FROM_WHERE, listener.consumeFromWhere());
                    //定义消费者Bean的名称格式为：factoryMethodName + RocketConsumer
                    beanDefinitionRegistry.registerBeanDefinition(buildRocketConsumerBeanName(entry.getKey().getName()), rocketConsumer);
                    LOGGER.info("注册[{}]标识的[{}]方法为RocketMQ Push消费者", RocketListener.class.getSimpleName(), entry.getKey().toString());
                }
            }
        }
    
        /** * 内省{@link RocketListener}标识的方法 * * @param beanClass * @return */
        private Map<Method, RocketListener> introspectorRocketAnnotatedMethod(Class<?> beanClass) {
            return MethodIntrospector.selectMethods(beanClass,
                (MetadataLookup<RocketListener>)(method) -> method.getAnnotation(RocketListener.class));
        }
    
        /** * 解析Bean Class对象 * * @param beanDefinition * @param beanDefinitionRegistry * @return * @throws ClassNotFoundException */
        private Class<?> resolveBeanClass(BeanDefinition beanDefinition, BeanDefinitionRegistry beanDefinitionRegistry) throws ClassNotFoundException {
            String beanClassName = beanDefinition.getBeanClassName();
            Class<?> beanClass = null;
            if (beanClassName != null) {
                beanClass = resolveClass(beanClassName);
            } else if (beanDefinition.getFactoryMethodName() != null) {
                String factoryMethodName = beanDefinition.getFactoryMethodName();
                String factoryBeanClassName = beanDefinitionRegistry.getBeanDefinition(beanDefinition.getFactoryBeanName()).getBeanClassName();
                Class<?> factoryBeanClass = resolveClass(factoryBeanClassName);
                Set<Method> candidateMethods = new HashSet<>();
                ReflectionUtils.doWithMethods(factoryBeanClass,
                    candidateMethods::add,
                    (method) -> method.getName().equals(factoryMethodName) && method.isAnnotationPresent(Bean.class)
                );
                Assert.state(candidateMethods.size() == 1, String.format("[%s]类中标识@Bean的方法[%s]不止一个", factoryBeanClass.getName(), factoryMethodName));
                beanClass = candidateMethods.iterator().next().getReturnType();
            }
            Assert.notNull(beanClass, String.format("解析[%s]BeanDefinition出现异常,BeanClass解析失败", beanDefinition.toString()));
            return beanClass;
    
        }
    
        /** * @param className * @return */
        private Class<?> resolveClass(String className) throws ClassNotFoundException {
            Class<?> beanClass;
            try {
                beanClass = Class.forName(className);
            } catch (ClassNotFoundException e) {
                LOGGER.error("不存在[{}]相应的Class", className);
                throw e;
            }
            return beanClass;
        }
    
        /** * 构建{@link RocketListener}定义形式的Consumer的Bean的名称 * * @param factoryMethodName * @return */
        private String buildRocketConsumerBeanName(String factoryMethodName) {
            return factoryMethodName + ROCKETMQ_CONSUMER_BEAN_NAME_SUFFIX;
        }
    
        @Override
        public void postProcessBeanFactory(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
    
        }
    
    }

**RocketListenerAnnotationBeanPostProcessor**：提取之前放入BeanDefinition中的@RocketListener的注解信息，填充如DefaultMQPushConsumer的Bean中。

import java.beans.PropertyDescriptor;
    import java.lang.reflect.Method;
    
    import com.bob.intergrate.rocket.ann.RocketListener;
    import com.bob.intergrate.rocket.listener.RocketMessageListener;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.aop.support.AopUtils;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.PropertyValues;
    import org.springframework.beans.factory.BeanCreationException;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.config.ConfigurableBeanFactory;
    import org.springframework.beans.factory.config.InstantiationAwareBeanPostProcessorAdapter;
    import org.springframework.core.MethodIntrospector;
    import org.springframework.util.StringUtils;
    
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUMER_GROUP;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUME_BEAN_NAME;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUME_FROM_WHERE;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.CONSUME_METHOD;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.NAMESRV_ADDR;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.TAG;
    import static com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant.TOPIC;
    
    /** * @author wb-jjb318191 * @create 2018-03-20 14:01 */
    public class RocketListenerAnnotationBeanPostProcessor extends InstantiationAwareBeanPostProcessorAdapter { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RocketListenerAnnotationBeanPostProcessor.class);
    
        @Autowired
        private ConfigurableBeanFactory beanFactory;
    
        /** * 初始化消费者属性 * * @see RocketListener */
        @Override
        public PropertyValues postProcessPropertyValues(PropertyValues pvs, PropertyDescriptor[] pds, Object bean, String beanName) throws BeansException {
            if (bean instanceof DefaultMQPushConsumer) {
                DefaultMQPushConsumer consumer = (DefaultMQPushConsumer)bean;
                consumer.setConsumerGroup(getStringProperty(pvs, CONSUMER_GROUP));
                String namesrvAddr = getStringProperty(pvs, NAMESRV_ADDR);
                if (StringUtils.hasText(namesrvAddr)) {
                    consumer.setNamesrvAddr(namesrvAddr);
                }
                consumer.setConsumeFromWhere(getProperty(pvs, CONSUME_FROM_WHERE, ConsumeFromWhere.class));
                //订阅信息
                String topic = getStringProperty(pvs, TOPIC);
                String tag = getStringProperty(pvs, TAG);
                try {
                    consumer.subscribe(topic, tag);
                } catch (MQClientException e) {
                    throw new BeanCreationException(String.format("订阅基于Topic:[%s],Tag:[%s]的RocketMQ消费者创建失败", topic, tag));
                }
                Object consumeBean = beanFactory.getBean(getStringProperty(pvs, CONSUME_BEAN_NAME));
                Method consumeMethod = getProperty(pvs, CONSUME_METHOD, Method.class);
                if (AopUtils.isAopProxy(consumeBean)) {
                    consumeMethod = MethodIntrospector.selectInvocableMethod(consumeMethod, consumeBean.getClass());
                }
                consumer.registerMessageListener(new RocketMessageListener(consumeBean, consumeMethod));
                LOGGER.info("订阅基于ConsumeGroup:[{}],Topic:[{}],Tag:[{}]的RocketMQ消费者创建成功", consumer.getConsumerGroup(), topic, tag);
                return null;
            }
            return super.postProcessPropertyValues(pvs, pds, bean, beanName);
        }
    
        /** * 解析配置,如果配置以"$"开始,则从Properties文件中查找相应的值 * * @param pvs * @param propertyName * @return */
        private String getStringProperty(PropertyValues pvs, String propertyName) {
            String value = (String)pvs.getPropertyValue(propertyName).getValue();
            if (value.startsWith("$")) {
                value = beanFactory.resolveEmbeddedValue(value);
            }
            return value;
        }
    
        /** * @param pvs * @return */
        private <T> T getProperty(PropertyValues pvs, String name, Class<T> targetClass) {
            return (T)pvs.getPropertyValue(name).getValue();
        }
    }

**RocketMessageListener**：这个就是将@RocketListener封装成的消息消费器

import java.lang.reflect.Method;
    import java.lang.reflect.Modifier;
    import java.lang.reflect.Parameter;
    import java.util.List;
    
    import com.bob.intergrate.rocket.ann.RocketListener;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.core.ResolvableType;
    import org.springframework.util.Assert;
    import org.springframework.util.ReflectionUtils;
    
    /** * 封装{@link RocketListener}标识的方法成一个消息消费器 * * @author wb-jjb318191 * @create 2018-03-20 10:03 */
    public class RocketMessageListener implements MessageListenerConcurrently { 
    
        private Object consumeBean;
        private Method consumeMethod;
        private static final String ERROR_MSG_PREFIX = "[@RocketListener]标识的方法";
    
        public RocketMessageListener(Object consumeBean, Method consumeMethod) {
            this.consumeBean = consumeBean;
            this.consumeMethod = consumeMethod;
            checkConsumeMethod();
        }
    
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            Class<?>[] types = consumeMethod.getParameterTypes();
            Object[] args = new Object[types.length];
            if (types[0] == List.class) {
                args[0] = msgs;
            } else {
                args[0] = msgs.get(0);
            }
            if (types.length == 2) {
                args[1] = context;
            }
            boolean result = (boolean)ReflectionUtils.invokeMethod(consumeMethod, consumeBean, args);
            return result ? ConsumeConcurrentlyStatus.CONSUME_SUCCESS : ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    
        /** * 校验{@link RocketListener}标识的方法正确性 */
        private void checkConsumeMethod() {
            int paramLength = consumeMethod.getParameterCount();
            Assert.state(paramLength == 1 || paramLength == 2, ERROR_MSG_PREFIX + "参数长度只能在1和2之间");
            Parameter param0 = consumeMethod.getParameters()[0];
            boolean isMsg = param0.getType().isAssignableFrom(MessageExt.class);
            boolean isList = param0.getType().isAssignableFrom(List.class);
            Assert.state(isMsg || isList, ERROR_MSG_PREFIX + "第一个参数只能是MessageExt或List<MessageExt>类型");
            if (isList) {
                Class<?> generic = ResolvableType.forMethodParameter(consumeMethod, 0).resolveGeneric(0);
                Assert.isAssignable(MessageExt.class, generic, ERROR_MSG_PREFIX + "第一个参数的泛型只能是MessageExt类型");
            }
            if (paramLength == 2) {
                Assert.state(consumeMethod.getParameters()[1].getType() == ConsumeConcurrentlyContext.class,
                    ERROR_MSG_PREFIX + "第2个参数类型只能是[ConsumeConcurrentlyContext]");
            }
            Assert.state(Modifier.isPublic(consumeMethod.getModifiers()), ERROR_MSG_PREFIX + "修饰符必须为[Public]");
            Assert.state(consumeMethod.getReturnType() == boolean.class, ERROR_MSG_PREFIX + "返回值类型必须为[boolean]");
        }
    
    }

**RocketConsumerLifecycleProcessor**：这个Bean会在Spring容器启动完成后启动消费者，开始消费，同时在Spring关闭时终止消费客户端。

import java.util.Map;
    import java.util.Map.Entry;
    
    import javax.annotation.PreDestroy;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.SmartLifecycle;
    
    /** * RocketMQ Consumer生命周期处理器 * * @author wb-jjb318191 * @create 2018-03-20 9:41 */
    public class RocketConsumerLifecycleProcessor implements SmartLifecycle { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RocketConsumerLifecycleProcessor.class);
    
        @Autowired
        private Map<String, DefaultMQPushConsumer> rocketMQConsumers;
    
        private volatile boolean isRunning = false;
    
        @Override
        public void start() {
            for (Entry<String, DefaultMQPushConsumer> entry : rocketMQConsumers.entrySet()) {
                try {
                    entry.getValue().start();
                } catch (MQClientException e) {
                    LOGGER.error("启动[{}]消费者失败", entry.getKey(), e);
                }
            }
            isRunning = true;
        }
    
        @Override
        public boolean isAutoStartup() {
            return true;
        }
    
        @Override
        public boolean isRunning() {
            return isRunning;
        }
    
        @Override
        public int getPhase() {
            return Integer.MAX_VALUE;
        }
    
        @PreDestroy
        public void destroy() {
            isRunning = false;
            for (DefaultMQPushConsumer consumer : rocketMQConsumers.values()) {
                consumer.shutdown();
            }
        }
    
        @Override
        public void stop(Runnable runnable) {
    
        }
    
        @Override
        public void stop() {
    
        }
    
    }

**RocketBeanDefinitionConstant**：一个接口，存放整合解析时的一些常量定义。

/** * RocketMQ定义时的常量 * * @author wb-jjb318191 * @create 2018-03-20 14:38 */
    public interface RocketBeanDefinitionConstant { 
    
        /** * 消费者组 */
        String CONSUMER_GROUP = "consumerGroup";
    
        /** * 主题 */
        String TOPIC = "topic";
    
        /** * 小标题 */
        String TAG = "tag";
    
        /** * */
        String NAMESRV_ADDR = "namesrvAddr";
    
        /** * 消费Bean */
        String CONSUME_BEAN_NAME = "consumeBeanName";
    
        /** * 消费方法 */
        String CONSUME_METHOD = "consumeMethod";
    
        /** * 消费偏移模式 */
        String CONSUME_FROM_WHERE = "consumeFromWhere";
    
        /** * RocketMQ消费者Bean名称后缀 */
        String ROCKETMQ_CONSUMER_BEAN_NAME_SUFFIX = "RocketConsumer";
    
    }

以上就是整合的所有代码了，总共就8个类，不多，但是基本的功能实现了，仅整合了消费者，生产者没有太多整合的必要。

下面看使用的方法：

**rocket-config.properties**：RocketMQ环境配置信息，Consumer和Producer的一些配置信息可以放在其中，Spring启动时将此配置文件内的信息注册到System的Property中，这样客户端在实例化时就能够用这些信息作为默认值。同时@RocketListener的配置信息可以使用 “$\{\}” 这种形式来引用配置文件中的信息，我在解析时做了相应的处理。

rocketmq.namesrv.addr = 127.0.0.1:9876
    com.rocketmq.sendMessageWithVIPChannel = false
    rocketmq.client.name = project@001
    service.consumerGroup = rmq-group
    service.topic = test-topic

**RocketConsumerConfiguration**：消费者配置类，定义@RocketListener标识的方法

import java.util.List;
    
    import com.bob.intergrate.rocket.ann.RocketListener;
    import com.bob.intergrate.rocket.constant.RocketBeanDefinitionConstant;
    import com.google.gson.Gson;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.beans.factory.annotation.Autowired;
    
    /** * RocketMQ消费者配置 * * @author wb-jjb318191 * @create 2018-03-20 16:06 */
    public class RocketConsumerConfiguration { 
    
        private Gson gson = new Gson();
    
        /** * 通过{@link RocketListener}形式定义的Consumer,名称形式以为： * consumeMethodName + {@link RocketBeanDefinitionConstant#ROCKETMQ_CONSUMER_BEAN_NAME_SUFFIX} * BeanName形式可自定义 */
        @Autowired
        private DefaultMQPushConsumer serviceRocketConsumer;
    
        /** * 定义RocketMQ消费器 * * @param msg * @param context * @return true:消费成功; false:消费失败,发回给Broker,一段时间后重试 */
        @RocketListener(consumerGroup = "${service.consumerGroup}", topic = "${service.topic}")
        public boolean service(MessageExt msg, ConsumeConcurrentlyContext context) {
            System.out.println(gson.toJson(msg));
            return true;
        }
    
        //@RocketListener(consumerGroup = "demo", topic = "test-topic")
        public boolean demo(MessageExt msg) {
            System.out.println(gson.toJson(msg));
            return true;
        }
    
        //@RocketListener(consumerGroup = "project", topic = "test-topic")
        public boolean project(List<MessageExt> msgs) {
            for (MessageExt msg : msgs) {
                System.out.println(gson.toJson(msg));
            }
            return true;
        }
    
    }

**RocketContextConfig**：RocketMQ使用时的总配置类，引入Properties配置文件，解析其信息注册到System的property中，定义Consumer和Producer。标识@EnableRocket，引入Spring整合RocketMQ组件。

import java.util.Properties;
    
    import javax.annotation.PostConstruct;
    
    import com.bob.intergrate.rocket.ann.EnableRocket;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.log.ClientLogger;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.PropertySource;
    import org.springframework.core.env.ConfigurableEnvironment;
    
    /** * Rocket消息队列配置 * * @author wb-jjb318191 * @create 2018-02-11 15:09 */
    @EnableRocket
    @Configuration
    @PropertySource(name = "rocket-config", value = "classpath:rocket-config.properties")
    public class RocketContextConfig { 
    
        @Autowired
        private ConfigurableEnvironment environment;
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RocketContextConfig.class);
    
        static {
            //设置Rocket的日志
            ClientLogger.setLog(LOGGER);
        }
    
        /** * 将rocket-config.properties内的配置信息配置到System中 * 这样Consumer,Producer实例化时有些属性就有默认值 */
        @PostConstruct
        private void initRocketContext() {
            Properties properties = (Properties)environment.getPropertySources().get("rocket-config").getSource();
            for (String key : properties.stringPropertyNames()) {
                System.setProperty(key, properties.getProperty(key));
            }
        }
    
        @Bean
        public RocketConsumerConfiguration rocketMQConsumerConfiguration() {
            return new RocketConsumerConfiguration();
        }
    
        @Bean
        public DefaultMQProducer RocketMQProducer() throws MQClientException {
            DefaultMQProducer producer = new DefaultMQProducer("rmq_group");
            producer.setNamesrvAddr("127.0.0.1:9876");
            producer.setInstanceName("192.168.0.1@360");
            // 必须设为false否则连接broker10909端口
            producer.setVipChannelEnabled(false);
            producer.start();
            return producer;
        }
    
    }

最后将RocketContextConfig 作为RocketMQ模块的配置类，在Spring里可以通过@Import的形式引入框架中，这样能最大程度的解耦框架间的整合，引入与否由是否Import这个配置类来决定。

@Configuration
    @EnableAsync
    @EnableWebMvc
    @EnableDataValidate
    @EnableAspectJAutoProxy(proxyTargetClass = true)
    @ComponentScan(basePackages = {
       "com.bob.web.mvc"})
    @Import({
        AppUserContextConfig.class,
        MysqlContextConfig.class,
        TransactionContextConfig.class,
        RedisContextConfig.class,
        AopContextConfig.class,
        RocketContextConfig.class
    })
    public class WebContextConfig extends WebMvcConfigurerAdapter { 
        ......
    }

[SouthEast]: /images/20220529/3f2ea2e6a5b345c19f981b272d832284.png