MongoDB设置连接池操作百万级以上数据

开发环境

spring 4.3.7 + springBoot 1.5.2 + dubbo 2.6.5 + mongoDB 4.0.0

连接池配置

mongo-pool.properties

spring.data.mongodb.address=127.0.0.1:27111,127.0.0.1:27112,127.0.0.1:27113
spring.data.mongodb.replica-set=mySet
spring.data.mongodb.database=test
spring.data.mongodb.username=test
spring.data.mongodb.password=test
 
# Configure spring.data.mongodbDB Pool
spring.data.mongodb.min-connections-per-host=10
spring.data.mongodb.max-connections-per-host=100
spring.data.mongodb.threads-allowed-to-block-for-connection-multiplier=5
spring.data.mongodb.server-selection-timeout=30000
spring.data.mongodb.max-wait-time=120000
spring.data.mongodb.max-connection-idel-time=0
spring.data.mongodb.max-connection-life-time=0
spring.data.mongodb.connect-timeout=10000
spring.data.mongodb.socket-timeout=0
spring.data.mongodb.socket-keep-alive=false
spring.data.mongodb.ssl-enabled=false
spring.data.mongodb.ssl-invalid-host-name-allowed=false
spring.data.mongodb.always-use-m-beans=false
spring.data.mongodb.heartbeat-socket-timeout=20000
spring.data.mongodb.heartbeat-connect-timeout=20000
spring.data.mongodb.min-heartbeat-frequency=500
spring.data.mongodb.heartbeat-frequency=10000
spring.data.mongodb.local-threshold=15
spring.data.mongodb.authentication-database=admin

MongoSettingsProperties

import java.util.List;

import lombok.Data;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;

@Data
@Component
@PropertySource(value = "classpath:mongo-pool.properties")
@ConfigurationProperties(prefix = "spring.data.mongodb")
public class MongoSettingsProperties {
 
    private List<String> address;
    private String replicaSet;
    private String database;
    private String username;
    private String password;
    private Integer minConnectionsPerHost = 0;
    private Integer maxConnectionsPerHost = 100;
    private Integer threadsAllowedToBlockForConnectionMultiplier = 5;
    private Integer serverSelectionTimeout = 30000;
    private Integer maxWaitTime = 120000;
    private Integer maxConnectionIdleTime = 0;
    private Integer maxConnectionLifeTime = 0;
    private Integer connectTimeout = 10000;
    private Integer socketTimeout = 0;
    private Boolean socketKeepAlive = false;
    private Boolean sslEnabled = false;
    private Boolean sslInvalidHostNameAllowed = false;
    private Boolean alwaysUseMBeans = false;
    private Integer heartbeatConnectTimeout = 20000;
    private Integer heartbeatSocketTimeout = 20000;
    private Integer minHeartbeatFrequency = 500;
    private Integer heartbeatFrequency = 10000;
    private Integer localThreshold = 15;
    private String authenticationDatabase;
 
 
}

MongoConfig

import java.util.ArrayList;
import java.util.List;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;

