Spring Boot集成RabbitMQ源码阅读笔记

「爱情、让人受尽委屈。」 2022-01-15 23:01 321阅读 0赞

# Spring Boot集成RabbitMQ源码阅读笔记 #

1. 概述  
RabbitMQ是实现了高级消息队列协议AMQP的开源消息中间件，服务器端由Erlang语言编写，支持多种客户端。  
Spring Boot集成RabbitMQ只需要添加依赖：

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

下面将从源码中分析Spring Boot如何集成RabbitMQ。  
`spring-boot-starter-parent`版本2.0.6.RELEASE。  
2. 一般入口  
首先，启动类上有一个`@SpringBootApplication`注解，它包含了`@SpringBootConfiguration`, `@EnableAutoConfiguration`, `@ComponentScan`三个注解。  
`@EnableAutoConfiguration`会去加载每个依赖包下META-INF目录中的spring.factories，并将`org.springframework.boot.autoconfigure.EnableAutoConfiguration`的value中每个配置类进行实例化。

name=spring-boot-autoconfigure/META-INF/spring.factories
    org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
    ...
    org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration,\

可以看到，这个文件中有很多XxxAutoConfiguration，例如RedisAutoConfiguration、JdbcTemplateAutoConfiguration、QuartzAutoConfiguration，他们都是通过该入口进行实例化的。  
另外，还有一种方式即通过@EnableXxx。  
(NOTE)如果不想自动配置某些组件，可以使用`@EnableAutoConfiguration(exclude = {XxxAutoConfiguration.class})`。

3. RabbitAutoConfiguration  
=3.1 条件  
想要通过上面的方式被实例化，其实还有一个条件，就是满足类上的条件注解。

name=RabbitAutoConfiguration.class,lines=10
    @Configuration
    @ConditionalOnClass({ RabbitTemplate.class, Channel.class })
    @EnableConfigurationProperties(RabbitProperties.class)    // 将RabbitProperties注入当前类
    @Import(RabbitAnnotationDrivenConfiguration.class)    // 将RabbitAnnotationDrivenConfiguration导入IOC容器中。
    public class RabbitAutoConfiguration {
    ...
    }

`@ConditionalOnClass({ RabbitTemplate.class, Channel.class })`表示当前classpath下若存在类RabbitTemplate和Channel，则实例化RabbitAutoconfiguration。  
RabbitTemplate类来自spring-rabbit包，Channel类来自amqp-client包。  
=3.2 类定义=  
此类定义了几个bean，包括连接池、RabbitTemplate(AmqpTemplate)、AmqpAdmin和RabbitMessagingTemplate。这几个bean足以用来操作rabbitmq。  
分段介绍该类的定义：  
(1) RabbitConnectionFactoryCreator

