Spring整合RocketMQ(Version 1.2)

落日映苍穹つ 2022-05-28 02:06 157阅读 0赞

之前发过一篇整合相关的博客，随着对RocketMQ的理解加深，感觉写的不是很好，所以这次做了一些改进：

1.  修复定义@RocketListener的Bean以非@Bean的形式定义时，设置DependsOn为null的异常
2.  调整@RocketListener配置选项，支持properties文件形式配置
3.  增加有序消费模式，在@RocketListener上orderly=true时，会注册有序模式的消费者

Version 1.1有个明显的BUG没有测试出来，坑了，有序消费和并发消费不能使用同一个消息监听器：

public class RocketMessageListener implements MessageListenerConcurrently, MessageListenerOrderly

应为消费者在启动时是通过 MessageListener instanceOf MessageListenerOrderly 来判断当前是有序消费还是并发消费，因此通过@RocketListener生成的消费者都是有序的。

因为version 1.1增加了有序消费模式，所以并发消费就没有测试了，今天在测试事务消息时，发现问题了，马上修复下。

类结构如下图：  
![这里写图片描述][SouthEast]

按顺序由上往下依次看代码：

1.**EnableRocket**：整合模块引入注解，在Spring配置类上标识此注解就引入了整合模块

@Documented
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ ElementType.TYPE, ElementType.ANNOTATION_TYPE })
    @Import(RocketBootstrapConfiguration.class)
    public @interface EnableRocket { 
    }

2.**RocketListener**：标识方法的注解，整合模块会将指定方法封装成一个MessageLitener，同时提取注解上的数据，生成一个DefaultMQPushConsumer，将MessageListener注册到此消费者中。

@Documented
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ElementType.METHOD})
    public @interface RocketListener { 
    
        /** * 消费者组 * * @return */
        String consumerGroup() default "";
    
        /** * @return */
        String topic() default "";
    
        /** * @return */
        String tag() default "";
    
        /** * @return */
        String namesrvAddr() default "";
    
        /** * 是否有序 * * @return */
        boolean orderly() default false;
    
        /** * 配置Properties文件名称 * * @return */
        String configProperties() default "";
    
    }

1.  **RocketBeanDefinitionConstant**：在整合模块解析过程中定义的常量

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";
    
        /** * RocketMQ消费者Bean名称后缀 */
        String ROCKETMQ_CONSUMER_BEAN_NAME_SUFFIX = "RocketConsumer";
    
        /** * 是否有序消费 */
        String ORDERLY = "orderly";
    
        /** * 配置Properties文件 */
        String CONFIG_PROPERTIES = "configProperties";
    
    }

4.**AbstractMessageListener**：@RocketListener注解标识方法封装后的信息抽象

import java.lang.reflect.Method;
    import java.lang.reflect.Modifier;
    import java.lang.reflect.Parameter;
    import java.util.List;
    
    import com.bob.common.utils.rocket.ann.RocketListener;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.core.ResolvableType;
    import org.springframework.util.Assert;
    import org.springframework.util.ClassUtils;
    
    /** * RocketMQ消息监听器 * * @author wb-jjb318191 * @create 2018-04-02 16:26 */
    public abstract class AbstractMessageListener{ 
    
        protected Object consumeBean;
        protected Method consumeMethod;
        private static final String ERROR_MSG_PREFIX = "[@RocketListener]标识的方法";
    
        public RocketMessageListener(Object consumeBean, Method consumeMethod) {
            this.consumeBean = consumeBean;
            this.consumeMethod = consumeMethod;
            this.checkConsumeMethod();
        }
    
        protected Object[] buildConsumeArguments(List<MessageExt> msgs, Object 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;
            }
            return args;
        }
    
        /** * 校验{@link RocketListener}标识的方法正确性 */
        protected void checkConsumeMethod() {
            RocketListener listener = consumeMethod.getAnnotation(RocketListener.class);
            Assert.state(listener != null, String.format("RocketMQ消费者方法必须标识[@%s]注解", ClassUtils.getShortName(RocketListener.class)));
            //校验方法参数
            int paramLength = consumeMethod.getParameterCount();
            Assert.state(paramLength == 1 || paramLength == 2, ERROR_MSG_PREFIX + "参数长度只能在1和2之间");
            Parameter param0 = consumeMethod.getParameters()[0];
            boolean isMsg = MessageExt.class.isAssignableFrom(param0.getType());
            boolean isList = List.class.isAssignableFrom(param0.getType());
            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) {
                if (listener.orderly()) {
                    Assert.state(consumeMethod.getParameters()[1].getType() == ConsumeOrderlyContext.class,
                        ERROR_MSG_PREFIX + "有序消费时第2个参数类型只能是[ConsumeOrderlyContext]");
                } else {
                    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]");
        }
    
    }

