RabbitMQ 默认CachingConnectionFactory配置源码分析

自动配置类

org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration

自动配置激活条件：

1. classpath下包含RabbitMQ 和 Spring AMQP 客户端。
2. 没有自定义ConnectionFactory。

配置源码实现

1. 新建 CachingConnectionFactory；
2. 配置 CachingConnectionFactory；
   1). org.springframework.boot.autoconfigure.amqp.AbstractConnectionFactoryConfigurer#configure(CachingConnectionFactory connectionFactory)

/**
* Configures the given {@code connectionFactory} with sensible defaults.
* @param connectionFactory connection factory to configure
*/
public final void configure(T connectionFactory) {
   Assert.notNull(connectionFactory, "ConnectionFactory must not be null");
   PropertyMapper map = PropertyMapper.get();
   // 配置连接地址信息，可以是多个。
   map.from(this.rabbitProperties::determineAddresses).to(connectionFactory::setAddresses);
   
   // 配置地址随机模式（NONE,RANDOM,INORDER）
   map.from(this.rabbitProperties::getAddressShuffleMode).whenNonNull()
         .to(connectionFactory::setAddressShuffleMode);


   // 配置连接命名策略
   map.from(this.connectionNameStrategy).whenNonNull().to(connectionFactory::setConnectionNameStrategy);
   configure(connectionFactory, this.rabbitProperties);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2). org.springframework.boot.autoconfigure.amqp.CachingConnectionFactoryConfigurer#configure(CachingConnectionFactory connectionFactory, RabbitProperties rabbitProperties)

@Override
public void configure(CachingConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
   PropertyMapper map = PropertyMapper.get();


   // 配置是否开启publisher returns，默认为false
   map.from(rabbitProperties::isPublisherReturns).to(connectionFactory::setPublisherReturns);
   
   // 配置 publisher confirms type，默认是NONE, 可选值还有 SIMPLE 和 CORRELATED
   map.from(rabbitProperties::getPublisherConfirmType).whenNonNull()
         .to(connectionFactory::setPublisherConfirmType);


   RabbitProperties.Cache.Channel channel = rabbitProperties.getCache().getChannel();


   // 配置缓存channels的个数，默认值是25
   map.from(channel::getSize).whenNonNull().to(connectionFactory::setChannelCacheSize);


   // 配置 channel 检出超时时间，单位是毫秒。默认是0，每次新建channel，不受channelCacheSize影响。当大于0时，
      受channelCacheSize影响，超时没检出channel会抛报错。
   map.from(channel::getCheckoutTimeout).whenNonNull().as(Duration::toMillis)
         .to(connectionFactory::setChannelCheckoutTimeout);


   RabbitProperties.Cache.Connection connection = rabbitProperties.getCache().getConnection();


   // 配置缓存模式，默认为CHANNEL， 也可以配置为 CONNECTION
   map.from(connection::getMode).whenNonNull().to(connectionFactory::setCacheMode);


   // 配置缓存connection的大小，默认是1
   map.from(connection::getSize).whenNonNull().to(connectionFactory::setConnectionCacheSize);
}

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

关于RabbitProperties 配置类, 该配置以 spring.rabbitmq 开头

package org.springframework.boot.autoconfigure.amqp;

@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitProperties {


     // some configs
}
1
2
3
4
5
6
7
8