lines=10
    	@Configuration
    	@ConditionalOnMissingBean(ConnectionFactory.class)
    	protected static class RabbitConnectionFactoryCreator {
    
    		// Only available in rabbitmq-java-client 5.4.0 +
    		private static final boolean CAN_ENABLE_HOSTNAME_VERIFICATION = ReflectionUtils
    				.findMethod(com.rabbitmq.client.ConnectionFactory.class,
    						"enableHostnameVerification") != null;
    
    		@Bean
    		public CachingConnectionFactory rabbitConnectionFactory(
    				RabbitProperties properties,
    				ObjectProvider<ConnectionNameStrategy> connectionNameStrategy)
    				throws Exception {
    			PropertyMapper map = PropertyMapper.get();
    			CachingConnectionFactory factory = new CachingConnectionFactory(
    					getRabbitConnectionFactoryBean(properties).getObject());
    			map.from(properties::determineAddresses).to(factory::setAddresses);
    			map.from(properties::isPublisherConfirms).to(factory::setPublisherConfirms);
    			map.from(properties::isPublisherReturns).to(factory::setPublisherReturns);
    			RabbitProperties.Cache.Channel channel = properties.getCache().getChannel();
    			map.from(channel::getSize).whenNonNull().to(factory::setChannelCacheSize);
    			map.from(channel::getCheckoutTimeout).whenNonNull().as(Duration::toMillis)
    					.to(factory::setChannelCheckoutTimeout);
    			RabbitProperties.Cache.Connection connection = properties.getCache()
    					.getConnection();
    			map.from(connection::getMode).whenNonNull().to(factory::setCacheMode);
    			map.from(connection::getSize).whenNonNull()
    					.to(factory::setConnectionCacheSize);
    			map.from(connectionNameStrategy::getIfUnique).whenNonNull()
    					.to(factory::setConnectionNameStrategy);
    			return factory;
    		}

定义一个ConnectionFactory类型的bean，其实现类为CachingConnectionFactory。仅当容器中没有ConnectionFactory类型的bean时才会被实例化。  
(2) RabbitTemplateConfiguration

lines=10
    	@Configuration
    	@Import(RabbitConnectionFactoryCreator.class)
    	protected static class RabbitTemplateConfiguration {
    
    		private final ObjectProvider<MessageConverter> messageConverter;
    
    		private final RabbitProperties properties;
    
    		public RabbitTemplateConfiguration(
    				ObjectProvider<MessageConverter> messageConverter,
    				RabbitProperties properties) {
    			this.messageConverter = messageConverter;
    			this.properties = properties;
    		}
    
    		@Bean
    		@ConditionalOnSingleCandidate(ConnectionFactory.class)
    		@ConditionalOnMissingBean
    		public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    			PropertyMapper map = PropertyMapper.get();
    			RabbitTemplate template = new RabbitTemplate(connectionFactory);
    			MessageConverter messageConverter = this.messageConverter.getIfUnique();
    			if (messageConverter != null) {
    				template.setMessageConverter(messageConverter);
    			}
    			template.setMandatory(determineMandatoryFlag());
    			RabbitProperties.Template properties = this.properties.getTemplate();
    			if (properties.getRetry().isEnabled()) {
    				template.setRetryTemplate(createRetryTemplate(properties.getRetry()));
    			}
    			map.from(properties::getReceiveTimeout).whenNonNull().as(Duration::toMillis)
    					.to(template::setReceiveTimeout);
    			map.from(properties::getReplyTimeout).whenNonNull().as(Duration::toMillis)
    					.to(template::setReplyTimeout);
    			map.from(properties::getExchange).to(template::setExchange);
    			map.from(properties::getRoutingKey).to(template::setRoutingKey);
    			return template;
    		}
    
    		private boolean determineMandatoryFlag() {
    			Boolean mandatory = this.properties.getTemplate().getMandatory();
    			return (mandatory != null) ? mandatory : this.properties.isPublisherReturns();
    		}
    
    		private RetryTemplate createRetryTemplate(RabbitProperties.Retry properties) {
    			PropertyMapper map = PropertyMapper.get();
    			RetryTemplate template = new RetryTemplate();
    			SimpleRetryPolicy policy = new SimpleRetryPolicy();
    			map.from(properties::getMaxAttempts).to(policy::setMaxAttempts);
    			template.setRetryPolicy(policy);
    			ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    			map.from(properties::getInitialInterval).whenNonNull().as(Duration::toMillis)
    					.to(backOffPolicy::setInitialInterval);
    			map.from(properties::getMultiplier).to(backOffPolicy::setMultiplier);
    			map.from(properties::getMaxInterval).whenNonNull().as(Duration::toMillis)
    					.to(backOffPolicy::setMaxInterval);
    			template.setBackOffPolicy(backOffPolicy);
    			return template;
    		}
    
    		@Bean
    		@ConditionalOnSingleCandidate(ConnectionFactory.class)
    		@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "dynamic", matchIfMissing = true)
    		@ConditionalOnMissingBean
    		public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
    			return new RabbitAdmin(connectionFactory);
    		}
    
    	}

定义了2个bean，RabbitTemplate和AmqpAdmin：

*  RabbitTemplate：当只有一个ConnectionFactory且classpath中没有同类型的bean，才会初始化该类。该类用来简化RabbitMQ发送和接收消息操作。
 *  AmqpAdmin：当只有一个ConnectionFactory且classpath中没有同类型的bean且spring.rabbitmq.dynamic=true(默认true)时，才会初始化该类。该类封装了对RabbitMQ的管理操作，例如声明Queue、Exchange、Binding。  
    (3) MessagingTemplateConfiguration

