VIP视频自助网站建设,网站注销,网站开发课程论文,佛山网站建设专业品牌概述
RabbitMessageContainer注解 的主要作用就是 替换掉Configuration配置类中的各种Bean配置#xff1b;
采用注解的方式可以让我们 固化配置#xff0c;降低代码编写复杂度、减少配置错误情况的发生#xff0c;提升编码调试的效率、提高业务的可用性。
为什么说“降低…概述
RabbitMessageContainer注解 的主要作用就是 替换掉Configuration配置类中的各种Bean配置
采用注解的方式可以让我们 固化配置降低代码编写复杂度、减少配置错误情况的发生提升编码调试的效率、提高业务的可用性。
为什么说“降低代码编写的复杂度”呢因为用一行注解代替了原本好几十行的代码。为什么说“减少配置错误情况的发生提升编码调试的效率”呢因为开发者从其他Configuration配置文件复制粘贴的代码有时会忘记修改某些Bean名称而启动又不会报错最终会导致队列没有消费者需要浪费时间排查问题。 为什么说“提高业务的可用性”呢因为组件默认配置了死信队列机制当消费失败的时候将异常抛出即可重试避免因为没有配置死信队列而导致消息丢失。如果继承AbstractJdkSerializeListener/AbstractJsonSerializeListener可以在重试一定次数后将消息落库并且丢弃
接入方式
该组件使用Spring Boot的自动装配能力只需要引入pom依赖即可完成接入。
dependencygroupIdcom.ccbscf/groupIdartifactIdccbscf-biz-enhancer-rabbitmq-starter/artifactIdversion1.0.1-SNAPSHOT/version
/dependency
支持哪些能力
简单来说以前Bean注入方式常用的能力这个组件都支持以下是具体注解信息及属性配置 com.ccbscf.biz.enhancer.rabbitmq.annotation.RabbitMessageContainer注解
/*** 向spring中注入SimpleMessageListenerContainer容器* 暂时只对Container的acknowledgeMode、exposeListenerChannel、prefetchCount、concurrentConsumers、maxConcurrentConsumers提供了赋值的扩展如果需要其他的字段赋值需要升级组件*/
Target({ElementType.TYPE})
Retention(RetentionPolicy.RUNTIME)
Documented
public interface RabbitMessageContainer {/*** container的name,向spring容器注入bean* return*/String value();/*** 定义绑定关系队列、交换器、路由key的定义都在这里面* 这里为什么是定义数组呢因为同一个Container是可以绑定多个队列的因此这里是数组* return*/QueueBinding[] bindings();/*** return* see AbstractMessageListenerContainer#setAcknowledgeMode(org.springframework.amqp.core.AcknowledgeMode)*/AcknowledgeMode acknowledgeMode() default AcknowledgeMode.MANUAL;/*** return* see AbstractMessageListenerContainer#setExposeListenerChannel(boolean)*/boolean exposeListenerChannel() default true;/*** return* see SimpleMessageListenerContainer#setPrefetchCount(int)*/int prefetchCount() default 5;/*** return* see SimpleMessageListenerContainer#setConcurrentConsumers(int)*/int concurrentConsumers() default 1;/*** return* see SimpleMessageListenerContainer#setMaxConcurrentConsumers(int)*/int maxConcurrentConsumers() default 1;/*** 失败 抛出异常 捕捉到异常以后 是否进行重试 默认重试* return*/boolean needRetry() default true;/*** 自定义的Listener维度的重试次数上限* return*/int customerRetriesLimitForListener() default -1;/*** 重试时间间隔* return*/long retryTimeInterval() default -1;
}
上面是RabbitMessageContainer注解的源代码原本Bean中SimpleMessageListenerContainer常用的参数设置这里都进行了支持如果有新的个性化字段赋值可以对组件进行扩展给注解增加字段同时注入BeanDefinition的时候赋值即可。
除了实现Bean方式常用字段另外增加了以下几个功能字段
needRetry失败 抛出异常 捕捉到异常以后 是否进行重试 默认重试customerRetriesLimitForListener自定义的Listener维度的重试次数上限此优先级高于全局的次数上限配置retryTimeInterval重试时间间隔固定时间间隔不支持梯度这个配置是加在队列参数上的一旦配置生效就无法修改这个RabbitMQ的特性
为了理解起来更直观下面展示出原有的Bean注入方式的示例
public static SimpleMessageListenerContainer buildSimpleMessageListenerContainer(Queue queue, ConnectionFactory connectionFactory, Object messageListener) {SimpleMessageListenerContainer simpleMessageListenerContainer new SimpleMessageListenerContainer(connectionFactory);simpleMessageListenerContainer.setQueues(queue);simpleMessageListenerContainer.setMaxConcurrentConsumers(1);simpleMessageListenerContainer.setConcurrentConsumers(1);simpleMessageListenerContainer.setPrefetchCount(5);simpleMessageListenerContainer.setExposeListenerChannel(true);simpleMessageListenerContainer.setMessageListener(messageListener);simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);return simpleMessageListenerContainer;
} com.ccbscf.biz.enhancer.rabbitmq.annotation.QueueBinding注解
Target({})
Retention(RetentionPolicy.RUNTIME)
public interface QueueBinding {/*** 绑定关系的name主要用于向容器中注入bean的名称* return*/String value();/*** return the queue.*/Queue queue();/*** return the exchange.*/Exchange exchange();/*** return the routing key or pattern for the binding.*/String key() default ;
}
上面是QueueBinding注解的源代码原本Bean中Binding常用的参数设置这里都进行了支持如果有新的个性化字段赋值可以对组件进行扩展给注解增加字段同时注入BeanDefinition的时候赋值即可。
为了理解起来更直观下面展示出原有的Bean注入方式的示例 Beanpublic Binding sendSuperviseBinding(TopicExchange approveDocDatumTopicExchange) {return BindingBuilder.bind(sendSuperviseQueue()).to(approveDocDatumTopicExchange).with(DOC_DATUM_TOPIC_APPROVE_ROUTING_KEY);} com.ccbscf.biz.enhancer.rabbitmq.annotation.Queue注解
Target({})
Retention(RetentionPolicy.RUNTIME)
public interface Queue {/*** return the queue name or for a generated queue name (default).*/String value();/*** return true if the queue is to be declared as durable.*/boolean durable() default true;/*** return true if the queue is to be declared as exclusive.*/boolean exclusive() default false;/*** return true if the queue is to be declared as auto-delete.*/boolean autoDelete() default false;/*** 是否延迟队列* return*/boolean delayConsumer() default false;/*** delayConsumer为true的情况下该字段才会生效单位ms* 如果设置了delayConsumertrue延迟队消费开启但是未设置delayTime延迟消费时间默认值是10分钟* return*/long delayTime() default -1;
}
上面是Queue注解的源代码原本Bean中Queue常用的参数设置这里都进行了支持如果有新的个性化字段赋值可以对组件进行扩展给注解增加字段同时注入BeanDefinition的时候赋值即可。
除了实现Bean方式常用字段另外增加了以下几个功能字段
delayConsumer是否延迟队列默认为false如果需要开启延迟消费的功能需要配置为truedelayTimedelayConsumer为true的情况下该字段才会生效单位ms如果设置了delayConsumertrue延迟队消费开启但是未设置delayTime延迟消费时间默认值是10分钟
为了理解起来更直观下面展示出原有的Bean注入方式的示例
new Queue(queueName, true, false, false, params) com.ccbscf.biz.enhancer.rabbitmq.annotation.Exchange注解
Target({})
Retention(RetentionPolicy.RUNTIME)
public interface Exchange {/*** return the exchange name.*/String value();/*** The exchange type - only DIRECT, FANOUT TOPIC, and HEADERS exchanges are supported.* return the exchange type.*/String type() default ExchangeTypes.TOPIC;/*** return true if the exchange is to be declared as durable.*/boolean durable() default true;/*** return true if the exchange is to be declared as auto-delete.*/boolean autoDelete() default false;
}
上面是Exchange注解的源代码原本Bean中Exchange常用的参数设置这里都进行了支持如果有新的个性化字段赋值可以对组件进行扩展给注解增加字段同时注入BeanDefinition的时候赋值即可。
为了理解起来更直观下面展示出原有的Bean注入方式的示例 Beanpublic TopicExchange bizCcbDefaultTopicExchange() {return new TopicExchange(BIZ_CCB_DEFAULT_TOPIC_EXCHANGE, true, false);}
核心代码逻辑
其实实现思路非常简单原有方式通过开发者定义Bean配置向spring容器中添加BeanDefinition并生成单例Bean新的方式根据开发者配置的注解信息集中式的生成BeanDefinition并注册到spring容器即可。
至于绑定关系、队列、交换器向MQ消息中心注册的过程不受任何影响因为本来Bean就是在向容器注入bean而已
核心代码都在这一个RabbitMqEnhancerBeanDefinitionRegistry类这个类实现了BeanDefinitionRegistryPostProcessor接口当然BeanDefinitionRegistryPostProcessor也继承了BeanFactoryPostProcessor接口只不过我们只使用了BeanDefinitionRegistryPostProcessor具有的特性向容器中注入BeanDefinition信息至于spring生成单例bean的过程我们不去干预还是交给spring来自行完成。
从RabbitMessageContainer、Queue、Exchange、QueueBinding注解中获取信息创建相应的BeanDefinition并注册到容器中由spring容器管理充分利用spring现有机制自动创建bean实例尽可能减少硬编码干预spring的流程。
源代码如下
/*** ClassName RabbitMqEnhancerBeanDefinitionRegistry* Description* 处理RabbitMessageContainer、Queue、Exchange、QueueBinding注解以及创建相应的BeanDefinition注册到容器中* 由spring容器管理充分利用spring现有机制自动创建bean实例尽可能减少硬编码干预spring的流程。* 还有一种实现思路是* 自定义一个BeanPostProcessor的实现类同时实现BeanFactoryAware接口目的是获取到BeanFactory用ApplicationContextAware也行但是BeanFactoryAware更好些* 调用postProcessAfterInitialization方法拦截Listener并识别注解信息创建并注册BeanDefinition调用BeanFactory的getBean方法创建单例bean对象* 这种方式不仅个性化spring的BeanDefinition的注册而且还个性化了bean的创建过程因此不是最优的方式。* Author zhangyuxuan* Date 2023/9/13 15:29* Version 1.0*/
public class RabbitMqEnhancerBeanDefinitionRegistry implements BeanDefinitionRegistryPostProcessor, EnvironmentAware {private Environment environment;/*** 处理RabbitMessageContainer、Queue、Exchange、QueueBinding注解以及创建相应的BeanDefinition注册到容器中* 由spring容器管理充分利用spring现有机制自动创建bean实例尽可能减少硬编码干预spring的流程。** param registry* throws BeansException*/Overridepublic void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {for (String beanDefinitionName : registry.getBeanDefinitionNames()) {BeanFactory beanFactory (BeanFactory) registry;//获取bean对应的ClassClass? type beanFactory.getType(beanDefinitionName);//获取RabbitMessageContainer注解RabbitMessageContainer rabbitMessageContainer AnnotationUtils.findAnnotation(type, RabbitMessageContainer.class);if (rabbitMessageContainer null) {continue;}//获取QueueBinding注解QueueBinding[] bindings rabbitMessageContainer.bindings();if (bindings.length 0) {continue;}//存储queue信息都是实际消费消息 绑定Listener的队列ListString queueNameList new ArrayList();// 这里为什么是定义数组呢因为同一个Container是可以绑定多个队列的因此这里是数组for (QueueBinding binding : bindings) {Queue queue binding.queue();Exchange exchange binding.exchange();//是否开启延迟消费功能boolean needDelay queue.delayConsumer();//是否开启重试功能boolean needRetry rabbitMessageContainer.needRetry();//死信重试路由keyString retryRoutingKey obtainDoConsumeQueue(queue, needDelay) DL_ROUTING_KEY_SUFFIX;//延迟消费 实际消费的交换器String exchangeForDelay environment.getProperty(spring.application.name, ) DELAY_EXCHANGE_NAME_SUFFIX;//失败重试 死信交换器String exchangeForDl environment.getProperty(spring.application.name, ) DL_EXCHANGE_NAME_SUFFIX;//失败重试 重试交换器String exchangeForRetry environment.getProperty(spring.application.name, ) RETRY_EXCHANGE_NAME_SUFFIX;if (needDelay) {//延迟消费String delayRoutingKey queue.value() DELAY_CONSUME_ROUTE_SUFFIX;//用于延迟消费//用户定义的原队列BindingWrapper bindingWrapper BindingWrapper.generateBinding(binding.value(), binding.key()).buildQueue(queue.value(), obtainMapForDelayQueue(delayRoutingKey, exchangeForDelay, queue.delayTime()), queue.durable(), queue.exclusive(), queue.autoDelete()).buildExchange(exchange.value(), exchange.type(), exchange.durable(), exchange.autoDelete());//注册用户定义的原队列相关配置configRabbitMq(registry, bindingWrapper, true);//实际消费消息的队列BindingWrapper bindingWrapperConsume BindingWrapper.generateBinding(binding.value() DELAY_CONSUME_BINDING_SUFFIX, delayRoutingKey).buildQueue(obtainDoConsumeQueue(queue, true), obtainMapForConsumeQueue(needRetry, retryRoutingKey, exchangeForDl), queue.durable(), queue.exclusive(), queue.autoDelete()).buildExchange(exchangeForDelay, exchange.type(), exchange.durable(), exchange.autoDelete());//注册实际消费消息的队列相关配置延迟交换器已经在配置中注册configRabbitMq(registry, bindingWrapperConsume, false);//存储queue信息都是实际消费消息 绑定Listener的队列queueNameList.add(bindingWrapperConsume.getQueueWrapper().getQueueName());} else {//非延迟消费BindingWrapper bindingWrapper BindingWrapper.generateBinding(binding.value(), binding.key()).buildQueue(queue.value(), obtainMapForConsumeQueue(needRetry, retryRoutingKey, exchangeForDl), queue.durable(), queue.exclusive(), queue.autoDelete()).buildExchange(exchange.value(), exchange.type(), exchange.durable(), exchange.autoDelete());//用户定义的原队列configRabbitMq(registry, bindingWrapper, true);//存储queue信息都是实际消费消息 绑定Listener的队列queueNameList.add(bindingWrapper.getQueueWrapper().getQueueName());}if (needRetry) {//是否需要重试//死信队列BindingWrapper bindingWrapperDl BindingWrapper.generateBinding(binding.value() DL_BINDING_SUFFIX, retryRoutingKey).buildQueue(queue.value() DL_QUEUE_SUFFIX, obtainMapForDlQueue(retryRoutingKey, exchangeForRetry, rabbitMessageContainer.retryTimeInterval()), queue.durable(), queue.exclusive(), queue.autoDelete()).buildExchange(exchangeForDl, DIRECT, exchange.durable(), exchange.autoDelete());//注册死信队列相关配置死信交换器已经在配置中注册configRabbitMq(registry, bindingWrapperDl, false);//重试队列 用于重新消费BindingWrapper bindingWrapperRetry BindingWrapper.generateBinding(binding.value() RETRY_BINDING_SUFFIX, retryRoutingKey).buildQueue(obtainDoConsumeQueue(queue, needDelay), Collections.emptyMap(), queue.durable(), queue.exclusive(), queue.autoDelete()).buildExchange(exchangeForRetry, exchange.type(), exchange.durable(), exchange.autoDelete());// 向容器中注册binding的BeanDefinition队列复用用户定义的重试交换器已经在配置中创建registryBinding(registry, bindingWrapperRetry);}}// 向容器中注册container的BeanDefinitionregistryContainer(registry, beanDefinitionName, rabbitMessageContainer, queueNameList);}}/*** 因为延迟消费情况的存在因此需要获取实际消费队列的逻辑* param queue* param needDelay* return*/private String obtainDoConsumeQueue(Queue queue, boolean needDelay) {return needDelay ? queue.value() DELAY_CONSUME_QUEUE_SUFFIX : queue.value();}/*** 向容器中注册mq的配置包括queue、exchange、binding* param registry* param bindingWrapper*/private void configRabbitMq(BeanDefinitionRegistry registry, BindingWrapper bindingWrapper, boolean isNeedCreateExchange) {// 向容器中注册queue的BeanDefinitionregistryQueue(registry, bindingWrapper);// 向容器中注册exchange的BeanDefinitionif (isNeedCreateExchange) {registryExchangeIfNecessary(registry, bindingWrapper);}// 向容器中注册binding的BeanDefinitionregistryBinding(registry, bindingWrapper);}/*** 向容器中注册container的BeanDefinition* param registry* param beanDefinitionName* param rabbitMessageContainer* param queueNameList*/private void registryContainer(BeanDefinitionRegistry registry, String beanDefinitionName, RabbitMessageContainer rabbitMessageContainer, ListString queueNameList) {ManagedArray managedArray new ManagedArray(org.springframework.amqp.core.Queue, queueNameList.size());for (String queueName : queueNameList) {managedArray.add(new RuntimeBeanReference(queueName));}AbstractBeanDefinition containerBeanDefinition BeanDefinitionBuilder.genericBeanDefinition(SimpleMessageListenerContainer.class).addConstructorArgReference(shadowConnectionFactory).addPropertyValue(queues, managedArray).addPropertyReference(messageListener, beanDefinitionName).addPropertyValue(acknowledgeMode, rabbitMessageContainer.acknowledgeMode()).addPropertyValue(maxConcurrentConsumers, rabbitMessageContainer.maxConcurrentConsumers()).addPropertyValue(concurrentConsumers, rabbitMessageContainer.concurrentConsumers()).addPropertyValue(prefetchCount, rabbitMessageContainer.prefetchCount()).addPropertyValue(exposeListenerChannel, rabbitMessageContainer.exposeListenerChannel()).getBeanDefinition();registry.registerBeanDefinition(rabbitMessageContainer.value(), containerBeanDefinition);}/*** 向容器中注册queue的BeanDefinition* param registry* param bindingWrapper*/private void registryQueue(BeanDefinitionRegistry registry, BindingWrapper bindingWrapper) {BindingWrapper.QueueWrapper queueWrapper bindingWrapper.getQueueWrapper();AbstractBeanDefinition queueBeanDefinition BeanDefinitionBuilder.genericBeanDefinition(org.springframework.amqp.core.Queue.class).addConstructorArgValue(queueWrapper.getQueueName()).addConstructorArgValue(queueWrapper.isDurable()).addConstructorArgValue(queueWrapper.isExclusive()).addConstructorArgValue(queueWrapper.isAutoDelete()).addConstructorArgValue(queueWrapper.getParams()).getBeanDefinition();registry.registerBeanDefinition(queueWrapper.getQueueName(), queueBeanDefinition);}/*** 如果有必要向容器注入交换器* param registry* param bindingWrapper*/private void registryExchangeIfNecessary(BeanDefinitionRegistry registry, BindingWrapper bindingWrapper) {// 如果容器中已经被ConfigurationClassPostProcessor添加了同名的Exchange的BeanDefinition那就不在添加了// 一是兼容项目原有代码已经通过Bean方式注入了BeanDefinition// 二是Exchange本来原则上就是应该尽可能服用的所以多个Listener一定会存在使用相同的Exchange的情况if (!registry.containsBeanDefinition(bindingWrapper.getExchangeWrapper().getExchangeName())) {registryExchange(registry, bindingWrapper);}}/*** 向容器中注册exchange的BeanDefinition* param registry* param bindingWrapper*/private void registryExchange(BeanDefinitionRegistry registry, BindingWrapper bindingWrapper) {BindingWrapper.ExchangeWrapper exchangeWrapper bindingWrapper.getExchangeWrapper();AbstractBeanDefinition exchangeBeanDefinition BeanDefinitionBuilder.genericBeanDefinition(this.obtainExchangeType(exchangeWrapper.getType())).addConstructorArgValue(exchangeWrapper.getExchangeName()).addConstructorArgValue(exchangeWrapper.isDurable()).addConstructorArgValue(exchangeWrapper.isAutoDelete()).getBeanDefinition();registry.registerBeanDefinition(exchangeWrapper.getExchangeName(), exchangeBeanDefinition);}/*** 向容器中注册binding的BeanDefinition* param registry* param bindingWrapper*/private void registryBinding(BeanDefinitionRegistry registry, BindingWrapper bindingWrapper) {AbstractBeanDefinition bindingBeanDefinition BeanDefinitionBuilder.genericBeanDefinition(org.springframework.amqp.core.Binding.class).addConstructorArgValue(bindingWrapper.getQueueWrapper().getQueueName()).addConstructorArgValue(Binding.DestinationType.QUEUE).addConstructorArgValue(bindingWrapper.getExchangeWrapper().getExchangeName()).addConstructorArgValue(bindingWrapper.getKey()).addConstructorArgValue(Collections.String, ObjectemptyMap()).getBeanDefinition();registry.registerBeanDefinition(bindingWrapper.getBindingName(), bindingBeanDefinition);}/*** 延迟消费 存储消息的制造延迟效果 的队列 上面的param* return*/private MapString, Object obtainMapForDelayQueue(String delayRoutingKey, String exchangeForConsume, long delayTime) {MapString, Object paramsForDelay new HashMap();paramsForDelay.put(X_MESSAGE_TTL_DEFAULT, delayTime -1 ? TTL_DEFAULT_VALUE : delayTime);//默认10分钟paramsForDelay.put(X_DEAD_LETTER_EXCHANGE, exchangeForConsume);//延迟交换器paramsForDelay.put(X_DEAD_LETTER_ROUTING_KEY, delayRoutingKey);//延迟消费路由keyreturn paramsForDelay;}/*** 和Listener绑定实际消费消息 的队列 上面的param* return*/private MapString, Object obtainMapForConsumeQueue(boolean needRetry, String dlRoutingKey, String exchangeForDl) {if (!needRetry) {return Collections.emptyMap();}MapString, Object paramsForDl new HashMap();paramsForDl.put(X_DEAD_LETTER_EXCHANGE, exchangeForDl);//死信交换器paramsForDl.put(X_DEAD_LETTER_ROUTING_KEY, dlRoutingKey);//死信消费路由keyreturn paramsForDl;}/*** 重试场景下 死信队列 上面的param* return*/private MapString, Object obtainMapForDlQueue(String bindingWrapperForRetry, String exchangeForRetry, long delayTime) {MapString, Object paramsForOriginal new HashMap();paramsForOriginal.put(X_DEAD_LETTER_EXCHANGE, exchangeForRetry);//重试交换器paramsForOriginal.put(X_DEAD_LETTER_ROUTING_KEY, bindingWrapperForRetry);//重试消费路由keyparamsForOriginal.put(X_MESSAGE_TTL_DEFAULT, delayTime -1 ? TTL_DEFAULT_VALUE : delayTime);//默认10分钟return paramsForOriginal;}/*** 根据注解中的属性值返回对应的交换机类型* param exchangeTypes* return*/private Class? obtainExchangeType(String exchangeTypes) {switch (exchangeTypes) {case DIRECT:return DirectExchange.class;case FANOUT:return FanoutExchange.class;case HEADERS:return HeadersExchange.class;case TOPIC:default:return TopicExchange.class;}}Overridepublic void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {//do nothing}Overridepublic void setEnvironment(Environment environment) {this.environment environment;}
}
MQ组件配置关系图
