SpringBoot整合mongoDB并自定义连接池实现多数据源配置教程_Java

springboot整合mongodb并自定义连接池实现多数据源配置

要想在同一个springboot项目中使用多个数据源，最主要是每个数据源都有自己的mongotemplate和mongodbfactory。mongotemplate和mongodbfactory是负责对数据源进行交互的并管理链接的。

spring提供了一个注解@enablemongorepositories 用来注释在某些路径下的mongorepositor实现类使用哪个mongotemplate实例。当然如果我们是直接使用mongotemplate操作，那么只需要使用于数据库对应的mongotemplate即可。

代码结果如下：

首先实现两个config，实现对mongotemplate和mongodbfactory的配置

mongotemplate1和mongodbfactory1

并使用@enablemongorepositories指定在“com.zhong.springdemo.mangodbdome.mongodb1”目录下的mongorepositor使用这些配置。

@configuration
//指定com.zhong.springdemo.mangodbdome.mongodb1路径下的mongorepository使用 容器中的  mongotemplate1实例
@enablemongorepositories(mongotemplateref = "mongotemplate1",basepackages = {"com.zhong.springdemo.mangodbdome.mongodb1"})
public class mongodbconfigure {

    @autowired
    mongodbfactoryproperties mongodbfactoryproperties;

    /**
     * 自定义 mongotemplate 实现多数据源配置
     */

    @bean("mongotemplate1")
    public mongotemplate mongotemplate(mongodbfactory mongodbfactory1, mongomappingcontext context){
        mappingmongoconverter mappingmongoconverter = mappingmongoconverter(mongodbfactory1, context);
        mongotemplate mongotemplate = new mongotemplate(mongodbfactory1, mappingmongoconverter);
        return mongotemplate;
    }


    /**
     * 自定义mongo连接池
     * @param properties 私有配置
     * @return
     */
    @bean("mongodbfactory1")
    public mongodbfactory mongodbfactory(mongodbproperties properties) {
        //创建客户端参数
        mongoclientoptions options = mongoclientoptions(properties);

        //解析地址
        list<serveraddress> serveraddresses = new arraylist<>();
        for (string address : properties.getaddress().split(",")) {
            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.get(0), mongocredential, options);
        //集群模式
        if (serveraddresses.size() > 1) {
            mongoclient = new mongoclient(serveraddresses, mongocredential, null);
        }
        /** 创建非认证客户端*/
        //mongoclient mongoclient = new mongoclient(serveraddresses, mongoclientoptions);
        return new simplemongodbfactory(mongoclient, properties.getdatabase());
    }

    /**
     * mongo客户端参数配置
     * @return
     */
    private mongoclientoptions mongoclientoptions(mongodbproperties properties) {
        mongodbfactoryproperties factoryproperties = this.mongodbfactoryproperties;
        return mongoclientoptions.builder()
                .connecttimeout(factoryproperties.getconnectiontimeoutms())
                .sockettimeout(factoryproperties.getreadtimeoutms()).applicationname(factoryproperties.getapplicationname())
                .heartbeatconnecttimeout(factoryproperties.getheartbeatconnectiontimeoutms())
                .heartbeatsockettimeout(factoryproperties.getheartbeatreadtimeoutms())
                .heartbeatfrequency(factoryproperties.getheartbeatfrequencyms())
                .minheartbeatfrequency(factoryproperties.getminheartbeatfrequencyms())
                .maxconnectionidletime(factoryproperties.getconnectionmaxidletimems())
                .maxconnectionlifetime(factoryproperties.getconnectionmaxlifetimems())
                .maxwaittime(factoryproperties.getpoolmaxwaittimems())
                .connectionsperhost(factoryproperties.getconnectionsperhost())
                .threadsallowedtoblockforconnectionmultiplier(
                        factoryproperties.getthreadsallowedtoblockforconnectionmultiplier())
                .minconnectionsperhost(factoryproperties.getminconnectionsperhost()).build();
    }

    /**
     * monogo 转换器
     * @return
     */
    private mappingmongoconverter mappingmongoconverter(mongodbfactory mongodbfactory1, mongomappingcontext context) {
        dbrefresolver dbrefresolver = new defaultdbrefresolver(mongodbfactory1);
        mappingmongoconverter mappingconverter = new mappingmongoconverter(dbrefresolver, context);
        //此处是去除插入数据库的 _class 字段
        mappingconverter.settypemapper(new defaultmongotypemapper(null));
        return mappingconverter;
    }
}

mongotemplate2和mongodbfactory2

并使用@enablemongorepositories指定在“com.zhong.springdemo.mangodbdome.mongodb2”目录下的mongorepositor使用这些配置。

