MongoDB Java API操作很全的整理

mongodb 是一个基于分布式文件存储的数据库。由 c++ 语言编写，一般生产上建议以共享分片的形式来部署。 但是mongodb官方也提供了其它语言的客户端操作api。如下图所示：

提供了c、c++、c#、.net、go、java、node.js、php、python、scala等各种语言的版本，如下图所示：

mongodb的操作分为同步操作和异步操作以及响应式编程操作
一、同步操作api

官方java api的路径：  我们这里以3.11的java 版本为例。各个版本的api对mongodb服务的支持情况。

使用api时，先引入maven依赖

<!-- https://mvnrepository.com/artifact/org.mongodb/mongo-java-driver -->
<dependency>
    <groupid>org.mongodb</groupid>
    <artifactid>mongo-java-driver</artifactid>
    <version>3.11.1</version>
</dependency>

　　

 1、关于mongodb client的初始化和关闭。

从官方介绍来看，一般建议client只需要一个建立一个长连接实例，然后使用时，都使用这个实例就可以，也就是可以用java的单例模式来创建连接实例。

 

//mongoclient连接
protected static mongoclient mongoclient;
 public synchronized static mongodbclient getinstance(string mongodburl) {
        if (null == mongoclient) {
            mongoclient = mongoclients.create(mongodburl);
            if(null != mongoclient){
                log.info("mongoclient init success!");
            }
            else{
                log.info("mongoclient init failed!");
            }
        }
        return mongodbclient;
    }　

　　

直接通过mongodb的host和port来创建client：　

mongoclient mongoclient = mongoclients.create("mongodb://host1:27017");

client连接到一个 replica set：

mongoclient mongoclient = mongoclients.create("mongodb://host1:27017,host2:27017,host3:27017");

mongoclient mongoclient = mongoclients.create("mongodb://host1:27017,host2:27017,host3:27017/?replicaset=myreplicaset");

　或者通过mongoclientsettings.builder() 来辅助生成连接字符串来创建client：

mongoclient mongoclient = mongoclients.create( mongoclientsettings.builder() .applytoclustersettings(builder -> builder.hosts(arrays.aslist( new serveraddress("host1", 27017), new serveraddress("host2", 27017), new serveraddress("host3", 27017)))) .build());

　　连接关闭：

    public void close() {
        if(null!=mongoclient){
            mongoclient.close();
            mongoclient=null;
        }
    }

　 2、关于mongodb 的基本操作

//创建collection

 public void createcollection(string databasename,string collectionname){
        getdatabase(databasename).createcollection(collectionname);
    }
//查询databasename

public mongodatabase getdatabase(string databasename){ return mongoclient.getdatabase(databasename); }
//查询collection

public list<string> listcollectionnames(string databasename){
 list<string> stringlist = new arraylist<string>(); 
mongoclient.getdatabase(databasename).listcollectionnames().foreach((consumer<? super string>) t->{ stringlist.add(t); });
 return stringlist; } 