@Configuration
public class MongoConfig {
    // 覆盖默认的MongoDbFactory
    @Bean
    MongoDbFactory mongoDbFactory(MongoSettingsProperties mongoSettingsProperties) {
        // 客户端配置（连接数、副本集群验证）
        MongoClientOptions.Builder builder = new MongoClientOptions.Builder();
        builder.connectionsPerHost(mongoSettingsProperties.getMaxConnectionsPerHost());
        builder.minConnectionsPerHost(mongoSettingsProperties.getMinConnectionsPerHost());
        if (mongoSettingsProperties.getReplicaSet() != null) {
            builder.requiredReplicaSetName(mongoSettingsProperties.getReplicaSet());
        }
        builder.threadsAllowedToBlockForConnectionMultiplier(
                mongoSettingsProperties.getThreadsAllowedToBlockForConnectionMultiplier());
        builder.serverSelectionTimeout(mongoSettingsProperties.getServerSelectionTimeout());
        builder.maxWaitTime(mongoSettingsProperties.getMaxWaitTime());
        builder.maxConnectionIdleTime(mongoSettingsProperties.getMaxConnectionIdleTime());
        builder.maxConnectionLifeTime(mongoSettingsProperties.getMaxConnectionLifeTime());
        builder.connectTimeout(mongoSettingsProperties.getConnectTimeout());
        builder.socketTimeout(mongoSettingsProperties.getSocketTimeout());
        builder.sslEnabled(mongoSettingsProperties.getSslEnabled());
        builder.sslInvalidHostNameAllowed(mongoSettingsProperties.getSslInvalidHostNameAllowed());
        builder.alwaysUseMBeans(mongoSettingsProperties.getAlwaysUseMBeans());
        builder.heartbeatFrequency(mongoSettingsProperties.getHeartbeatFrequency());
        builder.minHeartbeatFrequency(mongoSettingsProperties.getMinHeartbeatFrequency());
        builder.heartbeatConnectTimeout(mongoSettingsProperties.getHeartbeatConnectTimeout());
        builder.heartbeatSocketTimeout(mongoSettingsProperties.getHeartbeatSocketTimeout());
        builder.localThreshold(mongoSettingsProperties.getLocalThreshold());
        
        MongoClientOptions mongoClientOptions = builder.build();

        // MongoDB地址列表
        List<ServerAddress> serverAddresses = new ArrayList<>();
        for (String address : mongoSettingsProperties.getAddress()) {
            String[] hostAndPort = address.split(":");
            String host = hostAndPort[0];
            Integer port = Integer.parseInt(hostAndPort[1]);
            ServerAddress serverAddress = new ServerAddress(host, port);
            serverAddresses.add(serverAddress);
        }
        System.out.println("serverAddresses:" + serverAddresses.toString());

        // 连接认证
        List<MongoCredential> mongoCredentialList = new ArrayList<>();
        if (mongoSettingsProperties.getUsername() != null) {
            mongoCredentialList.add(MongoCredential.createScramSha1Credential(
                            mongoSettingsProperties.getUsername(),
                            mongoSettingsProperties.getAuthenticationDatabase() != null ? mongoSettingsProperties
                                    .getAuthenticationDatabase() : mongoSettingsProperties.getDatabase(),
                            mongoSettingsProperties.getPassword().toCharArray()));
        }
        System.out.println("mongoCredentialList:" + mongoCredentialList.toString());

        // 创建客户端和Factory
        MongoClient mongoClient = new MongoClient(serverAddresses,mongoCredentialList, mongoClientOptions);
        MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(mongoClient,mongoSettingsProperties.getDatabase());

        return mongoDbFactory;
    }
}

查询百万数据

第一种：

        DBObject query = new BasicDBObject(); //setup the query criteria 设置查询条件
        query.put("method", method);
        query.put("ctime", (new BasicDBObject("$gte", bTime)).append("$lt", eTime));
 
        logger.debug("query: {}", query);
 
        DBObject fields = new BasicDBObject(); //only get the needed fields. 设置需要获取哪些域
        fields.put("_id", 0);
        fields.put("uId", 1);
        fields.put("ctime", 1);
 
        DBCursor dbCursor = mongoTemplate.getCollection("collectionName").find(query, fields);
 
        while (dbCursor.hasNext()){
            DBObject object = dbCursor.next();
            logger.debug("object: {}", object);
            //do something.
        }

第二种：

 Query query = new Query();
 query.addCriteria(Criteria.where("_id").is(orderId));

mongoTemplate.find(query, this.getEntityClass(), collectionName);

对于数据量巨大的情况下，第二种方法使用mongoTemplate.find()方法返回一个列表，如果不分页的话恐怕比较麻烦。第一种方法，使用游标逐个获取数据，同时可以指定只获取哪些域，而不是全部获取回来。

记录：

插入40万 耗时大约10秒

查询70万总数 耗时3毫秒