5.**ConcurrentlyMessageListener** ：并发消费器

import java.lang.reflect.Method;
    import java.util.List;
    
    import com.bob.common.utils.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.util.ReflectionUtils;
    
    /** * 封装{@link RocketListener}标识的方法成一个消息并发消费器 * * @author wb-jjb318191 * @create 2018-03-20 10:03 */
    public class ConcurrentlyMessageListener extends AbstractMessageListener implements MessageListenerConcurrently { 
    
        public ConcurrentlyMessageListener(Object consumeBean, Method consumeMethod) {
            super(consumeBean, consumeMethod);
        }
    
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            Object[] args = buildConsumeArguments(msgs, context);
            boolean result = (boolean)ReflectionUtils.invokeMethod(consumeMethod, consumeBean, args);
            return result ? ConsumeConcurrentlyStatus.CONSUME_SUCCESS : ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    
    }

6.**OrderlyMessageListener** ：有序消费器

import java.lang.reflect.Method;
    import java.util.List;
    
    import com.bob.common.utils.rocket.ann.RocketListener;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.util.ReflectionUtils;
    
    /** * 封装{@link RocketListener}方法,生成有序的消息监听器 * * @author wb-jjb318191 * @create 2018-04-02 16:24 */
    public class OrderlyMessageListener extends AbstractMessageListener implements MessageListenerOrderly { 
    
        public OrderlyMessageListener(Object consumeBean, Method consumeMethod) {
            super(consumeBean, consumeMethod);
        }
    
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            Object[] args = buildConsumeArguments(msgs, context);
            boolean result = (boolean)ReflectionUtils.invokeMethod(consumeMethod, consumeBean, args);
            return result ? ConsumeOrderlyStatus.SUCCESS : ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }

7.**RocketConsumerLifecycleManager**：RocketMQ消费者生命周期管理器，负责在Spring容器启动成功后start消费者，以及在Spring关闭时shutdown消费者

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;
    import org.springframework.util.CollectionUtils;
    
    /** * RocketMQ Consumer生命周期处理器 * * @author wb-jjb318191 * @create 2018-03-20 9:41 */
    public class RocketConsumerLifecycleManager implements SmartLifecycle { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RocketConsumerLifecycleManager.class);
    
        @Autowired(required = false)
        private Map<String, DefaultMQPushConsumer> rocketMQConsumers;
    
        private volatile boolean isRunning = false;
    
        @Override
        public void start() {
            if (CollectionUtils.isEmpty(rocketMQConsumers)) {
                return;
            }
            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;
            if (CollectionUtils.isEmpty(rocketMQConsumers)) {
                return;
            }
            for (DefaultMQPushConsumer consumer : rocketMQConsumers.values()) {
                consumer.shutdown();
            }
        }
    
        @Override
        public void stop(Runnable runnable) {
    
        }
    
        @Override
        public void stop() {
    
        }
    
    }

8.**RocketListenerAnnotationPostProcessor**：@RocketListener注解解析器，注册消费者BeanDefinition，将注解上的数据填充入BeanDefinition中。