lines=10
    	@Configuration
    	@ConditionalOnClass(RabbitMessagingTemplate.class)
    	@ConditionalOnMissingBean(RabbitMessagingTemplate.class)
    	@Import(RabbitTemplateConfiguration.class)
    	protected static class MessagingTemplateConfiguration {
    
    		@Bean
    		@ConditionalOnSingleCandidate(RabbitTemplate.class)
    		public RabbitMessagingTemplate rabbitMessagingTemplate(
    				RabbitTemplate rabbitTemplate) {
    			return new RabbitMessagingTemplate(rabbitTemplate);
    		}
    
    	}

定义了bean RabbitMessagingTemplate，条件是：当classpath中存在RabbitMessagingTemplate类且不存在同类型bean且只有一个RabbitTemplate bean时，初始化此类。

4. RabbitAnnotationDrivenConfiguration  
此类是RabbitMQ的注解驱动配置，用来支持以注解的方式使用rabbitmq。  
该配置类的条件是：`@ConditionalOnClass(EnableRabbit.class)`  
(1) SimpleRabbitListenerContainerFactory

lines=10
    	@Bean
    	@ConditionalOnMissingBean
    	public SimpleRabbitListenerContainerFactoryConfigurer simpleRabbitListenerContainerFactoryConfigurer() {
    		SimpleRabbitListenerContainerFactoryConfigurer configurer = new SimpleRabbitListenerContainerFactoryConfigurer();
    		configurer.setMessageConverter(this.messageConverter.getIfUnique());
    		configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique());
    		configurer.setRabbitProperties(this.properties);
    		return configurer;
    	}
    
    	@Bean(name = "rabbitListenerContainerFactory")
    	@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
    	@ConditionalOnProperty(prefix = "spring.rabbitmq.listener", name = "type", havingValue = "simple", matchIfMissing = true)
    	public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(
    			SimpleRabbitListenerContainerFactoryConfigurer configurer,
    			ConnectionFactory connectionFactory) {
    		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    		configurer.configure(factory, connectionFactory);
    		return factory;
    	}

定义了一个简单的RabbitListenerContainerFactory bean。  
(2) DirectRabbitListenerContainerFactory

lines=10
    	@Bean
    	@ConditionalOnMissingBean
    	public DirectRabbitListenerContainerFactoryConfigurer directRabbitListenerContainerFactoryConfigurer() {
    		DirectRabbitListenerContainerFactoryConfigurer configurer = new DirectRabbitListenerContainerFactoryConfigurer();
    		configurer.setMessageConverter(this.messageConverter.getIfUnique());
    		configurer.setMessageRecoverer(this.messageRecoverer.getIfUnique());
    		configurer.setRabbitProperties(this.properties);
    		return configurer;
    	}
    
    	@Bean(name = "rabbitListenerContainerFactory")
    	@ConditionalOnMissingBean(name = "rabbitListenerContainerFactory")
    	@ConditionalOnProperty(prefix = "spring.rabbitmq.listener", name = "type", havingValue = "direct")
    	public DirectRabbitListenerContainerFactory directRabbitListenerContainerFactory(
    			DirectRabbitListenerContainerFactoryConfigurer configurer,
    			ConnectionFactory connectionFactory) {
    		DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
    		configurer.configure(factory, connectionFactory);
    		return factory;
    	}

该RabbitListenerContainerFactory不常用。  
(3)

@Configuration
    	@EnableRabbit
    	@ConditionalOnMissingBean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
    	protected static class EnableRabbitConfiguration {
    
    	}

空类，用来启用`@EnableRabbit`，而`@EnableRabbit`唯一作用是引入`RabbitBootstrapConfiguration`类。

5. RabbitBootstrapConfiguration  
该类定义了2个bean，`RabbitListenerAnnotationBeanPostProcessor`和`RabbitListenerEndpointRegistry`。前者用来在扫描使用了Rabbit的注解的bean，并将扫描到的结果封装为`RabbitListenerEndpoint`注册到后者上。

@Configuration
    public class RabbitBootstrapConfiguration {
    
    	@Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
    	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    	public RabbitListenerAnnotationBeanPostProcessor rabbitListenerAnnotationProcessor() {
    		return new RabbitListenerAnnotationBeanPostProcessor();
    	}
    
    	@Bean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
    	public RabbitListenerEndpointRegistry defaultRabbitListenerEndpointRegistry() {
    		return new RabbitListenerEndpointRegistry();
    	}
    
    }