@configuration
//指定com.zhong.springdemo.mangodbdome.mongodb2路径下的mongorepository使用 容器中的  mongotemplate2实例
@enablemongorepositories(mongotemplateref = "mongotemplate2",basepackages = {"com.zhong.springdemo.mangodbdome.mongodb2"})
public class mongodbconfigure2 {

    @autowired
    mongodbfactoryproperties mongodbfactoryproperties;

    /**
     * 自定义 mongotemplate 实现多数据源配置
     */

    @bean("mongotemplate2")
    public mongotemplate mongotemplate(mongodbfactory mongodbfactory2, mongomappingcontext context){
        mappingmongoconverter mappingmongoconverter = mappingmongoconverter(mongodbfactory2, context);
        mongotemplate mongotemplate = new mongotemplate(mongodbfactory2, mappingmongoconverter);
        return mongotemplate;
    }

    /**
     * 自定义mongo连接池
     * @param properties 私有配置
     * @return
     */
    @bean("mongodbfactory2")
    public mongodbfactory mongodbfactory2(mongodbproperties2 properties) {
        //创建客户端参数
        mongoclientoptions options = mongoclientoptions(properties);

        //解析地址
        list<serveraddress> serveraddresses = new arraylist<>();
        for (string address : properties.getaddress().split(",")) {
            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.get(0), mongocredential, options);
        //集群模式
        if (serveraddresses.size() > 1) {
            mongoclient = new mongoclient(serveraddresses, mongocredential, null);
        }
        /** 创建非认证客户端*/
        //mongoclient mongoclient = new mongoclient(serveraddresses, mongoclientoptions);
        return new simplemongodbfactory(mongoclient, properties.getdatabase());
    }

    /**
     * mongo客户端参数配置
     * @return
     */
    private mongoclientoptions mongoclientoptions(mongodbproperties2 properties) {
        mongodbfactoryproperties factoryproperties = this.mongodbfactoryproperties;
        return mongoclientoptions.builder()
                .connecttimeout(factoryproperties.getconnectiontimeoutms())
                .sockettimeout(factoryproperties.getreadtimeoutms()).applicationname(factoryproperties.getapplicationname())
                .heartbeatconnecttimeout(factoryproperties.getheartbeatconnectiontimeoutms())
                .heartbeatsockettimeout(factoryproperties.getheartbeatreadtimeoutms())
                .heartbeatfrequency(factoryproperties.getheartbeatfrequencyms())
                .minheartbeatfrequency(factoryproperties.getminheartbeatfrequencyms())
                .maxconnectionidletime(factoryproperties.getconnectionmaxidletimems())
                .maxconnectionlifetime(factoryproperties.getconnectionmaxlifetimems())
                .maxwaittime(factoryproperties.getpoolmaxwaittimems())
                .connectionsperhost(factoryproperties.getconnectionsperhost())
                .threadsallowedtoblockforconnectionmultiplier(
                        factoryproperties.getthreadsallowedtoblockforconnectionmultiplier())
                .minconnectionsperhost(factoryproperties.getminconnectionsperhost()).build();
    }

    /**
     * monogo 转换器
     * @return
     */
    private mappingmongoconverter mappingmongoconverter(mongodbfactory factory, mongomappingcontext context) {
        dbrefresolver dbrefresolver = new defaultdbrefresolver(factory);
        mappingmongoconverter mappingconverter = new mappingmongoconverter(dbrefresolver, context);
        //此处是去除插入数据库的 _class 字段
        mappingconverter.settypemapper(new defaultmongotypemapper(null));
        return mappingconverter;
    }
}

repository实现

实现mongdb1下的repository---userinfotestrepository，userinfotestrepository使用的是mongotemplate2和mongodbfactory2

@repository
public interface userinfotestrepository extends mongorepository<userinfoentity, string> {
    list<userinfoentity> findbyusernamelike(string username);
    list<userinfoentity> findbyusername(string username);
}

实现mongdb2下的repository---userinfrepository，userinfrepository使用的是mongotemplate1和mongodbfactory1

@repository
public interface userinforepository extends mongorepository<userinfoentity, string> {
    list<userinfoentity> findbyusernamelike(string username);
    list<userinfoentity> findbyusername(string username);
}

实现service

使用repository实现的访问的service

@service
public class userinfoserviceimpl implements userinfoservice {
    @autowired
    private userinforepository userinforepository;

    @autowired
    private userinfotestrepository userinfotestrepository;

    @override
    public list<userinfoentity> findbyusername(string username){
        return userinforepository.findbyusername(username);
    }