import java.lang.reflect.Method;
    import java.util.HashSet;
    import java.util.Map;
    import java.util.Set;
    
    import com.bob.common.utils.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.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONFIG_PROPERTIES;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONSUMER_GROUP;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONSUME_BEAN_NAME;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONSUME_METHOD;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.NAMESRV_ADDR;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.ORDERLY;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.ROCKETMQ_CONSUMER_BEAN_NAME_SUFFIX;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.TAG;
    import static com.bob.common.utils.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 = introspectRocketAnnotatedMethod(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(name);
                    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(ORDERLY, listener.orderly());
                    mpv.add(CONFIG_PROPERTIES, listener.configProperties());
                    //定义消费者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> introspectRocketAnnotatedMethod(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 = Class.forName(beanClassName);
            } else if (beanDefinition.getFactoryMethodName() != null) {
                String factoryMethodName = beanDefinition.getFactoryMethodName();
                String factoryBeanClassName = beanDefinitionRegistry.getBeanDefinition(beanDefinition.getFactoryBeanName()).getBeanClassName();
                Class<?> factoryBeanClass = Class.forName(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;
    
        }
    
        /** * 构建{@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 {
    
        }
    
    }

9.**RocketListenerAnnotationBeanPostProcessor**：提起@RocketListener上的数据初始化DefaultMQPushConsumer。

import java.beans.PropertyDescriptor;
    import java.io.IOException;
    import java.lang.reflect.Method;
    import java.util.Arrays;
    import java.util.List;
    import java.util.Properties;
    
    import com.bob.common.utils.rocket.ann.RocketListener;
    import com.bob.common.utils.rocket.listener.ConcurrentlyMessageListener;
    import com.bob.common.utils.rocket.listener.OrderlyMessageListener;
    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.BeanWrapper;
    import org.springframework.beans.BeanWrapperImpl;
    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.core.io.ClassPathResource;
    import org.springframework.core.io.Resource;
    import org.springframework.core.io.support.PropertiesLoaderUtils;
    import org.springframework.util.Assert;
    import org.springframework.util.ClassUtils;
    import org.springframework.util.StringUtils;
    
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONFIG_PROPERTIES;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONSUMER_GROUP;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONSUME_BEAN_NAME;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.CONSUME_METHOD;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.NAMESRV_ADDR;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.ORDERLY;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.TAG;
    import static com.bob.common.utils.rocket.constant.RocketBeanDefinitionConstant.TOPIC;
    
    /** * RocketMQ Bean处理器 * * @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;
    
        /** * Properties文件里非直接注入的属性 */
        private List<String> excludeProperties = Arrays.asList(TOPIC, TAG);
    
        /** * 初始化消费者属性 * * @see RocketListener */
        @Override
        public PropertyValues postProcessPropertyValues(PropertyValues pvs, PropertyDescriptor[] pds, Object bean, String beanName) throws BeansException {
            if (bean instanceof DefaultMQPushConsumer) {
                DefaultMQPushConsumer consumer = (DefaultMQPushConsumer)bean;
                //若配置了了Properties文件,首先用配置文件的数据初始化Bean
                Properties properties = null;
                String configPath = getStringProperty(pvs, CONFIG_PROPERTIES);
                if (StringUtils.hasText(configPath)) {
                    properties = loadProperties(configPath);
                    initBeanByProperties(bean, properties, excludeProperties);
                }
                String consumeGroup = getStringProperty(pvs, CONSUMER_GROUP);
                if (StringUtils.hasText(consumeGroup)) {
                    consumer.setConsumerGroup(consumeGroup);
                }
                String namesrvAddr = getStringProperty(pvs, NAMESRV_ADDR);
                if (StringUtils.hasText(namesrvAddr)) {
                    consumer.setNamesrvAddr(namesrvAddr);
                }
                //订阅信息
                String topic = getStringProperty(pvs, TOPIC);
                if (!StringUtils.hasText(topic) && properties != null) {
                    topic = properties.getProperty(TOPIC);
                    Assert.hasText(topic, "注册RocketMQ Consume时Topic不能为空");
                }
                String tag = getStringProperty(pvs, TAG);
                if (!StringUtils.hasText(tag) && properties != null) {
                    tag = properties.getProperty(TAG);
                }
                try {
                    consumer.subscribe(topic, tag);
                } catch (MQClientException e) {
                    throw new BeanCreationException(String.format("订阅Topic:[%s],Tag:[%s]消息队列失败", topic, tag), e);
                }
                Object consumeBean = beanFactory.getBean(getStringProperty(pvs, CONSUME_BEAN_NAME));
                Method consumeMethod = getProperty(pvs, CONSUME_METHOD, Method.class);
                if (ClassUtils.isCglibProxy(consumeBean)) {
                    consumeMethod = MethodIntrospector.selectInvocableMethod(consumeMethod, consumeBean.getClass());
                }
                //是否有序
                boolean ordered = getProperty(pvs, ORDERLY, boolean.class);
                if (ordered) {
                    consumer.registerMessageListener(new OrderlyMessageListener(consumeBean, consumeMethod));
                } else {
                    consumer.registerMessageListener(new ConcurrentlyMessageListener(consumeBean, consumeMethod));
                }
                LOGGER.info("订阅基于ConsumeGroup:[{}],Topic:[{}],Tag:[{}]的RocketMQ消费者创建成功", consumer.getConsumerGroup(), topic, tag);
                return null;
            }
            return super.postProcessPropertyValues(pvs, pds, bean, beanName);
        }
    
        /** * 通过Properties文件初始化Bean对象属性 * * @param bean * @param properties * @param excludePorperties */
        private void initBeanByProperties(Object bean, Properties properties, List<String> excludePorperties) {
            BeanWrapper beanWrapper = new BeanWrapperImpl(bean);
            for (String key : properties.stringPropertyNames()) {
                if (!excludePorperties.contains(key)) {
                    beanWrapper.setPropertyValue(key, properties.get(key));
                }
            }
        }
    
        /** * @param filePath * @return */
        private Properties loadProperties(String filePath) {
            Resource resource = new ClassPathResource(filePath);
            Assert.isTrue(resource.exists(), String.format("配置文件:[%s]不存在", filePath));
            Properties properties;
            try {
                properties = PropertiesLoaderUtils.loadProperties(resource);
            } catch (IOException e) {
                throw new BeanCreationException(String.format("加载RocketMQ Consumer配置文件[%s]出现异常", filePath), e);
            }
            return properties;
        }
    
        /** * 解析配置,如果配置以"$"开始,则从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();
        }
    }

10.**RocketBootstrapConfiguration**：整合模块配置类

@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 RocketConsumerLifecycleManager rocketConsumerLifecycleManager() {
            return new RocketConsumerLifecycleManager();
        }
    
    }

使用方法 ：

1.在Spring配置类上标识@EnableRocket，引入整合模块：

@EnableRocket
    @Configuration
    @PropertySource(name = "rocket-processor", value = "classpath:rocket-config.properties")
    @Import({RocketConsumerConfiguration.class, RocketProducerConfiguration.class})
    public class RocketContextConfig { 
    
        @Autowired
        private ConfigurableEnvironment environment;
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RocketContextConfig.class);
    
        static {
            //设置Rocket的日志
            ClientLogger.setLog(LOGGER);
        }
    
    }

1.  在Spring任意Bean上用@RocketListener标识相应方法，方法要求返回值为boolean类型，public修饰符，第一个参数为MessageExt或者List< MessageExt>，第二个参数可有可没有，若有，当orderly = true时，第二个参数必须为ConsumeOrderlyContext，orderly = false时，为ConsumeConcurrentlyContext。参考MessageListenerOrderly和MessageListenerConcurrently接口。

@Configuration
    public class RocketConsumerConfiguration { 
    
        private Gson gson = new Gson();
    
        /** * 通过{@link RocketListener}形式定义的Consumer,名称形式以为： * consumeMethodName + {@link RocketBeanDefinitionConstant#ROCKETMQ_CONSUMER_BEAN_NAME_SUFFIX} * BeanName形式可自定义 */
        @Autowired
        private DefaultMQPushConsumer orderlyRocketConsumer;
    
    
        /** * 定义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;
        }
    
        /** * 创建有序消费者 * * @param msg * @param context * @return */
        @RocketListener(orderly = true, configProperties = "rocket-orderly-config.properties")
        public boolean orderly(MessageExt msg, ConsumeOrderlyContext context) {
            String threadName = Thread.currentThread().getName();
            System.out.println(String.format("threadName:[%s], QueueOffset:[%d]", threadName, msg.getQueueOffset()));
            return true;
        }
    }
    
    rocket-orderly-config.properties:
    
    namesrvAddr=127.0.0.1:9876
    instanceName=orderly
    vipChannelEnabled=false
    consumerGroup=orderly-consume
    topic=orderly-topic
    tag=*
    consumeFromWhere=CONSUME_FROM_LAST_OFFSET
    consumeThreadMin=4
    consumeThreadMax=4
    pullBatchSize=3

若消费者配置选项不多，可以使用@RocketListener上的已有的几个选项，值可以直接写在注解上，也可以使用“$” 的形式从Spring的公共环境里取(properties文件)；若配置选项较多，可使用自定义properties的形式，这个properties不会被引入Spring中，是此消费者独享的。

[SouthEast]: /images/20220528/f0f3f032518e4cffa0c9b57ab9ea2584.png