MongoDB 是一个基于分布式文件存储的数据库。由 C++ 语言编写,一般生产上建议以共享分片的形式来部署。 但是MongoDB官方也提供了其它语言的客户端操作API。如下图所示:
提供了C、C++、C#、.net、GO、java、Node.js、PHP、python、scala等各种语言的版本,如下图所示:
MongoDB的操作分为同步操作和异步操作以及响应式编程操作
一、同步操作API
官方JAVA API的路径:https://docs.mongodb.com/ecosystem/drivers/java/ 我们这里以3.11的java 版本为例。各个版本的API对MongoDB服务的支持情况。
使用API时,先引入maven依赖
org.mongodb
mongo-java-driver
3.11.1
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 listCollectionNames(String dataBaseName){
List stringList = new ArrayList();
mongoClient.getDatabase(dataBaseName).listCollectionNames().forEach((Consumer super String>) t->{ stringList.add(t); });
return stringList; }
public MongoCollection getCollectionByName(String dataBaseName, String collectionName){ return getDatabase(dataBaseName).getCollection(collectionName); }
3、关于MongoDB 的查询操作
//通过id(objectid)精确查询
public FindIterable findMongoDbDocById(String dataBaseName, String collectionName, String id){
BasicDBObject searchDoc = new BasicDBObject().append("_id", id);
return getCollectionByName(dataBaseName,collectionName).find(searchDoc);
}
//通过id(objectid)模糊查询
public FindIterable 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 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 findMongoDbDoc(String dataBaseName, String collectionName,BasicDBObject basicDBObject){
return getCollectionByName(dataBaseName,collectionName).find(basicDBObject);
}
//限制查询返回的条数
public FindIterable findMongoDbDoc(String dataBaseName, String collectionName,BasicDBObject basicDBObject,Integer limitNum){
return findMongoDbDoc(dataBaseName,collectionName,basicDBObject).limit(limitNum) ;
}
public FindIterable findMongoDbDocById(String dataBaseName, String collectionName, String startId,String endId,Integer limitNum){
return findMongoDbDocById(dataBaseName,collectionName,startId,endId).limit(limitNum);
}
public FindIterable 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 findMongoDbDocByIdDescSort(String dataBaseName, String collectionName, String startId,String endId,String sortField,Integer limitNum){
return findMongoDbDocByIdDescSort(dataBaseName,collectionName,startId,endId,sortField).limit(limitNum);
}
public FindIterable 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 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 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依赖
org.mongodb
mongodb-driver-async
3.11.1
官方地址:http://mongodb.github.io/mongo-java-driver/3.11/driver-async/getting-started/installation/
异步操作必然会涉及到回调,回调时采用ResultCallback
SingleResultCallback callbackPrintDocuments = new SingleResultCallback() {
@Override
public void onResult(final Document document, final Throwable t) {
System.out.println(document.toJson());
}
};
SingleResultCallback callbackWhenFinished = new SingleResultCallback() {
@Override
public void onResult(final Void result, final Throwable t) {
System.out.println("Operation Finished!");
}
};
异步insert操作
collection.insertMany(documents, new SingleResultCallback() {
@Override
public void onResult(final Void result, final Throwable t) {
System.out.println("Documents inserted!");
}
});
异步删除操作
collection.deleteMany(gte("i", 100), new SingleResultCallback() {
@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() {
@Override
public void onResult(final UpdateResult result, final Throwable t) {
System.out.println(result.getModifiedCount());
}
});
异步统计操作
collection.countDocuments(
new SingleResultCallback() {
@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等。
org.mongodb
mongodb-driver-reactivestreams
1.12.0
官方地址:http://mongodb.github.io/mongo-java-driver-reactivestreams/
会包含如下三部分:
- Publisher:Publisher 是数据的发布者。Publisher 接口只有一个方法 subscribe,用于添加数据的订阅者,也就是 Subscriber。
- Subscriber: 是数据的订阅者。Subscriber 接口有4个方法,都是作为不同事件的处理器。在订阅者成功订阅到发布者之后,其 onSubscribe(Subscription s) 方法会被调用。
- Subscription:表示的是当前的订阅关系。
API问的地址:http://mongodb.github.io/mongo-java-driver-reactivestreams/1.12/javadoc/
代码示例:
//建立连接
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.");
}
});