SpringBoot 整合mongoDB并自定义连接池
为了自定义连接池，我们在配置类中主要与MongoClientOptions、MongoCredential、MongoClient、MongoDbFactory打交道。最终的目的就是配置好一个MongoDbFactory的bean交由Spring管理。
Maven 依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-mongodb</artifactId>
</dependency>
配置文件
mongodb:
database: bfa_mongo
username: "xxx"
password: "xxxxx"
address: "host:port"
authenticationDatabase: [设置你的认证数据库，如果有的话]
# 连接池配置
clientName: ${spring.application.name} # 客户端的标识，用于定位请求来源等
connectionTimeoutMs: 10000 # TCP连接超时，毫秒
readTimeoutMs: 15000 # TCP读取超时，毫秒
poolMaxWaitTimeMs: 3000 #当连接池无可用连接时客户端阻塞等待的时长，单位毫秒
connectionMaxIdleTimeMs: 60000 #TCP连接闲置时间，单位毫秒
connectionMaxLifeTimeMs: 120000 #TCP连接最多可以使用多久，单位毫秒
heartbeatFrequencyMs: 20000 #心跳检测发送频率，单位毫秒
minHeartbeatFrequencyMs: 8000 #最小的心跳检测发送频率，单位毫秒
heartbeatConnectionTimeoutMs: 10000 #心跳检测TCP连接超时，单位毫秒
heartbeatReadTimeoutMs: 15000 #心跳检测TCP连接读取超时，单位毫秒
connectionsPerHost: 20 # 每个host的TCP连接数
minConnectionsPerHost: 5 #每个host的最小TCP连接数
#计算允许多少个线程阻塞等待可用TCP连接时的乘数，算法：threadsAllowedToBlockForConnectionMultiplier*connectionsPerHost，当前配置允许10*20个线程阻塞
threadsAllowedToBlockForConnectionMultiplier: 10
MongoConfig配置类
配置类中使用了lombok，如果你没有用lombok依赖和IDE插件，你要重写getter、Setter方法：
代码稍长，可以复制在IDEA中查看：
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
import org.springframework.data.mongodb.core.convert.DbRefResolver;
import org.springframework.data.mongodb.core.convert.DefaultDbRefResolver;
import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
import org.springframework.data.mongodb.core.convert.MongoCustomConversions;
import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
import org.springframework.validation.annotation.Validated;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;
import javax.validation.constraints.Size;
import java.util.ArrayList;
import java.util.List;
@Slf4j
@Configuration
public class MongoConfig {
/**
* monogo 转换器
* @return
*/
@Bean
public MappingMongoConverter mappingMongoConverter(MongoDbFactory factory,
MongoMappingContext context, BeanFactory beanFactory, MongoCustomConversions conversions) {
DbRefResolver dbRefResolver = new DefaultDbRefResolver(factory);
MappingMongoConverter mappingConverter = new MappingMongoConverter(dbRefResolver, context);
// 删除 _class field
// mappingConverter.setTypeMapper(new DefaultMongoTypeMapper(null));
mappingConverter.setCustomConversions(conversions);
return mappingConverter;
}
/**
* 自定义mongo连接池
* @param properties
* @return
*/
@Bean
public MongoDbFactory mongoDbFactory(MongoClientOptionProperties properties) {
//创建客户端参数
MongoClientOptions options = mongoClientOptions(properties);
//创建客户端和Factory
List<ServerAddress> serverAddresses = new ArrayList<>();
for (String address : properties.getAddress()) {
String[] hostAndPort = address.split(":");
String host = hostAndPort[0];
Integer port = Integer.parseInt(hostAndPort[1]);
ServerAddress serverAddress = new ServerAddress(host, port);
serverAddresses.add(serverAddress);
}
//创建认证客户端
MongoCredential mongoCredential = MongoCredential.createScramSha1Credential(properties.getUsername(),
properties.getAuthenticationDatabase() != null ? properties.getAuthenticationDatabase() : properties.getDatabase(),
properties.getPassword().toCharArray());
MongoClient mongoClient = new MongoClient(serverAddresses, mongoCredential, options);
/** ps: 创建非认证客户端*/
//MongoClient mongoClient = new MongoClient(serverAddresses, mongoClientOptions);
return new SimpleMongoDbFactory(mongoClient, properties.getDatabase());
}
/**
* mongo客户端参数配置
* @return
*/
@Bean
public MongoClientOptions mongoClientOptions(MongoClientOptionProperties properties) {
return MongoClientOptions.builder()
.connectTimeout(properties.getConnectionTimeoutMs())
.socketTimeout(properties.getReadTimeoutMs()).applicationName(properties.getClientName())
.heartbeatConnectTimeout(properties.getHeartbeatConnectionTimeoutMs())
.heartbeatSocketTimeout(properties.getHeartbeatReadTimeoutMs())
.heartbeatFrequency(properties.getHeartbeatFrequencyMs())
.minHeartbeatFrequency(properties.getMinHeartbeatFrequencyMs())
.maxConnectionIdleTime(properties.getConnectionMaxIdleTimeMs())
.maxConnectionLifeTime(properties.getConnectionMaxLifeTimeMs())
.maxWaitTime(properties.getPoolMaxWaitTimeMs())
.connectionsPerHost(properties.getConnectionsPerHost())
.threadsAllowedToBlockForConnectionMultiplier(
properties.getThreadsAllowedToBlockForConnectionMultiplier())
.minConnectionsPerHost(properties.getMinConnectionsPerHost()).build();
}
@Primary
@Bean
@ConfigurationProperties(prefix = "mongodb")
public MongoClientOptionProperties mongoClientOptionProperties() {
return new MongoClientOptionProperties();
}
@Getter
@Setter
@Validated
@Configuration
public static class MongoClientOptionProperties {
/** 基础连接参数 */
private String database;
private String username;
private String password;
@NotNull
private List<String> address;
private String authenticationDatabase;
/** 客户端连接池参数 */
@NotNull
@Size(min = 1)
private String clientName;
/** socket连接超时时间 */
@Min(value = 1)
private int connectionTimeoutMs;
/** socket读取超时时间 */
@Min(value = 1)
private int readTimeoutMs;
/** 连接池获取链接等待时间 */
@Min(value = 1)
private int poolMaxWaitTimeMs;
/** 连接闲置时间 */
@Min(value = 1)
private int connectionMaxIdleTimeMs;
/** 连接最多可以使用多久 */
@Min(value = 1)
private int connectionMaxLifeTimeMs;
/** 心跳检测发送频率 */
@Min(value = 2000)
private int heartbeatFrequencyMs;
/** 最小的心跳检测发送频率 */
@Min(value = 300)
private int minHeartbeatFrequencyMs;
/** 计算允许多少个线程阻塞等待时的乘数，算法：threadsAllowedToBlockForConnectionMultiplier*connectionsPerHost */
@Min(value = 1)
private int threadsAllowedToBlockForConnectionMultiplier;
/** 心跳检测连接超时时间 */
@Min(value = 200)
private int heartbeatConnectionTimeoutMs;
/** 心跳检测读取超时时间 */
@Min(value = 200)
private int heartbeatReadTimeoutMs;
/** 每个host最大连接数 */
@Min(value = 1)
private int connectionsPerHost;
/** 每个host的最小连接数 */
@Min(value = 1)
private int minConnectionsPerHost;
}
}
MappingMongoConverter可以自定义mongo转换器，主要自定义存取mongo数据时的一些操作，例如 mappingConverter.setTypeMapper(new DefaultMongoTypeMapper(null)) 方法会将mongo数据中的_class字段去掉。
最后通过 new SimpleMongoDbFactory(mongoClient, properties.getDatabase())方法配置了一个MongoDbFactory交由Spring管理，Springboot会拿这个MongoDbFactory工厂bean来new一个MongoTemplate，在MongoDbFactoryDependentConfiguration类下可以看到SpringBoot帮你做得事：
/**
* Configuration for Mongo-related beans that depend on a {@link MongoDbFactory}.
*
* @author Andy Wilkinson
*/
@Configuration
@ConditionalOnBean(MongoDbFactory.class)
class MongoDbFactoryDependentConfiguration {
private final MongoProperties properties;
MongoDbFactoryDependentConfiguration(MongoProperties properties) {
this.properties = properties;
}
//SpringBoot创建MongoTemplate实例
@Bean
@ConditionalOnMissingBean
public MongoTemplate mongoTemplate(MongoDbFactory mongoDbFactory, MongoConverter converter) {
return new MongoTemplate(mongoDbFactory, converter);
}
@Bean
@ConditionalOnMissingBean(MongoConverter.class)
public MappingMongoConverter mappingMongoConverter(MongoDbFactory factory, MongoMappingContext context,
MongoCustomConversions conversions) {
DbRefResolver dbRefResolver = new DefaultDbRefResolver(factory);
MappingMongoConverter mappingConverter = new MappingMongoConverter(dbRefResolver, context);
mappingConverter.setCustomConversions(conversions);
return mappingConverter;
}
//...
}
SpringBoot利用我们配置好的MongoDbFactory在配置类中生成一个MongoTemplate，之后我们就可以在项目代码中直接@Autowired了。因为用于生成MongoTemplate的MongoDbFactory是我们自己在MongoConfig配置类中生成的，所以我们自定义的连接池参数也就生效了。