public mongocollection<document> getcollectionbyname(string databasename, string collectionname){ return getdatabase(databasename).getcollection(collectionname); }

 3、关于mongodb 的查询操作

    //通过id（objectid）精确查询
	public finditerable<document>  findmongodbdocbyid(string databasename, string collectionname, string id){
        basicdbobject searchdoc = new basicdbobject().append("_id", id);
       return getcollectionbyname(databasename,collectionname).find(searchdoc);
    }
    //通过id（objectid）模糊查询
    public finditerable<document>  findmongodbdocbyidregex(string databasename, string collectionname, string id){
        basicdbobject searchdoc = new basicdbobject().append("_id", new basicdbobject("$regex",id));
        return getcollectionbyname(databasename,collectionname).find(searchdoc);
    }
	//通过开始id和结束id 查询（根据objectid范围查询）
    public finditerable<document>  findmongodbdocbyid(string databasename, string collectionname, string startid,string endid){
        basicdbobject searchdoc = new basicdbobject().append("_id", new basicdbobject("$gte", startid).append("$lte", endid));
        return getcollectionbyname(databasename,collectionname).find(searchdoc);
    }
    public finditerable<document> findmongodbdoc(string databasename, string collectionname,basicdbobject basicdbobject){
        return getcollectionbyname(databasename,collectionname).find(basicdbobject);
    }
	//限制查询返回的条数
    public finditerable<document> findmongodbdoc(string databasename, string collectionname,basicdbobject basicdbobject,integer limitnum){
        return findmongodbdoc(databasename,collectionname,basicdbobject).limit(limitnum) ;
    }
    public finditerable<document>  findmongodbdocbyid(string databasename, string collectionname, string startid,string endid,integer limitnum){
        return findmongodbdocbyid(databasename,collectionname,startid,endid).limit(limitnum);
    }

    /**
     * 降序查询（排序）
     * @param databasename
     * @param collectionname
     * @param startid
     * @param endid
     * @param sortfield  排序字段
     * @return
     */
    public finditerable<document>  findmongodbdocbyiddescsort(string databasename, string collectionname, string startid,string endid,string sortfield){
      return findmongodbdocbyid(databasename,collectionname,startid,endid).sort(new document().append(sortfield, -1));
    }
    public finditerable<document>  findmongodbdocbyiddescsort(string databasename, string collectionname, string startid,string endid,string sortfield,integer limitnum){
        return findmongodbdocbyiddescsort(databasename,collectionname,startid,endid,sortfield).limit(limitnum);
    }

    /**
     * 降序查询（排序）
     * @param databasename
     * @param collectionname
     * @param startid
     * @param endid
     * @param sortfield  排序字段
     * @return
     */
    public finditerable<document>  findmongodbdocbyidascsort(string databasename, string collectionname, string startid,string endid,string sortfield){
        return findmongodbdocbyid(databasename,collectionname,startid,endid).sort(new document().append(sortfield, 1));
    }
    public finditerable<document>  findmongodbdocbyidascsort(string databasename, string collectionname, string startid,string endid,string sortfield,integer limitnum){
        return findmongodbdocbyidascsort(databasename,collectionname,startid,endid,sortfield).limit(limitnum);
    }

　　 4、关于mongodb 的插入操作

   //插入操作，注意插入时，如果数据已经存在会报错，插入时必须数据不存在，不会自动进行覆盖
   //插入单条记录   
   public void insertdoc(string databasename, string collectionname, document document){
        getcollectionbyname(databasename,collectionname).insertone(document);
    }
	//插入多条记录
    public void insertdoc(string databasename, string collectionname,list<? extends document> listdata){
        getcollectionbyname(databasename,collectionname).insertmany(listdata);
    }

　　 5、关于mongodb 的更新操作

	//更新单条
    public void updatedoc(string databasename, string collectionname, bson var1, bson var2){
        getcollectionbyname(databasename,collectionname).updateone(var1,var2);
    }
    public void updatedoc(string databasename, string collectionname, bson var1, list<? extends bson> list){
        getcollectionbyname(databasename,collectionname).updateone(var1,list);
    }
	//批量更新
    public void updatedocs(string databasename, string collectionname, bson var1, bson var2){
        getcollectionbyname(databasename,collectionname).updatemany(var1,var2);
    }
    public void updatedocs(string databasename, string collectionname, bson var1, list<? extends bson> list){
        getcollectionbyname(databasename,collectionname).updatemany(var1,list);
    }

　 6、关于mongodb 的删除操作　

//单条删除  
  public deleteresult deletedoc(string databasename, string collectionname, bson var1){
        return getcollectionbyname(databasename,collectionname).deleteone(var1);
    }
//批量删除	
    public deleteresult deletedocs(string databasename, string collectionname,bson var1){
       return getcollectionbyname(databasename,collectionname).deletemany(var1);
    }

7、关于mongodb 的替换操作

	//存在就替换，不存在的话就插入
   public updateresult replacedoc(string databasename, string collectionname, bson var1, document var2){
        return getcollectionbyname(databasename,collectionname).replaceone(var1,var2);
    }

8、关于mongodb 的bulkwrite操作　（批量写入），对于数据很多时，效率很高

    public bulkwriteresult bulkwrite(string databasename, string collectionname, list<? extends writemodel<? extends document>> listdata){
       return getcollectionbyname(databasename,collectionname).bulkwrite(listdata);
    }

 9、关于mongodb 的分页查询

mongodb的分页查询可以有多种思路来实现。

思路一：采用类似mysql的limit start end 的这种。

获取到总的数量：

    //查询总数
    public long countdocs(string databasename, string collectionname,bson var1){
        if(null==var1){
            return getcollectionbyname(databasename,collectionname).countdocuments();
        }
       return getcollectionbyname(databasename,collectionname).countdocuments(var1);
    }

//　　分页查询，采用skip+limit的方式，在用了总数后，就可以分页了，skip的意思是前面跳过多少数据。但是这种方式在数据量大的时候效率不高，因为skip会导致全表扫描。

    public finditerable<document> findmongodbdoc(string databasename, string collectionname,basicdbobject basicdbobject,integer skip,integer limit){
        return getcollectionbyname(databasename,collectionname).find(basicdbobject).skip(skip).limit(limit);
    }

