SpringBoot中多数据库下使用MongoDB连接池
背景
对于PaaS平台注册租户，每个租户的数据库隔离，也就是每个租户拥有自己单独的数据库。每个租户下可以创建多个应用，每个应用也是单独的数据库。在实际业务处理时，会根据租户和应用，动态的链接数据库。由于是动态的创建数据库，如果连接池不能复用，就意味着系统会由于日益增多的租户，而导致数据库数量持续增长，最终由于不能复用连接池，导致DB连接数过多，系统会逐渐变慢，最终导致CPU负荷过高而导致系统崩溃。
MongoDB连接池配置
#============== mongoDB连接配置 ===================
#格式: mongodb://账号:密码@ip:端口/数据库?认证数据库。例如：mongodb://root:admin@localhost/mongotest?authSource=admin
spring.data.mongodb.uri=mongodb://ip:port/your_db?waitQueueMultiple=100&connectTimeoutMS=100&minPoolSize=10&maxPoolSize=10&slaveOk=true&write=1&readPreference=secondaryPreferred
#以逗号分隔的主机：端口对列表
spring.data.mongodb.address=ip:port
#默认链接的数据库（需要与URI对应）
spring.data.mongodb.database=your_db
# Configure spring.data.mongodbDB Pool
# 每个host的TCP连接数
spring.data.mongodb.connectionsPerHost=10
# 每个host的最小TCP连接数
spring.data.mongodb.minConnectionsPerHost=10
# 计算允许多少个线程阻塞等待可用TCP连接时的乘数（ 连接池里的连接数 ）
# 算法：threadsAllowedToBlockForConnectionMultiplier*connectionsPerHost
spring.data.mongodb.threadsAllowedToBlockForConnectionMultiplier=1
spring.data.mongodb.serverSelectionTimeout=5000
# 当连接池无可用连接时客户端阻塞等待的时长，单位毫秒
spring.data.mongodb.maxWaitTime=15000
# TCP连接闲置时间,单位毫秒
spring.data.mongodb.maxConnectionIdleTime=3600000
# TCP连接最多可以使用多久,单位毫秒
spring.data.mongodb.maxConnectionLifeTime=3600000
# 连接超时时间,必须大于0,单位毫秒
spring.data.mongodb.connectTimeout=5000
# socket超时时间,单位毫秒
spring.data.mongodb.socketTimeout=5000
spring.data.mongodb.socketKeepAlive=true
spring.data.mongodb.sslEnabled=false
spring.data.mongodb.sslInvalidHostNameAllowed=false
spring.data.mongodb.alwaysUseMBeans=false
# 设置心跳频率。 这是驱动程序将尝试确定群集中每个服务器的当前状态的频率。 默认值为10,000毫秒
spring.data.mongodb.heartbeatFrequency=10000
# 设置最小心跳频率。 如果驱动程序必须经常重新检查服务器的可用性，它将至少在上一次检查后等待很长时间，以避免浪费精力。 默认值为500毫秒。
spring.data.mongodb.minHeartbeatFrequency=500
# 设置用于集群心跳的连接的连接超时,单位毫秒
spring.data.mongodb.heartbeatConnectTimeout=20000
# 设置用于集群心跳的连接的套接字超时,单位毫秒
spring.data.mongodb.heartbeatSocketTimeout=20000
# 本地阈值
spring.data.mongodb.localThreshold=15SpringBoot中操作MongoDB操作实例产生的过程
从上的图中我们可以看出，产生MongoClient实例的方式有两种：一种是通过MongoClientURI；另外一种是直接实例化MongoClient。通过MongoClientURI内部会实例化一个ConnectionString
public ConnectionString(String connectionString) {
this.connectionString = connectionString;
boolean isMongoDBProtocol = connectionString.startsWith("mongodb://");
this.isSrvProtocol = connectionString.startsWith("mongodb+srv://");
if (!isMongoDBProtocol && !this.isSrvProtocol) {
throw new IllegalArgumentException(String.format("The connection string is invalid. Connection strings must start with either '%s' or '%s", "mongodb://", "mongodb+srv://"));
} else {
String unprocessedConnectionString;
if (isMongoDBProtocol) {
unprocessedConnectionString = connectionString.substring("mongodb://".length());
} else {
unprocessedConnectionString = connectionString.substring("mongodb+srv://".length());
}
int idx = unprocessedConnectionString.indexOf("/");
String userAndHostInformation;
if (idx == -1) {
if (unprocessedConnectionString.contains("?")) {
throw new IllegalArgumentException("The connection string contains options without trailing slash");
}
userAndHostInformation = unprocessedConnectionString;
unprocessedConnectionString = "";
} else {
userAndHostInformation = unprocessedConnectionString.substring(0, idx);
unprocessedConnectionString = unprocessedConnectionString.substring(idx + 1);
}
String userName = null;
char[] password = null;
idx = userAndHostInformation.lastIndexOf("@");
String hostIdentifier;
if (idx > 0) {
String userInfo = userAndHostInformation.substring(0, idx).replace("+", "%2B");
hostIdentifier = userAndHostInformation.substring(idx + 1);
int colonCount = this.countOccurrences(userInfo, ":");
if (userInfo.contains("@") || colonCount > 1) {
throw new IllegalArgumentException("The connection string contains invalid user information. If the username or password contains a colon (:) or an at-sign (@) then it must be urlencoded");
}
if (colonCount == 0) {
userName = this.urldecode(userInfo);
} else {
idx = userInfo.indexOf(":");
userName = this.urldecode(userInfo.substring(0, idx));
password = this.urldecode(userInfo.substring(idx + 1), true).toCharArray();
}
} else {
hostIdentifier = userAndHostInformation;
}
List<String> unresolvedHosts = Collections.unmodifiableList(this.parseHosts(Arrays.asList(hostIdentifier.split(","))));
if (this.isSrvProtocol) {
if (unresolvedHosts.size() > 1) {
throw new IllegalArgumentException("Only one host allowed when using mongodb+srv protocol");
}
if (((String)unresolvedHosts.get(0)).contains(":")) {
throw new IllegalArgumentException("Host for when using mongodb+srv protocol can not contain a port");
}
}
this.hosts = unresolvedHosts;
idx = unprocessedConnectionString.indexOf("?");
String nsPart;
if (idx == -1) {
nsPart = unprocessedConnectionString;
unprocessedConnectionString = "";
} else {
nsPart = unprocessedConnectionString.substring(0, idx);
unprocessedConnectionString = unprocessedConnectionString.substring(idx + 1);
}
if (nsPart.length() > 0) {
nsPart = this.urldecode(nsPart);
idx = nsPart.indexOf(".");
if (idx < 0) {
this.database = nsPart;
this.collection = null;
} else {
this.database = nsPart.substring(0, idx);
this.collection = nsPart.substring(idx + 1);
}
MongoNamespace.checkDatabaseNameValidity(this.database);
} else {
this.database = null;
this.collection = null;
}
String txtRecordsQueryParameters = this.isSrvProtocol ? (new DefaultDnsResolver()).resolveAdditionalQueryParametersFromTxtRecords((String)unresolvedHosts.get(0)) : "";
Map<String, List<String>> connectionStringOptionsMap = this.parseOptions(unprocessedConnectionString);
Map<String, List<String>> txtRecordsOptionsMap = this.parseOptions(txtRecordsQueryParameters);
if (!ALLOWED_OPTIONS_IN_TXT_RECORD.containsAll(txtRecordsOptionsMap.keySet())) {
throw new MongoConfigurationException(String.format("A TXT record is only permitted to contain the keys %s, but the TXT record for '%s' contains the keys %s", ALLOWED_OPTIONS_IN_TXT_RECORD, unresolvedHosts.get(0), txtRecordsOptionsMap.keySet()));
} else {
Map<String, List<String>> combinedOptionsMaps = this.combineOptionsMaps(txtRecordsOptionsMap, connectionStringOptionsMap);
if (this.isSrvProtocol && !combinedOptionsMaps.containsKey("ssl")) {
combinedOptionsMaps.put("ssl", Collections.singletonList("true"));
}
this.translateOptions(combinedOptionsMaps);
this.credential = this.createCredentials(combinedOptionsMaps, userName, password);
this.warnOnUnsupportedOptions(combinedOptionsMaps);
}
}
}从构造函数中可以看到，ConnectionString在实例化时，也是对传入进来的URI进行了一些的校验和处理，通过URI获取用户名、密码等等信息。我们再看下SimpleMongoDbFactory
public class SimpleMongoDbFactory extends MongoDbFactorySupport<MongoClient> implements DisposableBean {
public SimpleMongoDbFactory(MongoClientURI uri) {
this(new MongoClient(uri), uri.getDatabase(), true);
}
public SimpleMongoDbFactory(MongoClient mongoClient, String databaseName) {
this(mongoClient, databaseName, false);
}
private SimpleMongoDbFactory(MongoClient mongoClient, String databaseName, boolean mongoInstanceCreated) {
super(mongoClient, databaseName, mongoInstanceCreated, new MongoExceptionTranslator());
}
public DB getLegacyDb() {
return ((MongoClient)this.getMongoClient()).getDB(this.getDefaultDatabaseName());
}
public ClientSession getSession(ClientSessionOptions options) {
return ((MongoClient)this.getMongoClient()).startSession(options);
}
protected void closeClient() {
((MongoClient)this.getMongoClient()).close();
}
protected MongoDatabase doGetMongoDatabase(String dbName) {
return ((MongoClient)this.getMongoClient()).getDatabase(dbName);
}
}从第一个构造函数里面就可以得出一个信息，通过URI最终仍是构造出一个MongoClient实例，所以通过MongoClientURI方式创建的MongoTemplate，虽然有连接池的作用，但是由于MongoClient实例不同，数据库连接池并没有共享！
按照业务背景需要，每个数据库都应该有自己专属的MongoTemplate，否则数据库操作就乱套了。但是从底层往上创建MongoTemplate过程中，若是能够复用MongoClient，那么就可以做到不同的DB实例共享同一个数据库连接池。所以我们应该让MongoClient共享，这样数据库连接池才能都是从一处控制。
自定义一个动态获取指定DB操作实例的公共类
public class SelfMongoTemplate {
private static final Logger LOG = LoggerFactory.getLogger(SelfMongoTemplate.class);
private static MongoClient mongoClient = null;
//获取连接池配置对象
public static MongoClientOptions.Builder getBuiler() {
return BeanUtil.getBean("globalMongoPoolConfig", MongoClientOptions.Builder.class);
}
//带有连接池的Mongo客户端
public static MongoClient getMongoClient() {
// double check 确保单例
if (null == mongoClient) {
synchronized (SelfMongoTemplate.class) {
if (null == mongoClient) {
// MongoDB地址列表
String[] addressList = Sysconfig.mongoAddress.split(",");
List<ServerAddress> serverAddresses = new ArrayList<ServerAddress>();
for (String address : addressList) {
String[] hostAndPort = address.split(":");
ServerAddress serverAddress = new ServerAddress(hostAndPort[0], Integer.parseInt(hostAndPort[1]));
serverAddresses.add(serverAddress);
}
// 创建非认证客户端
mongoClient = new MongoClient(serverAddresses, getBuiler().build());
}
}
}
return mongoClient;
}
public static MongoDbFactory getDbFactory(String dbName) {
String beanName = FuncUtil.joinStr(dbName, "_mongoDbFactory");
if (BeanUtil.hasBean(beanName)) {
return BeanUtil.getBean(beanName, MongoDbFactory.class);
}
BeanUtil.registerBean(beanName, SimpleMongoDbFactory.class, getMongoClient(), dbName);
return BeanUtil.getBean(beanName, MongoDbFactory.class);
}
/**
* 获取数据库实例下面所有的数据库名称
* @return
*/
public static ArrayList<String> listDatabaseNames() {
ArrayList<String> dbs = new ArrayList<>();
MongoIterable<String> list = getMongoClient().listDatabaseNames();
for (String dbName : list) {
dbs.add(dbName);
}
return dbs;
}
public static MongoTemplate getMongoDBTemplate(String dbName) {
String beanName = FuncUtil.joinStr(dbName, "_mongoTemplate");
if (BeanUtil.hasBean(beanName)) {
return BeanUtil.getBean(beanName, MongoTemplate.class);
}
synchronized (SelfMongoTemplate.class) {
if (!BeanUtil.hasBean(beanName)) {
//动态注册一个MongoTemplate实例bean
BeanUtil.registerBean(beanName, MongoTemplate.class, getDbFactory(dbName));
}
}
return BeanUtil.getBean(beanName, MongoTemplate.class);
}
}