    @override
    public int savetestuser(list<userinfodto> userinfodtos) {
        list<userinfoentity> userinfoentities = lists.newarraylist();
        for(userinfodto userinfodto : userinfodtos){
            userinfoentity userinfoentity = new userinfoentity();
            beanutils.copyproperties(userinfodto, userinfoentity);
            userinfoentities.add(userinfoentity);
        }

        userinfotestrepository.saveall(userinfoentities);
        return userinfoentities.size();
    }

    @override
    public int saveuser(list<userinfodto> userinfodtos) {
        list<userinfoentity> userinfoentities = lists.newarraylist();
        for(userinfodto userinfodto : userinfodtos){
            userinfoentity userinfoentity = new userinfoentity();
            beanutils.copyproperties(userinfodto, userinfoentity);
            userinfoentities.add(userinfoentity);
        }

        userinforepository.saveall(userinfoentities);
        return userinfoentities.size();
    }
}

使用mongotemplate实现的访问的service

@service
public class userinfomongotemplateserviceimpl implements userinfomongotemplateservice {
    @autowired
    mongotemplate mongotemplate1;

    @autowired
    mongotemplate mongotemplate2;

    @override
    public list<userinfoentity> findbyusername(string username){
        criteria criteria = criteria.where("user_name").is(username);
        return mongotemplate1.find(getqueryfilter(criteria), userinfoentity.class);
    }

    @override
    public int savetestuser(list<userinfodto> userinfodtos) {
        list<userinfoentity> userinfoentities = lists.newarraylist();
        for(userinfodto userinfodto : userinfodtos){
            userinfoentity userinfoentity = new userinfoentity();
            beanutils.copyproperties(userinfodto, userinfoentity);
            userinfoentities.add(userinfoentity);
        }

        mongotemplate1.insert(userinfoentities, userinfoentity.class);
        return userinfoentities.size();
    }

    @override
    public int saveuser(list<userinfodto> userinfodtos) {
        list<userinfoentity> userinfoentities = lists.newarraylist();
        for(userinfodto userinfodto : userinfodtos){
            userinfoentity userinfoentity = new userinfoentity();
            beanutils.copyproperties(userinfodto, userinfoentity);
            userinfoentities.add(userinfoentity);
        }

        mongotemplate2.insert(userinfoentities, userinfoentity.class);
        return userinfoentities.size();
    }

    private query getqueryfilter(criteria criteria, string  ...parms) {
        criteria = criteria == null ? new criteria() : criteria;
        query query = new query();
        query.addcriteria(criteria);
        if(parms != null && parms.length > 0){
            field fields = query.fields();
            for(string parm : parms){
                fields.include(parm);
            }
        }
        return query;
    }
}

两个数据源信息配置properties.yaml：

zhong:
  #自定义的mongodb测试
  data:
    mongodb:
      database: zhong-mongo
      password: 123456
      address: 127.0.0.1:27017
      username: admin
      authenticationdatabase: admin
    mongodb2:
      database: test-mongo
      password: 123456
      address: 127.0.0.1:27017
      username: admin
      authenticationdatabase: admin

测试类：

@component
public class mongostarttest implements commandlinerunner {

    @autowired
    userinfoservice userinfoservice;

    @autowired
    userinfomongotemplateservice userinfomongotemplateservice;

    @override
    public void run(string... args) throws exception {
        for(int i = 0; i < 25; i++){
            userinfodto userinfodto = new userinfodto();
            userinfodto.setuserid(uuid.randomuuid().tostring().replace("-", ""));
            userinfodto.setusername("用户名" + i);
            userinfodto.setauthor("登录名" + i);
            userinfodto.setpwd("123456" + i);
            userinfodto.setcreatetime(new date());
            userinfoservice.savetestuser(lists.newarraylist(userinfodto));
            userinfoservice.saveuser(lists.newarraylist(userinfodto));
        }

        for(int i = 100; i < 125; i++){
            userinfodto userinfodto = new userinfodto();
            userinfodto.setuserid(uuid.randomuuid().tostring().replace("-", ""));
            userinfodto.setusername("用户名" + i);
            userinfodto.setauthor("登录名" + i);
            userinfodto.setpwd("123456" + i);
            userinfodto.setcreatetime(new date());
            userinfomongotemplateservice.savetestuser(lists.newarraylist(userinfodto));
            userinfomongotemplateservice.saveuser(lists.newarraylist(userinfodto));
        }
        userinfoservice.findbyusername("用户名");

        userinfomongotemplateservice.findbyusername("用户名");
    }
}

结果如图：

数据的确被插入到不同的库中了