思路二：利用limit 以及排序的方式，获取分页的上一页的最后一条记录的objectid，然后使用排序+$gte操作（大于）+limit 来获取当页的数据。找到一个可以排序的字段，比如objectid或者时间字段都可以排序。这个也是mongodb官方推荐的方式，这种做饭可以避免全表扫描。

思路三：在数据量不大的时候，使用代码进行分页。比如从mongodb中查询出一个list对象后，对list对象做代码分页。

public class listutil {
    public static list getpaginglist(list list,integer start,integer length){
        start = start<0?0:start;
        //默认为10
        length = length<=0?10:length;
        integer size = list.size();
        if(start>size){
            start = size;
        }
        integer toindex = (start+length-1)>=size?size:(start+length-1);
        if(toindex<=0){
            toindex = size;
        }
        return list.sublist(start,toindex);
    }

　　

二、异步操作api　　

　mongodb异步驱动程序提供了异步api，可以利用netty或java 7的asynchronoussocketchannel实现快速、无阻塞的i/o，maven依赖

<dependencies>
<dependency>
<groupid>org.mongodb</groupid>
<artifactid>mongodb-driver-async</artifactid>
<version>3.11.1</version>
</dependency>
</dependencies>

官方地址：

异步操作必然会涉及到回调，回调时采用resultcallback<document>

singleresultcallback<document> callbackprintdocuments = new singleresultcallback<document>() {
   @override
   public void onresult(final document document, final throwable t) {
       system.out.println(document.tojson());
   }
};

singleresultcallback<void> callbackwhenfinished = new singleresultcallback<void>() {
    @override
    public void onresult(final void result, final throwable t) {
        system.out.println("operation finished!");
    }
};

　　异步insert操作

collection.insertmany(documents, new singleresultcallback<void>() {
    @override
    public void onresult(final void result, final throwable t) {
        system.out.println("documents inserted!");
    }
});

　　异步删除操作

collection.deletemany(gte("i", 100), new singleresultcallback<deleteresult>() {
    @override
    public void onresult(final deleteresult result, final throwable t) {
        system.out.println(result.getdeletedcount());
    }
});

　　异步更新操作

collection.updatemany(lt("i", 100), inc("i", 100),
    new singleresultcallback<updateresult>() {
        @override
        public void onresult(final updateresult result, final throwable t) {
            system.out.println(result.getmodifiedcount());
        }
    });

　　异步统计操作

collection.countdocuments(
  new singleresultcallback<long>() {
      @override
      public void onresult(final long count, final throwable t) {
          system.out.println(count);
      }
  });

　　

三、mongodb reactive streams 操作api

官方的mongodb reactive streams java驱动程序，为mongodb提供异步流处理和无阻塞处理。

完全实现reactive streams api，以提供与jvm生态系统中其他reactive streams的互操作，一般适合于大数据的处理，比如spark，flink，storm等。

<dependencies>
    <dependency>
        <groupid>org.mongodb</groupid>
        <artifactid>mongodb-driver-reactivestreams</artifactid>
        <version>1.12.0</version>
    </dependency>
</dependencies>

　　官方地址：

会包含如下三部分：

1. publisher：publisher 是数据的发布者。publisher 接口只有一个方法 subscribe，用于添加数据的订阅者，也就是 subscriber。
2. subscriber： 是数据的订阅者。subscriber 接口有4个方法，都是作为不同事件的处理器。在订阅者成功订阅到发布者之后，其 onsubscribe(subscription s) 方法会被调用。
3. subscription：表示的是当前的订阅关系。

api问的地址：

 

 

 

代码示例：

//建立连接
mongoclient mongoclient = mongoclients.create(mongodburl);
//获得数据库对象
mongodatabase database = client.getdatabase(databasename);
//获得集合
mongocollection collection = database.getcollection(collectionname);

//异步返回publisher
findpublisher publisher = collection.find();

//订阅实现
publisher.subscribe(new subscriber() {
    @override
    public void onsubscribe(subscription str) {
        system.out.println("start...");
        //执行请求
        str.request(integer.max_value);
    }
    @override
    public void onnext(document document) {
        //获得文档
        system.out.println("document:" + document.tojson());
    }

    @override
    public void onerror(throwable t) {
        system.out.println("error occurs.");
    }

    @override
    public void oncomplete() {
        system.out.println("finished.");
    }
});