6. RabbitListenerAnnotationBeanPostProcessor  
`RabbitListenerAnnotationBeanPostProcessor`在postProcessAfterInitialization阶段扫描Rabbit的注解，每个扫描结果封装为MethodRabbitListenerEndpoint类型。  
`RabbitListenerAnnotationBeanPostProcessor`有一个成员变量`private final RabbitListenerEndpointRegistrar registrar = new RabbitListenerEndpointRegistrar();`，可以称之为注册员，用注册员来代劳向`RabbitListenerEndpointRegistry`注册的动作，注册员可以选择先将注册信息缓存在自己手里。  
`RabbitListenerAnnotationBeanPostProcessor`在afterSingletonsInstantiated阶段，将注册员中的containerFactoryBeanName设置为rabbitListenerContainerFactory；并将注册员的注册信息（endpoint和listenerContainerFactory）注册到注册处。  
endpoint UML图：  
\{F16272,size=full,layout=center\}

7. 启动  
最终，注册处持有了endpoint和对应的MessageListenerContainer。在start阶段，注册处开启了所有MessageListenerContainer，即调用了MessageListenerContainer的start方法。  
注册员持有的RabbitListenerContainerFactory的bean name是“rabbitListenerContainerFactory”，即上面定义的SimpleRabbitListenerContainerFactory，而：

name=SimpleRabbitListenerContainerFactory
    	protected SimpleMessageListenerContainer createContainerInstance() {
    		return new SimpleMessageListenerContainer();
    	}

所以上面的MessageListenerContainer即SimpleMessageListenerContainer。  
SimpleMessageListenerContainer继承自AbstractMessageListenerContainer。

name=AbstractMessageListenerContainer.start, lines=10
    	public void start() {
    		if (isRunning()) {
    			return;
    		}
    		if (!this.initialized) {
    			synchronized (this.lifecycleMonitor) {
    				if (!this.initialized) {
    					afterPropertiesSet();
    					this.initialized = true;
    				}
    			}
    		}
    		try {
    			if (logger.isDebugEnabled()) {
    				logger.debug("Starting Rabbit listener container.");
    			}
    			configureAdminIfNeeded();
    			checkMismatchedQueues();
    			doStart();
    		}
    		catch (Exception ex) {
    			throw convertRabbitAccessException(ex);
    		}
    	}

SimpleMessageListenerContainer重写了doStart方法，因此此处会调用SimpleMessageListenerContainer的doStart，开启了对containner对应的每个队列的监听。  
doStart中创建了消费者set：

Set<AsyncMessageProcessingConsumer> processors = new HashSet<AsyncMessageProcessingConsumer>();

AsyncMessageProcessingConsumer是一个Runnable类，其run 方法中有如下调用链：

receiveAndExecute -> doReceiveAndExecute -> executeListener -> invokeListener -> proxy.invokeListener

下面解释proxy.invokeListener方法。  
8. 处理  
在AbstractMessageListenerContainer中，afterPropertiesSet阶段有如下调用链：afterPropertiesSet() -> initialize() -> initializeProxy(this.delegate)。

name=delegate
    // 成员变量
    private final ContainerDelegate delegate = this::actualInvokeListener;
    private ContainerDelegate proxy = this.delegate;
    ...
    // 内部接口
    	@FunctionalInterface
    	private interface ContainerDelegate {
    
    		void invokeListener(Channel channel, Message message) throws Exception;
    
    	}

在initializeProxy中，使用为delegate创建了AOP代理this.proxy。

protected void initializeProxy(Object delegate) {
    		if (this.getAdviceChain().length == 0) {
    			return;
    		}
    		ProxyFactory factory = new ProxyFactory();
    		for (Advice advice : getAdviceChain()) {
    			factory.addAdvisor(new DefaultPointcutAdvisor(advice));
    		}
    		factory.addInterface(ContainerDelegate.class);
    		factory.setTarget(delegate);
    		this.proxy = (ContainerDelegate) factory.getProxy(ContainerDelegate.class.getClassLoader());
    	}

因此在调用proxy.invokeListener时，即调用actualInvokeListener。而actualInvokeListener里的获取的MessageListener里，已经封装了endpoint中的bean和method，所以最终回调了自定义的消费者方法。  
container UML图：  
\{F16274,size=full,layout=center\}

9. 总结  
自动配置时，各个类的引入顺序如下：  
\{F16279,size=full,layout=center\}

写在公司内部phabricator上的，图片忽略吧，不重要。